#include <mosquitto.h>
#include <pthread.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#ifdef _WIN32
    #include <windows.h>
#else
    #include <unistd.h>
#endif
#include "list.h"
#include "map.h"
#include "circle_buff.h"
#include "bmser_ipc.h"

//#define USER_STR_LEN 32
//#define PASSWD_STR_LEN 32
//#define CLIENT_ID_LEN 32
//
#define IPC_HOST_LEN 128
#define IPC_CIRCULAR_BUF_CAP 128 

struct ipc_client {
    uint32_t magic_num;
    struct list_head list;
    struct list_head retry;

    struct mosquitto *mosq;
    struct ipc_event_ops event_ops;
    void *priv_data;
    int clean_session;
    bool connected;
    bool in_loop;
    char srv_host[IPC_HOST_LEN + 1];
    int srv_port;
    int keep_alive;

    pthread_mutex_t lock;
    struct map *topic_callbacks;
    struct circle_buff *circle_buf;
};

struct ipc_timer {
    int interval;
    pthread_t tid;
};

struct ipc_context {
    struct list_head list_head;
    struct list_head retry_head;
    pthread_mutex_t lock;
    int ref_count;

    struct ipc_timer retry_timer;
};


static int map_error_code(int err_code)
{
    switch(err_code) {
        case MOSQ_ERR_SUCCESS:
	    return IPC_ERR_SUCCESS;
	    
        case MOSQ_ERR_NOMEM:
	    return IPC_ERR_NO_MEM;
	    
        case MOSQ_ERR_INVAL:
	    return IPC_ERR_INVAL;
	    
        case MOSQ_ERR_NO_CONN:
	    return IPC_ERR_NO_CONN;

        case MOSQ_ERR_PAYLOAD_SIZE:
	    return IPC_ERR_PAYLOAD_SIZE;

        case MOSQ_ERR_MALFORMED_UTF8:
	    return IPC_ERR_MALFORMED_UTF8;

        case MOSQ_ERR_QOS_NOT_SUPPORTED:
	    return IPC_ERR_QOS_NOT_SUPPORTED;

        case MOSQ_ERR_OVERSIZE_PACKET:
	    return IPC_ERR_OVERSIZE_PACKET;

	default:
	    return IPC_ERR_UNKNOWN;
    }
}

#define MQTT_CONTEXT_INIT(name) \
    { \
	.list_head = LIST_HEAD_INIT(name.list_head), \
        .retry_head = LIST_HEAD_INIT(name.retry_head), \
	.lock = PTHREAD_MUTEX_INITIALIZER, \
	.ref_count = 0, \
	.retry_timer = { .interval = 5 }\
    }
#define MQTT_CONTEXT(name) struct ipc_context name = MQTT_CONTEXT_INIT(name)

MQTT_CONTEXT(ipc_ctx);

static void *ipc_retry_thread_entry(void *priv)
{
    while (1) {
        pthread_mutex_lock(&ipc_ctx.lock);
        struct list_head *cur, *next;
        list_for_each_safe(cur, next, &(ipc_ctx.retry_head)) {
            struct ipc_client *c = container_of(cur, struct ipc_client, retry);

            if (MOSQ_ERR_SUCCESS == mosquitto_connect_async(c->mosq, c->srv_host, c->srv_port, c->keep_alive)) {
                c->connected = true;
        	list_del(&c->retry);
            }
        }
        pthread_mutex_unlock(&ipc_ctx.lock);

	//printf("in retry_thread_entry, after iterating....\n");
#ifdef WIN32
        Sleep(ipc_ctx.retry_timer.interval * 1000);
#else
        sleep(ipc_ctx.retry_timer.interval);
#endif
    }

    return NULL;
}

