// A.EXE - MQTT listener. Subscribes to a topic and, one second after each
// message arrives, prints an "echo" line.
// Usage: A <broker> <port> <topic>   (or MQTT_HOST/MQTT_PORT/MQTT_TOPIC in the environment)
#include "mqtt.h"

#define MAX_PENDING 8
struct Pending { clockTicks_t due; char text[96]; uint16_t used; };
static Pending pending[MAX_PENDING];

int main( int argc, char *argv[] ) {
  uint8_t pkt[200];
  char who[16], now[16];

  const char *host; uint16_t port; const char *topic;
  if ( parseTarget( argc, argv, &host, &port, &topic ) ) {
    printf( "Usage: A <broker> <port> <topic>\n   or set MQTT_HOST, MQTT_PORT and MQTT_TOPIC\n" );
    return 1;
  }

  printf( "A.EXE - MQTT listener (1 s delayed echo)\n" );
  if ( netInit( ) ) { printf( "TCP/IP init failed (is the packet driver loaded?)\n" ); return 2; }

  TcpSocket *s = openBroker( host, port );
  if ( !s ) netShutdown( 3 );

  makeClientId( who, "dosA" );
  int n = buildConnect( pkt, who );
  s->send( pkt, n );
  if ( waitFor( s, 2, 10000 ) ) { printf( "MQTT: CONNECT refused or timed out\n" ); netShutdown( 4 ); }
  printf( "MQTT: connected as %s\n", who );

  n = buildSubscribe( pkt, topic, 1 );
  s->send( pkt, n );
  if ( waitFor( s, 9, 10000 ) ) { printf( "MQTT: SUBSCRIBE failed\n" ); netShutdown( 5 ); }
  printf( "MQTT: subscribed to %s\n", topic );
  printf( "A: listening. Press Ctrl-Break to stop.\n" );

  clockTicks_t lastPing = TIMER_GET_CURRENT( );

  while ( !CtrlBreakDetected ) {
    pumpRx( s );

    uint8_t type; uint8_t *body; int bl;
    int used;
    while ( ( used = nextPacket( &type, &body, &bl ) ) != 0 ) {
      if ( type == 3 && bl >= 2 ) {                 // PUBLISH
        int tl = ( body[0] << 8 ) | body[1];
        if ( 2 + tl <= bl ) {
          int pl = bl - 2 - tl;
          if ( pl > 95 ) pl = 95;
          int slot = -1;
          for ( int i = 0; i < MAX_PENDING; i++ ) if ( !pending[i].used ) { slot = i; break; }
          memcpy( pkt, body + 2 + tl, pl );
          pkt[pl] = 0;
          stamp( now );
          printf( "[%s] A rx: %s\n", now, (char *)pkt );
          if ( slot >= 0 ) {
            pending[slot].used = 1;
            pending[slot].due  = TIMER_GET_CURRENT( ) + TIMER_MS_TO_TICKS( 1000 );
            strcpy( pending[slot].text, (char *)pkt );
          }
        }
      }
      dropPacket( used );
    }

    // Echo anything whose one-second delay has elapsed.
    for ( int i = 0; i < MAX_PENDING; i++ ) {
      if ( pending[i].used && Timer_diff( pending[i].due, TIMER_GET_CURRENT( ) ) < 0x80000000ul ) {
        stamp( now );
        printf( "[%s] A echo: %s\n", now, pending[i].text );
        pending[i].used = 0;
      }
    }

    if ( Timer_diff( lastPing, TIMER_GET_CURRENT( ) ) > TIMER_SECS_TO_TICKS( 30 ) ) {
      pkt[0] = 0xC0; pkt[1] = 0;                    // PINGREQ
      s->send( pkt, 2 );
      lastPing = TIMER_GET_CURRENT( );
    }

    if ( s->isRemoteClosed( ) ) { printf( "A: broker closed the connection\n" ); break; }
  }

  pkt[0] = 0xE0; pkt[1] = 0;                        // DISCONNECT
  s->send( pkt, 2 );
  sleepTicks( 300, 1 );
  s->close( );
  TcpSocketMgr::freeSocket( s );
  netShutdown( 0 );
  return 0;
}