update mqttd main.c, add mosquitto publisher and subscriber

guowenxue

2019-06-25 02294c1c430ab80bd4543af28f867c5c2927e5c1

update mqttd main.c, add mosquitto publisher and subscriber

 mqttd/main.c

@@ -14,16 +14,36 @@
#include <time.h>
#include <unistd.h>

#include <mosquitto.h>

#include "logger.h"
#include "proc.h"
#include "hal.h"

#define DAEMON_PIDFILE             "/tmp/.mqtt.pid"

#define MQTT_SYS_TOPIC             "$Sys/Studio/"
#define MQTT_SYS_DLINK_TOPIC       MQTT_SYS_TOPIC"Downlink"
#define MQTT_SYS_ULINK_TOPIC       MQTT_SYS_TOPIC"Uplink"

enum
{
    Qos0, /* 发送者只发送一次消息，不进行重试，Broker不会返回确认消息。在Qos0情况下，Broker可能没有接受到消息 */
    Qos1, /* 发送者最少发送一次消息，确保消息到达Broker，Broker需要返回确认消息PUBACK。在Qos1情况下，Broker可能接受到重复消息 */
    Qos2, /* Qos2使用两阶段确认来保证消息的不丢失和不重复。在Qos2情况下，Broker肯定会收到消息，且只收到一次  */
};

typedef struct mqtt_ctx_s 
{
    int        conf;
    char          host[128];  /* MQTT broker server name  */
    int           port;       /* MQTT broker listen port  */
    int           keepalive;  /* MQTT broker send PING message to subsciber/publisher keepalive timeout<seconds> */

    /* Publisher */
    int           pubQos;

    /* Subscriber */
    int           subQos;
} mqtt_ctx_t;  


@@ -40,6 +60,12 @@
    pthread_t          tid;
    mqtt_ctx_t         ctx;

    strcpy(ctx.host, "127.0.0.1");
    ctx.port = 10883;
    ctx.keepalive = 60;
    ctx.pubQos = Qos0;
    ctx.subQos = Qos2;

    //if( logger_init("mqttd.log", LOG_LEVEL_DEBUG, 1024) < 0 ) 
    if( logger_init(DBG_LOG_FILE, LOG_LEVEL_DEBUG, 1024) < 0 ) 
    {
@@ -47,26 +73,6 @@
        return -2;
    }
    log_dbg("Logger system initialise ok\n");

    install_proc_signal();

    if( check_set_program_running(daemon) < 0 ) 
    {
        goto OUT;
    }

    if( thread_start(&tid, mqtt_sub_worker, &ctx ) < 0 )
    {
        log_fatal("Start MQTT subscibe worker thread failure\n");
        goto OUT;
    }

    if( thread_start(&tid, mqtt_pub_worker, &ctx) < 0 )
    {
        log_fatal("Start MQTT publish worker thread failure\n");
        goto OUT;
    }


#if 0
    if( hal_init() < 0 )
@@ -77,15 +83,37 @@
    log_nrml("HAL initialise ok\n");
#endif
    
    install_proc_signal();

    if( check_set_program_running(daemon) < 0 ) 
        goto OUT;

    mosquitto_lib_init();

    if( thread_start(&tid, mqtt_sub_worker, &ctx ) < 0 )
    {
        log_fatal("Start MQTT subsciber worker thread failure\n");
        goto OUT;
    }
    log_nrml("Start MQTT subsciber worker thread ok\n");

    if( thread_start(&tid, mqtt_pub_worker, &ctx) < 0 )
    {
        log_fatal("Start MQTT publisher worker thread failure\n");
        goto OUT;
    }
    log_nrml("Start MQTT publisher worker thread ok\n");


    while( ! g_signal.stop )
    {
        msleep(1000);
        log_nrml("Main control thread continue running\n");
        //log_nrml("Main control thread continue running\n");
    }


OUT:
    mosquitto_lib_cleanup();
    logger_term();
    
    return 0;
@@ -122,27 +150,125 @@

