#include <sys/shm.h>
#include <sys/ipc.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <syslog.h>
#include "cJSON.h"
#include "ipc_session.h"
#include "mqtt_session.h"

#define INTERNAL_BROKER_ADDR "localhost"
#define INTERNAL_BROKER_PORT 1883
#define INTERNAL_BROKER_USER ""
#define INTERNAL_BROKER_PASS ""

#define KEEP_ALIVE_MAX  10
#define DEFAULT_IPC_QOS 0
#define CENTER_BROKER_CFG "/app/config/center_broker.json"

static inline int file_size2(const char *filename)
{
    struct stat statbuf;
    int size = 0;

    if (stat(filename, &statbuf) == 0)
    {
        size = statbuf.st_size;
    }

    return size;
}

static void warn_about_nng(const char * func)
{
    syslog(LOG_ERR, "Error, support for nng has been for %s\n", func);
}

static int load_broker_cfg(mqtt_client_t *client)
{
    int ret = 0;
    cJSON *root = NULL;
    cJSON *item = NULL;
    char *data = NULL;
    int size;
    FILE *fp = NULL;
    char *file = CENTER_BROKER_CFG;
    fp = fopen(file, "r");
    if (fp == NULL)
    {
        syslog(LOG_ERR, "The center broker config not found,%s no such file", CENTER_BROKER_CFG);
        ret = -1;
        goto out;
    }

    size = file_size2(file);
    data = malloc(size + 1);
    if (!data)
    {
        fclose(fp);
        syslog(LOG_ERR, "[UTILS] malloc faile, size:%d", size);
        ret = -1;
        goto out;
    }

    if (fread(data, sizeof(char), size, fp) == 0)
        data[0] = '\0';
    fclose(fp);
    data[size] = 0;

    root = cJSON_Parse(data);
    if (!root)
    {
        syslog(LOG_ERR, "Invalid config format,need a json format");
        ret = -1;
        goto out;
    }
    item = cJSON_GetObjectItem(root, "host");
    if (item)
    {
        if (item->valuestring)                   
            strcpy(client->addr, item->valuestring);   
    }
    item = cJSON_GetObjectItem(root, "port");
    if (item)
    {                
        client->port = item->valueint;   
    }
    item = cJSON_GetObjectItem(root, "user");
    if (item)
    {
        if (item->valuestring)                   
            strcpy(client->user, item->valuestring);   
    }
    item = cJSON_GetObjectItem(root, "pass");
    if (item)
    {
        if (item->valuestring)                   
            strcpy(client->pass, item->valuestring);   
    }
out:
    cJSON_Delete(root);
    if (data != NULL)
    {
        free(data);
    }
    return ret;
}

ipc_session_t *ipc_session_new(char *id, void *user_data, ipc_mode_e mode)
{
    mqtt_client_t client = {0};
    int ret = load_broker_cfg(&client);
    ipc_session_t *session = calloc(1, sizeof(ipc_session_t));
    if (session == NULL)
    {
        syslog(LOG_USER | LOG_ERR, "%s:%d malloc error", __FILE__, __LINE__);
        return NULL;
    }
    if (mode == IPC_DEFAULT)
    {
        mode = IPC_MQTT;
    }
    session->user_data = user_data;
    session->mode = mode;

    if (mode == IPC_MQTT) {
        session->s.mqtt_session = mqtt_session_new(id, user_data);
        if(ret == 0) 
            mqtt_session_set_address(session->s.mqtt_session, client.addr, client.port, client.user, client.pass);
        else
            mqtt_session_set_address(session->s.mqtt_session, INTERNAL_BROKER_ADDR, INTERNAL_BROKER_PORT, INTERNAL_BROKER_USER, INTERNAL_BROKER_PASS);
        mqtt_session_set_opts(session->s.mqtt_session, DEFAULT_IPC_QOS, KEEP_ALIVE_MAX);
		return session;
    }

    free(session);
    warn_about_nng(__func__);
    return NULL;
}

int ipc_session_subscribe(ipc_session_t *session, const char *topic)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_subscribe(session->s.mqtt_session, topic);
    }
    warn_about_nng(__func__);
    return -1;
}

int ipc_session_set_callbacks(ipc_session_t *session, int (*on_message)(void *obj, ipc_msg_t *msg),
                              int (*on_state_change)(void *obj, int state))
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_set_callbacks(session->s.mqtt_session, on_message, on_state_change);
    }
    warn_about_nng(__func__);
    return -1;
}

int ipc_session_publish(ipc_session_t *session, char *topic, const void *payload, int payloadlen)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_publish(session->s.mqtt_session, topic, payload, payloadlen);
    }
    warn_about_nng(__func__);
    return -1;
}

int ipc_session_start(ipc_session_t *session)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_start(session->s.mqtt_session);
    }
    warn_about_nng(__func__);
    return -1;
}

int ipc_session_destroy_and_free(ipc_session_t *session)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_destroy_and_free(session->s.mqtt_session);
    }
    warn_about_nng(__func__);
    return -1;
}

int ipc_session_set_opts(ipc_session_t *session, int qos, int keepalive)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_set_opts(session->s.mqtt_session, qos, keepalive);
    }
    return 0;
}

int ipc_session_set_address(ipc_session_t *session, char *addr, unsigned short port, char *user, char *pass)
{
    if (session->mode == IPC_MQTT) {
        return mqtt_session_set_address(session->s.mqtt_session, addr, port, user, pass);
    }
    return 0;
}