static int ipc_lib_init()
{
    int rc;

    pthread_mutex_lock(&ipc_ctx.lock);

    ipc_ctx.ref_count++;

    if (ipc_ctx.ref_count > 1) {
	rc = 0;
	goto end;
    }

    printf("call mosquitto_lib_init.\n");
    rc = mosquitto_lib_init();

    (void)pthread_create(&ipc_ctx.retry_timer.tid, NULL, ipc_retry_thread_entry, NULL);
#if 0
#ifndef WIN32
    pthread_setname_np(ipc_ctx.retry_timer.tid, "retry");
#endif
#endif

end:   
    pthread_mutex_unlock(&ipc_ctx.lock);
    return rc;
}

static void ipc_lib_cleanup()
{
    pthread_mutex_lock(&ipc_ctx.lock);

    if (!ipc_ctx.ref_count) {
	fprintf(stderr, "mqtt library NOT initialized yet.\n");
	goto end; 
    }

    if (ipc_ctx.ref_count > 1) {
        ipc_ctx.ref_count--;
	goto end;
    }

    printf("stopping retry thread....\n");
    pthread_cancel(ipc_ctx.retry_timer.tid);
    pthread_join(ipc_ctx.retry_timer.tid, NULL);
    printf("retry thread stopped\n");

    mosquitto_lib_cleanup();
    
end:
    pthread_mutex_unlock(&ipc_ctx.lock);
    return;
}

static void mosq_on_connect(struct mosquitto *mosq, void *obj, int rc)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    c->connected = true;

    if (c->event_ops.on_connect)
        c->event_ops.on_connect(c, c->priv_data, rc);
}

static void mosq_on_disconnect(struct mosquitto *mosqm, void *obj, int rc)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    if (c->event_ops.on_disconnect)
        c->event_ops.on_disconnect(c, c->priv_data, rc);
}

static void mosq_on_publish(struct mosquitto *mosq, void *obj, int mid)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    if (c->event_ops.on_publish)
        c->event_ops.on_publish(c, c->priv_data, mid);
}

static void mosq_on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_cnt, const int *qos_granted)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    if (c->event_ops.on_subscribe) {
        c->event_ops.on_subscribe(c, c->priv_data, mid, qos_cnt, qos_granted);
    }
}

static void mosq_on_unsubscribe(struct mosquitto *mosq, void *obj, int mid)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    if (c->event_ops.on_unsubscribe)
        c->event_ops.on_unsubscribe(c, c->priv_data, mid);
}

static void ipc_fill_message(struct ipc_message *m, const struct mosquitto_message *mosq_m, bool is_duplicate)
{
    m->mid = mosq_m->mid;

    if (is_duplicate) {
	m->t.name = (char *)calloc(1, strlen(mosq_m->topic) + 1);
	memcpy(m->t.name, mosq_m->topic, strlen(mosq_m->topic));
    } else {
        m->t.name = mosq_m->topic;
    }

    m->t.qos = mosq_m->qos;

    
    if (is_duplicate) {
        m->p.buf = calloc(1, mosq_m->payloadlen + 1);
	memcpy(m->p.buf, mosq_m->payload, mosq_m->payloadlen);

    } else {
        m->p.buf = mosq_m->payload;
    }
    m->p.len = mosq_m->payloadlen;
    m->retain = mosq_m->retain;
}

static void mosq_on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *mosq_m)
{
    struct ipc_client *c = (struct ipc_client *)obj;    

    struct ipc_message m;
    ipc_fill_message(&m, mosq_m, false);

    /* lookup topic in the rbl tree */
    pthread_mutex_lock(&c->lock);

    struct map_iterator *it = map_find(c->topic_callbacks, mosq_m->topic);

    if (it) {
	ipc_msg_cb_func cb = (ipc_msg_cb_func)it->val;
	cb(c, &m);

    } else if (c->event_ops.on_message) {
        c->event_ops.on_message(c, c->priv_data, &m);
    
    } else {
        struct ipc_message ipc_msg;
        ipc_fill_message(&ipc_msg, mosq_m, true);

	if (cb_is_full(c->circle_buf)) {
	    struct ipc_message tmp;
            cb_pop_front(c->circle_buf, &tmp);

	    free(tmp.t.name);
	    free(tmp.p.buf);
	}

	cb_push_back(c->circle_buf, &ipc_msg);
    }
    
    pthread_mutex_unlock(&c->lock);
}

