/********************************************************************************* * Copyright: (C) 2019 LingYun IoT System Studio * All rights reserved. * * Filename: main.c * Description: This file * * Version: 1.0.0(29/01/19) * Author: Guo Wenxue * ChangeLog: 1, Release initial version on "29/01/19 15:34:41" * ********************************************************************************/ #include #include #include #include #include #include #include #include "logger.h" #include "cJSON.h" #include "proc.h" #include "conf.h" #define PROG_VERSION "v1.0.0" #define DAEMON_PIDFILE "/tmp/.mqtt.pid" int check_set_program_running(int daemon); void *mqtt_sub_worker(void *args); static void program_usage(char *progname) { printf("Usage: %s [OPTION]...\n", progname); printf(" %s is LingYun studio MQTT daemon program running on RaspberryPi\n", progname); printf("\nMandatory arguments to long options are mandatory for short options too:\n"); printf(" -d[debug ] Running in debug mode\n"); printf(" -c[conf ] Specify configure file\n"); printf(" -h[help ] Display this help information\n"); printf(" -v[version ] Display the program version\n"); printf("\n%s version %s\n", progname, PROG_VERSION); return; } int main (int argc, char **argv) { float temp; float rh; int daemon = 1; pthread_t tid; mqtt_ctx_t ctx; char *conf_file=NULL; int debug = 0; int opt; char *progname=NULL; struct option long_options[] = { {"conf", required_argument, NULL, 'c'}, {"debug", no_argument, NULL, 'd'}, {"version", no_argument, NULL, 'v'}, {"help", no_argument, NULL, 'h'}, {NULL, 0, NULL, 0} }; progname = (char *)basename(argv[0]); /* Parser the command line parameters */ while ((opt = getopt_long(argc, argv, "c:dvh", long_options, NULL)) != -1) { switch (opt) { case 'c': /* Set configure file */ conf_file = optarg; break; case 'd': /* Set debug running */ daemon = 0; debug = 1; break; case 'v': /* Get software version */ printf("%s version %s\n", progname, PROG_VERSION); return 0; case 'h': /* Get help information */ program_usage(progname); return 0; default: break; } } if( !conf_file ) debug = 1; if( mqttd_parser_conf(conf_file, &ctx, debug)<0 ) { fprintf(stderr, "Parser mqtted configure file failure\n"); return -2; } install_proc_signal(); if( check_set_program_running(daemon) < 0 ) goto OUT; if( !debug ) 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"); while( ! g_signal.stop ) { sleep(1); //log_nrml("Main control thread continue running\n"); } OUT: mosquitto_lib_cleanup(); logger_term(); return 0; } /* ----- End of main() ----- */ int check_set_program_running(int daemon) { if( check_daemon_running(DAEMON_PIDFILE) ) { log_err("Program already running, process exit now"); return -1; } if( daemon ) { if( set_daemon_running(DAEMON_PIDFILE) < 0 ) { log_err("set program running as daemon failure\n"); return -2; } } else { if( record_daemon_pid(DAEMON_PIDFILE) < 0 ) { log_err("record program running PID failure\n"); return -3; } } return 0; } 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, ctx->subTopic, ctx->subQos); } 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); } int turn_camera(char *status) { FILE *fp = NULL; char buf[256]; int pid; int running; char cmd[256]; /* use lsusb command check Logic C920 camera exist or not */ fp = popen("lsusb | grep -v grep | grep C920", "r"); if( !fp ) { log_err("popen to check C920 failure: %s\n", strerror(errno)); return -1; } memset(buf, 0, sizeof(buf)); fgets(buf, sizeof(buf), fp); if( !strstr(buf, "C920") ) { log_err("Logic C920 camera not exist\n"); return -2; } log_nrml("Logic C920 camera exist\n"); fp=popen("ps aux | grep mjpg_streamer | grep -v grep | awk '{print $2}'", "r"); if( !fp ) { log_err("popen to check mjpg_streamer failure: %s\n", strerror(errno)); return -1; } memset(buf, 0, sizeof(buf)); fgets(buf, sizeof(buf), fp); pid = atoi(buf); if( pid > 1 ) { log_nrml("mjpg_streamer already working as pid[%d] now\n", pid); running = 1; } else { log_nrml("mjpg_streamer not running now\n"); running = 0; } if( running && strcasestr(status, "off") ) { log_warn("Stop mjpg_streamer working on C920 now\n"); system("/bin/video stop"); return 0; } if( !running && strcasestr(status, "on") ) { log_nrml("Start mjpg_streamer working on C920 now\n"); system("/bin/video start"); } return 0; } void proc_json_items(cJSON *root) { int i; cJSON *item; char *value; if( !root ) { log_err("Invalid input arguments $root\n"); return ; } item = cJSON_GetObjectItem(root, "camera"); if( !item ) { log_err("cJSON_Parse get camera failure: %s\n", cJSON_GetErrorPtr()); return ; } value = cJSON_PrintUnformatted(item); turn_camera(value); free(value); } void sub_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message) { mqtt_ctx_t *ctx = (mqtt_ctx_t *)userdata; cJSON *root = NULL; cJSON *item; char *value; if ( !message->payloadlen ) { log_err("%s (null)\n", message->topic); return ; } log_dbg("Subscriber receive message: '%s'\n", message->payload); root = cJSON_Parse(message->payload); if( !root ) { log_err("cJSON_Parse parser failure: %s\n", cJSON_GetErrorPtr()); return ; } item = cJSON_GetObjectItem(root, "id"); if( !item ) { log_err("cJSON_Parse get ID failure: %s\n", cJSON_GetErrorPtr()); goto OUT; } value = cJSON_PrintUnformatted(item); if( strcasecmp(value, ctx->id) ) { free(value); goto OUT; } free(value); log_nrml("Subscriber[%s] receive message: '%s'\n", ctx->id, message->payload); proc_json_items(root); OUT: cJSON_Delete(root); /* must delete it, or it will result memory leak */ return ; } void *mqtt_sub_worker(void *args) { mqtt_ctx_t *ctx = (mqtt_ctx_t *)args; struct mosquitto *mosq; bool session = true; mosq = mosquitto_new(NULL, session, ctx); if( !mosq ) { log_err("mosquitto_new failure\n"); return NULL; } /* set connnect to broker username and password */ if( strlen(ctx->uid)> 0 && strlen(ctx->pwd)> 0 ) mosquitto_username_pw_set(mosq, ctx->uid, ctx->pwd); /* 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 ) { /* connect to MQTT broker */ 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)); sleep(1); continue; } /* -1: use default timeout 1000ms 1: unused */ mosquitto_loop_forever(mosq, -1, 1); } mosquitto_destroy(mosq); return NULL; }