SmartPDCode/pro/src/websocket_handle.c

#include "websocket_handle.h"
#include "mongoose.h"
#include "pthread.h"
#include "elog.h"
#include "common.h"
#include "sys.h"
#include "upg.h"
#include "lock.h"
#include "file.h"
#include "thread.h"
#include "sqlite_handle.h"
#include "json_handle.h"
#include "cascade.h"
#include "cfg.h"

#define USE_WS2
#define WS_SEND_TLEN_MAX  (1024*1024*10)

#define WS_MAX    10
#define USE_ONE_CONN

typedef struct mg_mgr           mg_mgr_t;
typedef struct mg_connection    mg_conn_t;
typedef struct mg_ws_message    mg_ws_msg_t;
typedef struct mg_http_message  mg_http_msg_t;
typedef struct mg_tls_opts      mg_tls_t;

typedef struct {
    mg_mgr_t  mgr;
    char      wpath[200];
    char      root[200];
    mg_tls_t  tls;

    mg_conn_t *c;

    int       inited;
    int       ws_cnt;
    pwrall_info_t pwrall;
}ws_handle_t;

static ws_handle_t wsHandle;

static int ws_is_online(ws_handle_t *wh) 
{
#ifdef USE_ONE_CONN
    return (wh->c)?1:0;
#else
    return (wh->ws_cnt>0)?1:0;
#endif
}
static int ws_send(ws_handle_t *wh, void *data, int len, int isBinary) 
{
    if((!wh->c) || (!data) || (len<=0)) {
        return -1;
    }
#ifdef USE_WS2
    mg_ws_send2(wh->c, data, len, isBinary?WEBSOCKET_OP_BINARY:WEBSOCKET_OP_TEXT, WS_SEND_TLEN_MAX);
#else
    mg_ws_send(wh->c, data, len, isBinary?WEBSOCKET_OP_BINARY:WEBSOCKET_OP_TEXT);
#endif

    return 0;
}
static int ws_broadcast(ws_handle_t *wh, void *data, int len, int isBinary) 
{
    int r=-1;
    mg_conn_t *c;

    for (c=wh->mgr.conns; c!=NULL; c=c->next) {
#ifdef USE_WS2
        mg_ws_send2(c, data, len, isBinary?WEBSOCKET_OP_BINARY:WEBSOCKET_OP_TEXT, WS_SEND_TLEN_MAX);
#else
        mg_ws_send(c, data, len, isBinary?WEBSOCKET_OP_BINARY:WEBSOCKET_OP_TEXT);
#endif
        r = 0;
    }

    return r;
}
static int ws_send_data(ws_handle_t *wh, void *data, int len)
{
#ifdef USE_ONE_CONN
    return ws_send(wh, data, len, 0);
#else
    return ws_broadcast(wh, data, len, 0);
#endif
}
//////////////////////////////////////////////////////////////////////
static int ws_all_update(ws_handle_t *h)
{
    char *json;
    int r,r1,r2,r3;
    GlobalDeviceManager* gdm= &__globalDeviceManage;
    GlobalDeviceManager* gdm2=&__globalDeviceManage2;
    GlobalDeviceManager* pdm=NULL;

    pwrall_info_t *pall=&h->pwrall;
    if(gdm->_globalDevInfo.product.pwr_type==SmartPDU_Tree_AC_Tree ||
        gdm->_globalDevInfo.product.pwr_type==SmartPDU_Tree_AC_One ||
        gdm->_globalDevInfo.product.pwr_type==SmartPDU_Tree_AC_One_B)
    {
        pall->ph3 = 1;

        if (cur_dev_addr == 0)
        {
            lock_s_hold(LOCK_ID_POWER_UPDATE);
            r1 = dev_search_latest_t_ac_power_statistic_info(0, 0, &pall->ch[0], gdm);
            r2 = dev_search_latest_t_ac_power_statistic_info(0, 1, &pall->ch[1], gdm);
            r3 = dev_search_latest_t_ac_power_statistic_info(0, 2, &pall->ch[2], gdm);
            lock_s_release(LOCK_ID_POWER_UPDATE);
        }
        else
        {
            cascade_lock();
            r1 = dev_search_latest_t_ac_power_statistic_info(0, 0, &pall->ch[0], gdm2);
            r2 = dev_search_latest_t_ac_power_statistic_info(0, 1, &pall->ch[1], gdm2);
            r3 = dev_search_latest_t_ac_power_statistic_info(0, 2, &pall->ch[2], gdm2);
            cascade_unlock();
        }
        if (0 == r1 && 0 == r2 && 0 == r3)
        {
            json = over_all_pwr_monitor_Tree_AC_to_json(&pall->ch[0], &pall->ch[1], &pall->ch[2]);
            ws_send_data(h, json, strlen(json));
            cJSON_free(json);
        }
    }
    else
    {
        pall->ph3 = 0;
        if(cur_dev_addr==0) {
            lock_s_hold(LOCK_ID_POWER_UPDATE);
            r = dev_search_latest_power_statistic_info(gdm, &pall->ch[0]);
            lock_s_release(LOCK_ID_POWER_UPDATE);
        }
        else {
            cascade_lock();
            r = dev_search_latest_power_statistic_info(gdm2, &pall->ch[0]);
            cascade_unlock();
        }
        if (0==r)
        {
            json = ws_over_status_ack_to_json(0, &pall->ch[0]);
            ws_send_data(h, json, strlen(json));
            cJSON_free(json);
        }
    }

    return 0;
}
static int ws_chn_update(ws_handle_t *h)
{
    int r;
    char *json;
    _OverChnPwrAckInfo chInfo;
    GlobalDeviceManager* gdm= &__globalDeviceManage;
    GlobalDeviceManager* gdm2=&__globalDeviceManage2;
    GlobalDeviceManager* pdm=NULL;

    INIT_LIST_HEAD(&chInfo.list);
    if(cur_dev_addr==0) {
        lock_s_hold(LOCK_ID_POWER_UPDATE);
        r = dev_search_latest_power_All_info(gdm,&chInfo, cur_dev_addr);
        lock_s_release(LOCK_ID_POWER_UPDATE);
        pdm = gdm;
    }
    else {
        cascade_lock();
        r = dev_search_latest_power_All_info(gdm2,&chInfo, cur_dev_addr);
        cascade_unlock();
        pdm = gdm2;
    }
    if (0 == r)
    {
        json = ws_chn_status_ack_to_json(1, &chInfo, pdm);
        ws_send_data(h, json, strlen(json));
        cJSON_free(json);
    }
    _OverChnPwrAckInfo *node, *next;
    list_for_each_entry_safe(node, next, &chInfo.list, list)
    {
        list_del(&node->list);
        free(node);
    }

    return 0;
}
static int ws_sensor_update(ws_handle_t *h)
{
    char *json;
    GlobalDeviceManager* gdm= &__globalDeviceManage;

    //lock_s_hold(LOCK_ID_SENSOR);
    json = ws_sensor_status_ack_to_json(&gdm->_globalSensorManger);
    //lock_s_release(LOCK_ID_SENSOR);
    if(json) {
        ws_send_data(h, json, strlen(json));
        cJSON_free(json);
    }

    return 0;
}
static int ws_breaker_update(ws_handle_t *h)
{
    char *json;
    GlobalDeviceManager* gdm= &__globalDeviceManage;
    GlobalDeviceManager* gdm2=&__globalDeviceManage2;

    if(cur_dev_addr==0) { 
        json = breaker_status_ack_to_json(gdm,0);
    }
    else {
        json = breaker_status_ack_to_json(gdm2,cur_dev_addr);
    }
    ws_send_data(h, json, strlen(json));
    cJSON_free(json);

    return 0;
}
static int ws_threshold_update(ws_handle_t *h)
{
    char *json;
    GlobalDeviceManager* dm= &__globalDeviceManage;
    
    if(cur_dev_addr==0) { 
        json = threshold_status_ack_to_json(dm);
        ws_send_data(h, json, strlen(json));
        cJSON_free(json);
    }
    return 0;
}