static void ipc_set_event_ops(struct ipc_client *c, struct ipc_event_ops *ops)
{
    struct mosquitto *mosq;
    
    c->event_ops.on_connect = ops->on_connect;
    c->event_ops.on_disconnect = ops->on_disconnect;
    c->event_ops.on_publish = ops->on_publish;
    c->event_ops.on_subscribe = ops->on_subscribe;
    c->event_ops.on_unsubscribe = ops->on_unsubscribe;
    c->event_ops.on_message = ops->on_message;

    mosq = c->mosq;
    mosquitto_connect_callback_set(mosq, mosq_on_connect);
    mosquitto_disconnect_callback_set(mosq, mosq_on_disconnect);
    mosquitto_publish_callback_set(mosq, mosq_on_publish);
    mosquitto_subscribe_callback_set(mosq, mosq_on_subscribe);
    mosquitto_unsubscribe_callback_set(mosq, mosq_on_unsubscribe);
    mosquitto_message_callback_set(mosq, mosq_on_message);
}

static void ipc_message_free(void *message)
{
    struct ipc_message *ipc_msg = message;
    free(ipc_msg->t.name);
    free(ipc_msg->p.buf);
}

int bmser_ipc_create(const struct ipc_client_info *ci, struct ipc_client **client)
{
    if (!ci || !client) {
	fprintf(stderr, "param 'ci' OR 'client' is NULL\n");
        return IPC_ERR_INVAL;
    }

    if (ci->srv_host && strlen(ci->srv_host) > IPC_HOST_LEN) {
	fprintf(stderr, "srv_host string length > %d\n", IPC_HOST_LEN);
        return IPC_ERR_INVAL;	
    }

    int rc = IPC_ERR_SUCCESS;

    ipc_lib_init();

    struct ipc_client *c = malloc(sizeof(struct ipc_client));
    if (!c) {
	fprintf(stderr, "out of memory\n");
	rc = IPC_ERR_NO_MEM;
	goto err;
    }

    c->magic_num = 0xbeefbeef;
    pthread_mutex_init(&c->lock, NULL);
    c->priv_data = ci->priv_data;
    c->mosq = NULL;

    struct mosquitto *mosq = mosquitto_new(ci->id, ci->clean_session, c);
    if (!mosq) {
        if (errno == ENOMEM) 
	    rc = IPC_ERR_NO_MEM;
        
	if (errno == EINVAL)
	    rc = IPC_ERR_INVAL;

	goto err;
    }
    //printf(" in %s, mosq-ctx: %p created \n", __func__, mosq);
    c->mosq = mosq;
    c->connected = false;
    c->in_loop = false;
    c->topic_callbacks = map_str_cptr_create();
    c->circle_buf = cb_create(IPC_CIRCULAR_BUF_CAP, sizeof(struct ipc_message), ipc_message_free);
    if (!c->circle_buf) {
        rc = IPC_ERR_NO_MEM;
	goto err;
    }

    INIT_LIST_HEAD(&c->list);
    INIT_LIST_HEAD(&c->retry);

    ipc_set_event_ops(c, ci->event_ops);

#ifdef WIN32
    strcpy_s(c->srv_host, sizeof(c->srv_host), ci->srv_host ? ci->srv_host : "127.0.0.1");
#else
    strcpy(c->srv_host, ci->srv_host ? ci->srv_host : "127.0.0.1");
#endif
    c->srv_port = ci->srv_port ? ci->srv_port : 1883;
    c->keep_alive = ci->keep_alive;

    int ret;
    ret = mosquitto_connect_async(c->mosq, c->srv_host, c->srv_port, c->keep_alive);
    if (ret != MOSQ_ERR_SUCCESS) {
	    if (ret != MOSQ_ERR_ERRNO) {
	        rc = map_error_code(ret);
                goto err;
	    }

	    // do fault tolerance if MOSQ_ERR_ERRNO, will do reconnecting in retry timer
            //printf("mosquitto_connect_async failed(errno=%d), make sure MQTT broker is working normally!\n", errno);

    } else {
        c->connected = true;
    }

    ret = mosquitto_loop_start(c->mosq);
    if (ret != MOSQ_ERR_SUCCESS) {
	fprintf(stderr, "mosquitto_connect_async return %d\n", ret);
	rc = map_error_code(ret);
	goto err;
    }
    c->in_loop = true;

    pthread_mutex_lock(&ipc_ctx.lock);
    if (c->connected == false) {
        list_add_tail(&c->retry, &ipc_ctx.retry_head);
    }
    list_add_tail(&c->list, &ipc_ctx.list_head);
    pthread_mutex_unlock(&ipc_ctx.lock);

    *client = c;
    return rc;

err:
    if (c) {
	bmser_ipc_destroy(c);
    }

    *client = NULL;
    return rc;
}

