| |
| |
| |
| #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 ) { |
| 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 ); |
| } |
|
|
| |
| 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; |
| 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; |
| s->send( pkt, 2 ); |
| sleepTicks( 300, 1 ); |
| s->close( ); |
| TcpSocketMgr::freeSocket( s ); |
| netShutdown( 0 ); |
| return 0; |
| } |
|
|