static int ws_upg_update(ws_handle_t *h)
{
    handle_t l=upg_board_all_get();

    char* json = board_to_json(l, 1);
    ws_send_data(h, json, strlen(json));
    cJSON_free(json);

    return 0;
}
/////////////////////////////////////////////////////////////

static int ws_update(ws_handle_t *h)
{
    int r;

    if(!ws_is_online(h)) {
        return -1;
    }

    //overall refresh
    r = ws_all_update(h);
    
    //channel refresh
    r = ws_chn_update(h);

    //sensor refresh
    r = ws_sensor_update(h);
    
    //breaker refresh
    r = ws_breaker_update(h);

    //threshold update
    //r = ws_threshold_update();

    //upgrade refresh
    r = ws_upg_update(h);

    return r;
}
static void fn(mg_conn_t *c, int ev, void *ev_data) 
{
  ws_handle_t *wh=&wsHandle;

  switch(ev) {
    case MG_EV_OPEN:
    {
      
    }
    break;

    case MG_EV_CLOSE:
    {
      if (c->is_websocket )
      {
        if(wh->ws_cnt>0) {
            wh->ws_cnt--;
        }

        if(c==wh->c) {
            wh->c = NULL;
        }
      }
      log_i("ws upgrade del connection ID:%d\n", c->id);
    }
    break;

    case MG_EV_WS_OPEN:
    {
        wh->ws_cnt++;
    }
    break;

    case MG_EV_ACCEPT:
    {
        if (c->fn_data) {
            mg_tls_init(c, &wh->tls);
        }
    }
    break;

    case MG_EV_HTTP_MSG:
    {
        mg_http_msg_t *hm = (mg_http_msg_t *) ev_data;
        if (mg_match(hm->uri, mg_str("/websocket/PDU-WebSocket"), NULL)) {
            mg_ws_upgrade(c, hm, NULL);
            wh->c = c;
            log_i("ws upgrade new connection ID:%d\n", c->id);
        }
        else if (mg_match(hm->uri, mg_str("/rest"), NULL)) {
            mg_http_reply(c, 200, "", "{\"result\": %d}\n", 123);
        }
    }
    break;

    case MG_EV_WS_MSG:
    {
        // struct mg_ws_message *wm = (struct mg_ws_message *) ev_data;
        // mg_ws_send(c, wm->data.ptr, wm->data.len, WEBSOCKET_OP_TEXT);
    }
    break;
  }

}