void bmser_ipc_destroy(struct ipc_client *c)
{
    if (c->magic_num != 0xbeefbeef) {
        printf("invalid ipc client %p\n", c);
	return;
    }

    struct mosquitto *mosq = c->mosq;
    if (!mosq) {
	goto end;
    }

    /* pick up from lists */
    pthread_mutex_lock(&ipc_ctx.lock);
    list_del(&c->list);
    //if (!c->connected)
    	list_del(&c->retry);
    pthread_mutex_unlock(&ipc_ctx.lock);

    //if (c->connected) 
    {
	//printf("call mosquitto_disconnect(), mosq: %p. \n", c->mosq);
    	int rc = mosquitto_disconnect(mosq);
    
        if (rc == MOSQ_ERR_NO_CONN) {
            //fprintf(stderr, "the connection have been lost before, mosq=%p.\n", c->mosq);
        }
    }

    if (c->in_loop) {
        /*
         * mosquitto_loop_stop will be blocked
         */
        mosquitto_loop_stop(mosq, false);
	//printf("after calling mosquitto_loop_stop(), mosq: %p \n", c->mosq);
    }

    //printf("in %s(), destory mosq: %p... \n", __func__, c->mosq);
    mosquitto_destroy(mosq);

    //printf("in %s(), mosq: %p destroyed\n", __func__, c->mosq);

    cb_destroy(c->circle_buf);
    map_delete(c->topic_callbacks);

 end:
    pthread_mutex_destroy(&c->lock);
    c->magic_num = 0xdeaddead;
    free(c);
    ipc_lib_cleanup();
}

int bmser_ipc_subscribe(struct ipc_client *c, const struct ipc_topic *t, ipc_msg_cb_func f)
{
    struct mosquitto *mosq = c->mosq;

    pthread_mutex_lock(&c->lock);
    if (f)
    	map_insert(c->topic_callbacks, t->name, f);
    pthread_mutex_unlock(&c->lock);

    int rc = mosquitto_subscribe(mosq, NULL, t->name, t->qos);
    return map_error_code(rc);
}

int bmser_ipc_publish(struct ipc_client *c, const struct ipc_topic *t, const struct ipc_payload *p, bool retain)
{
    struct mosquitto *mosq = c->mosq;

    int rc = mosquitto_publish(mosq, NULL, t->name, p->len, p->buf, t->qos, retain);
    return map_error_code(rc);
}

int bmser_ipc_get_message(struct ipc_client *c, struct ipc_message **m)
{
    pthread_mutex_lock(&c->lock);

    if (cb_is_empty(c->circle_buf)) {
        pthread_mutex_unlock(&c->lock);
    	return IPC_ERR_EAGAIN;
    }

    struct ipc_message ipc_msg;
    cb_pop_front(c->circle_buf, &ipc_msg);
    pthread_mutex_unlock(&c->lock);

    struct ipc_message *msg_copy = malloc(sizeof(struct ipc_message));
    memcpy(msg_copy, &ipc_msg, sizeof(struct ipc_message));

    *m = msg_copy;

    return IPC_ERR_SUCCESS;
}

void bmser_ipc_release_message(struct ipc_message *m)
{
    free(m->t.name);
    free(m->p.buf);

    free(m);
}