File size: 3,193 Bytes
c722e6d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | // 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;
}
|