static void timer_fn(void *arg)
{
    ws_handle_t *h=(ws_handle_t*)arg;
    ws_update(h);
}

static void* ws_thread(void* arg)
{
    int r=0;
    thread_handle_t *th=(thread_handle_t*)arg;
    ws_handle_t *h=(ws_handle_t*)th->arg;
    const char *ws_addr="ws://[::]:6785";
    const char *wss_addr="wss://[::]:6786";

    //mg_log_set(MG_LL_DEBUG);
    mg_mgr_init(&h->mgr);  // Initialise event manager
    mg_http_listen(&h->mgr, ws_addr, fn, NULL);  // Create HTTP listener
    mg_http_listen(&h->mgr, wss_addr, fn, (void*)1);
#if (CHIP_TYPE == CHIP_T113s)
    h->tls.cert = mg_str(file_load2("/mnt/UDISK/app/cert/server.crt",0));
    h->tls.key = mg_str(file_load2("/mnt/UDISK/app/cert/server.key",0));
#else
    h->tls.cert = mg_str(file_load2("/root/run/app/cert/server.crt",0));
    h->tls.key = mg_str(file_load2("/root/run/app/cert/server.key",0));  
#endif
    h->inited = 1;

    mg_timer_add(&h->mgr, 1000, MG_TIMER_REPEAT, timer_fn, h);
    while(th->quit==0) {
        mg_mgr_poll(&h->mgr, 100);                    // Infinite event loop
    }
    h->inited = 0;
    mg_mgr_free(&h->mgr);

    pthread_exit(NULL);
}


int websocket_init(void) 
{
    ws_handle_t *wh=&wsHandle;

    memset(wh, 0, sizeof(ws_handle_t));
    thread_start(THREAD_ID_WS, ws_thread, wh);
    return 0;
}



pwrall_info_t *websocket_get_pwrall(void)
{
    return &wsHandle.pwrall;
}