void *mqtt_pub_worker(void *args)
{
    mqtt_ctx_t           *ctx = (mqtt_ctx_t *)args;
    mqtt_ctx_t             *ctx = (mqtt_ctx_t *)args;
    struct mosquitto       *mosq;
    bool                    session = true;
    char                    msg[128];  
    float                   temp = 25.64;
    float                   rh = 65.35;
    int                     rv = 0;

    while( !g_signal.stop )
    mosq = mosquitto_new(NULL, session, ctx);
    if( !mosq )
    {
        msleep(1000);
        log_nrml("MQTT publish thread running\n");
        log_err("mosquitto_new failure\n");
        return NULL;
    }

    while( !g_signal.stop )
    { 
        if( mosquitto_connect(mosq, ctx->host, ctx->port, ctx->keepalive) )
        {
            log_err("Publisher connect to broker[%s:%d] failure: %s\n", ctx->host, ctx->port, strerror(errno)); 
            msleep(1000);
            continue;
        }
        log_nrml("Publisher connect to broker server[%s:%d] successfully\n", ctx->host, ctx->port); 


        memset(msg, 0, sizeof(msg));
        snprintf(msg, sizeof(msg), "{ \"temp\":\"%.2f\",  \"R&H\":\"%.2f%%\" }", temp, rh);


        rv = mosquitto_publish(mosq, NULL, MQTT_SYS_ULINK_TOPIC, strlen(msg)+1, msg, ctx->pubQos, 0);
        if( rv )
        {
            log_err("Publisher publish message failure: %d\n", rv);
        }
        else
        {
            log_nrml("Publisher publish message '%s' ok\n", msg);
        }


        mosquitto_disconnect(mosq);
        msleep(3000);
    }

    mosquitto_destroy(mosq);
    return NULL;
}


void sub_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    mqtt_ctx_t             *ctx = (mqtt_ctx_t *)userdata;

    if( result )
    {
        log_err("Subscriber connect to broker server failed, rv=%d\n", result);
        return ;
    }
    
    log_nrml("Subscriber connect to broker server[%s:%d] successfully\n", ctx->host, ctx->port); 
    mosquitto_subscribe(mosq, NULL, MQTT_SYS_TOPIC"#", ctx->subQos);
    //mosquitto_subscribe(mosq, NULL, MQTT_SYS_DLINK_TOPIC, 2);
}

void sub_disconnect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    mqtt_ctx_t             *ctx = (mqtt_ctx_t *)userdata;

    log_warn("Subscriber disconnect to broker server[%s:%d], reason=%d\n", ctx->host, ctx->port, result); 
}


void sub_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
    if(message->payloadlen)
    {
        log_nrml("Subscriber Recv: %s->%s\n", message->topic, message->payload);
    }
    else 
    {
        log_nrml("%s (null)\n", message->topic);
    }
}


void *mqtt_sub_worker(void *args)
{
    mqtt_ctx_t           *ctx = (mqtt_ctx_t *)args;
    mqtt_ctx_t             *ctx = (mqtt_ctx_t *)args;
    struct mosquitto       *mosq;
    bool                    session = true;

    while( !g_signal.stop )
    mosq = mosquitto_new(NULL, session, ctx);
    if( !mosq )
    {
        msleep(1000);
        log_nrml("MQTT subscibe thread running\n");
        log_err("mosquitto_new failure\n");
        return NULL;
    }

    /* set callback functions */
    mosquitto_connect_callback_set(mosq, sub_connect_callback);
    mosquitto_disconnect_callback_set(mosq, sub_disconnect_callback);
    mosquitto_message_callback_set(mosq, sub_message_callback);


    while( !g_signal.stop )
    { 
        if( mosquitto_connect(mosq, ctx->host, ctx->port, ctx->keepalive) )
        {
            log_err("Subscriber connect to broker[%s:%d] failure: %s\n", ctx->host, ctx->port, strerror(errno)); 
            msleep(1000);
            continue;
        }

        /* -1: use default timeout 1000ms  1: unused */
        mosquitto_loop_forever(mosq, -1, 1);
    }

    mosquitto_destroy(mosq);
    return NULL;
}


 mqttd/makefile

@@ -40,6 +40,8 @@
CFLAGS+=-Ihal -Ilylib
LDFLAGS+=-L hal -lhal -Llylib -llylib

LDFLAGS+=-lmosquitto

SRCFILES = $(wildcard *.c)
IMAGE_NAME=$(shell basename ${PWD})