#include "plugin_manager_interna.h"
#include "stellar/session.h"
#include "stellar/utils.h"
#include "toml/toml.h"
#include "uthash/utlist.h"

#include <stdbool.h>

#include "stellar_core.h"
#include "tuple.h"
#include "packet_private.h"
#include "session_private.h"

UT_icd plugin_specs_icd = {sizeof(struct plugin_specific), NULL, NULL, NULL};

static struct plugin_specific *plugin_specs_load(const char *toml_conf_path, int *spec_num)
{
	*spec_num = 0;
    FILE* fp = fopen(toml_conf_path, "r");
	if(fp==NULL)return NULL;
    char errbuf[256];
    toml_table_t* conf = toml_parse_file(fp, errbuf, sizeof(errbuf));
    fclose(fp);	
    if (!conf) {
        fprintf(stderr, "Error parsing toml: %s\n", errbuf);
        return NULL;
    }	
    toml_array_t* plugin_array = toml_array_in(conf, "plugin");
	if(plugin_array==NULL)return NULL;
    *spec_num = toml_array_nelem(plugin_array);
     struct plugin_specific* plugins = CALLOC(struct plugin_specific, *spec_num);

    for (int i = 0; i < *spec_num; i++) {
        toml_table_t* plugin = toml_table_at(plugin_array, i);

        const char *path_raw = toml_raw_in(plugin, "path");
        const char *init_func_name_raw = toml_raw_in(plugin, "init");
        const char *exit_func_name_raw = toml_raw_in(plugin, "exit");
        char *path = NULL;
        char *init_func_name = NULL;
        char *exit_func_name = NULL;
        if (toml_rtos(path_raw, &path) || toml_rtos(init_func_name_raw, &init_func_name) ||
            toml_rtos(exit_func_name_raw, &exit_func_name))
        {
            goto PLUGIN_SPEC_LOAD_ERROR;
        }
        void* handle = dlopen(path, RTLD_NOW|RTLD_LAZY|RTLD_GLOBAL);
        if (!handle) {
            fprintf(stderr, "Error loading plugin %s: %s\n", path, dlerror());
            goto PLUGIN_SPEC_LOAD_ERROR;
        }

        plugins[i].load_cb = (plugin_on_load_func *) dlsym(handle, init_func_name);
        if (!plugins[i].load_cb) {
            fprintf(stderr, "Could not load init function %s: %s\n", init_func_name, dlerror());
        }

        plugins[i].unload_cb = (plugin_on_unload_func *) dlsym(handle, exit_func_name);
        if (!plugins[i].unload_cb) {
            fprintf(stderr, "Could not load exit function %s: %s\n", exit_func_name, dlerror());
        }
        FREE(path);
        FREE(init_func_name);
        FREE(exit_func_name);
    }
    toml_free(conf);
    return plugins;	
PLUGIN_SPEC_LOAD_ERROR:
	toml_free(conf);
	FREE(plugins);
	return NULL;	
}

static struct plugin_manager_per_thread_data *plugin_manager_per_thread_data_new(struct stellar *st)
{
	if(st == NULL)return NULL;
	int thread_num=stellar_get_worker_thread_num(st);
	struct plugin_manager_per_thread_data *per_thread_data = CALLOC(struct plugin_manager_per_thread_data, thread_num);
	return per_thread_data;
}

static void plugin_manager_per_thread_data_free(struct plugin_manager_per_thread_data *per_thread_data, struct stellar *st)
{
	if(per_thread_data == NULL || st == NULL)return;
	int thread_num=stellar_get_worker_thread_num(st);
	struct plugin_manager_per_thread_data *p_data;
	for (int i = 0; i < thread_num; i++)
	{
		p_data=per_thread_data+i;
		if(p_data->per_thread_pkt_exdata_array.exdata_array)FREE(p_data->per_thread_pkt_exdata_array.exdata_array);
	}
	FREE(per_thread_data);
	return;
}

struct plugin_manager_schema *plugin_manager_init(struct stellar *st, const char *plugin_spec_file_path)
{
	int spec_num;
	struct plugin_specific *specs = plugin_specs_load(plugin_spec_file_path, &spec_num);
	if(spec_num < 0)
	{
		return NULL;
	}
	struct plugin_manager_schema *plug_mgr = CALLOC(struct plugin_manager_schema, 1);
	//TODO: set max_message_dispatch as parameter
	plug_mgr->max_message_dispatch=MAX_MSG_PER_DISPATCH;
	if(spec_num > 0)
	{
		utarray_new(plug_mgr->plugin_load_specs_array,&plugin_specs_icd);
		utarray_reserve(plug_mgr->plugin_load_specs_array, spec_num);
	}

	plug_mgr->st = st;
	stellar_set_plugin_manger(st, plug_mgr);

	for(int i = 0; i < spec_num; i++)
	{
		if (specs[i].load_cb != NULL)
		{
			specs[i].plugin_ctx=specs[i].load_cb(st);
			utarray_push_back(plug_mgr->plugin_load_specs_array, &specs[i]);
		}
	}
	FREE(specs);
	plug_mgr->per_thread_data = plugin_manager_per_thread_data_new(st);	
	return plug_mgr;
}

void plugin_manager_exit(struct plugin_manager_schema *plug_mgr)
{
	if(plug_mgr==NULL)return;
	struct plugin_specific *p=NULL;
    if (plug_mgr->plugin_load_specs_array)
    {
        while ((p = (struct plugin_specific *)utarray_next(plug_mgr->plugin_load_specs_array, p)))
        {
            if (p->unload_cb)
                p->unload_cb(p->plugin_ctx);
        }
        utarray_free(plug_mgr->plugin_load_specs_array);
    }
    if(plug_mgr->stellar_mq_schema_array)
	{
		for(unsigned int i = 0; i < utarray_len(plug_mgr->stellar_mq_schema_array); i++)
		{
			stellar_mq_destroy_topic( plug_mgr->st, i);
		}
		utarray_free(plug_mgr->stellar_mq_schema_array);
	}
	if(plug_mgr->stellar_exdata_schema_array)utarray_free(plug_mgr->stellar_exdata_schema_array);
	if(plug_mgr->registered_polling_plugin_array)utarray_free(plug_mgr->registered_polling_plugin_array);
	if(plug_mgr->registered_packet_plugin_array)
	{
        struct registered_packet_plugin_schema *s = NULL;
        while ((s = (struct registered_packet_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, s)))
        {
			if(s->registed_packet_mq_subscriber_info)utarray_free(s->registed_packet_mq_subscriber_info);
        }
        utarray_free(plug_mgr->registered_packet_plugin_array);
	}	
	plugin_manager_per_thread_data_free(plug_mgr->per_thread_data, plug_mgr->st);
	FREE(plug_mgr);
	return;
}

/*******************************
 * STELLAR EXDATA *
 *******************************/

static void stellar_exdata_met_copy(void *_dst, const void *_src)
{
    struct stellar_exdata_schema *dst = (struct stellar_exdata_schema *)_dst, *src = (struct stellar_exdata_schema *)_src;
    dst->free_func = src->free_func;
    dst->free_arg = src->free_arg;
    dst->idx = src->idx;
    dst->name = src->name ? strdup(src->name) : NULL;
}

static void stellar_exdata_met_dtor(void *_elt)
{
    struct stellar_exdata_schema *elt = (struct stellar_exdata_schema *)_elt;
    if (elt->name)
        FREE(elt->name);
}

UT_icd stellar_exdata_meta_icd = {sizeof(struct stellar_exdata_schema), NULL, stellar_exdata_met_copy, stellar_exdata_met_dtor};

int stellar_exdata_new_index(struct stellar *st, const char *name, stellar_exdata_free *free_func,void *free_arg) 
{
	if(st==NULL || name==NULL)return -1;
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr->stellar_exdata_schema_array == NULL)
	{
		utarray_new(plug_mgr->stellar_exdata_schema_array, &stellar_exdata_meta_icd);
	}
	if(plug_mgr->stellar_exdata_schema_array == NULL)return -1;
	unsigned int len = utarray_len(plug_mgr->stellar_exdata_schema_array);
	struct stellar_exdata_schema *t_schema;
	for(unsigned int i = 0; i < len; i++)
	{
		t_schema = (struct stellar_exdata_schema *)utarray_eltptr(plug_mgr->stellar_exdata_schema_array, i);
		if(strcmp(t_schema->name, name) == 0)
		{
			t_schema->free_func=free_func;
			t_schema->free_arg=free_arg;
			return t_schema->idx;
		}
	}
	struct stellar_exdata_schema new_schema;
	memset(&new_schema, 0, sizeof(struct stellar_exdata_schema));
	new_schema.free_func=free_func;
	new_schema.name=(char *)name;
	new_schema.idx=len;
	new_schema.free_arg=free_arg;
	utarray_push_back(plug_mgr->stellar_exdata_schema_array, &new_schema);
	return new_schema.idx;
}

int stellar_exdata_set(UT_array *exdata_schema, struct stellar_exdata *exdata_array, int idx, void *ex_ptr)
{
	if(exdata_schema == NULL|| exdata_array == NULL)return -1;
	unsigned int len=utarray_len(exdata_schema);
	if(len < (unsigned int)idx)return -1;
	if((exdata_array+idx)->state == EXIT)return -1;
	(exdata_array+idx)->exdata=ex_ptr;
	return 0;
}

void *stellar_exdata_get(UT_array *exdata_schema, struct stellar_exdata *exdata_array, int idx)
{
	if(exdata_schema == NULL|| exdata_array == NULL)return NULL;
	unsigned int len = utarray_len(exdata_schema);
	if(len < (unsigned int)idx)return NULL;
	if((exdata_array+idx)->state == EXIT)return NULL;
	return (exdata_array+idx)->exdata;
}

/*******************************
 * PACKET EXDATA *
 *******************************/
static struct stellar_exdata *per_thread_packet_exdata_arrary_get(struct plugin_manager_schema *plug_mgr)
{
	if(plug_mgr==NULL || plug_mgr->stellar_exdata_schema_array == NULL)return NULL;
	int tid=stellar_get_current_thread_index();
	if(plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array == NULL)
	{
		unsigned int len = utarray_len(plug_mgr->stellar_exdata_schema_array);
		plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array = CALLOC(struct stellar_exdata, len);
	}
	return plug_mgr->per_thread_data[tid].per_thread_pkt_exdata_array.exdata_array;	
}

static void per_thread_packet_exdata_arrary_clean(struct plugin_manager_schema *plug_mgr)
{
	if(plug_mgr==NULL || plug_mgr->stellar_exdata_schema_array == NULL)return;
	unsigned int len=utarray_len(plug_mgr->stellar_exdata_schema_array);
	struct stellar_exdata *per_thread_pkt_exdata_arrary = per_thread_packet_exdata_arrary_get(plug_mgr);
	if(per_thread_pkt_exdata_arrary == NULL)return;
	for (unsigned int i = 0; i < len; i++)
	{
		void *exdata = (per_thread_pkt_exdata_arrary + i)->exdata;
		(per_thread_pkt_exdata_arrary + i)->state=EXIT;
		struct stellar_exdata_schema *schema = (struct stellar_exdata_schema *)utarray_eltptr(plug_mgr->stellar_exdata_schema_array, i);
		if (exdata)
		{
			if (schema->free_func)
			{
				schema->free_func(i, exdata, schema->free_arg);
			}
			(per_thread_pkt_exdata_arrary + i)->exdata=NULL;
		}
		(per_thread_pkt_exdata_arrary + i)->state=INIT;
	}
}

int packet_exdata_set(struct packet *pkt, int idx, void *ex_ptr)
{
	if(pkt == NULL)return -1;
	struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)packet_get_user_data(pkt);
	return stellar_exdata_set(plug_mgr->stellar_exdata_schema_array, per_thread_packet_exdata_arrary_get(plug_mgr), idx, ex_ptr);
}

void *packet_exdata_get(struct packet *pkt, int idx)
{
	if(pkt == NULL)return NULL;
	struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)packet_get_user_data(pkt);
	return stellar_exdata_get( plug_mgr->stellar_exdata_schema_array, per_thread_packet_exdata_arrary_get(plug_mgr), idx);
}

/*******************************
 * SESSION EXDATA *
 *******************************/

int session_exdata_set(struct session *sess, int idx, void *ex_ptr)
{
	struct stellar_exdata *sess_exdata_array = (struct stellar_exdata  *)session_get_user_data(sess);
	if(sess_exdata_array == NULL)return -1;
	if(sess_exdata_array->plug_mgr->stellar_exdata_schema_array == NULL)return -1;
	return stellar_exdata_set(sess_exdata_array->plug_mgr->stellar_exdata_schema_array, sess_exdata_array, idx, ex_ptr);
}

void *session_exdata_get(struct session *sess, int idx)
{
	struct stellar_exdata *sess_exdata_array = (struct stellar_exdata  *)session_get_user_data(sess);
	if(sess_exdata_array == NULL)return NULL;
	if(sess_exdata_array->plug_mgr->stellar_exdata_schema_array==NULL)return NULL;
	return stellar_exdata_get(sess_exdata_array->plug_mgr->stellar_exdata_schema_array, sess_exdata_array, idx);
}

/*******************************
 * STELLAR MQ *
 *******************************/
static void stellar_mq_topic_schema_copy(void *_dst, const void *_src)
{
    struct stellar_mq_topic_schema *dst = (struct stellar_mq_topic_schema *)_dst,
                                   *src = (struct stellar_mq_topic_schema *)_src;
    dst->subscribers = src->subscribers;
    dst->free_cb = src->free_cb;
    dst->free_cb_arg = src->free_cb_arg;
    dst->topic_id = src->topic_id;
    dst->subscriber_cnt = src->subscriber_cnt;
    dst->topic_name = src->topic_name ? strdup(src->topic_name) : NULL;
}

static void stellar_mq_topic_schema_dtor(void *_elt)
{
    struct stellar_mq_topic_schema *elt = (struct stellar_mq_topic_schema *)_elt;
    if (elt->topic_name)
        FREE(elt->topic_name);
    // FREE(elt); // free the item
}

UT_icd stellar_mq_topic_schema_icd = {sizeof(struct stellar_mq_topic_schema), NULL, stellar_mq_topic_schema_copy, stellar_mq_topic_schema_dtor};

int stellar_mq_get_topic_id(struct stellar *st, const char *topic_name)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	UT_array *mq_schema_array=plug_mgr->stellar_mq_schema_array;

	if(topic_name == NULL || mq_schema_array == NULL )return -1;
	unsigned int len = utarray_len(mq_schema_array);
	struct stellar_mq_topic_schema *t_schema;
	for(unsigned int i = 0; i < len; i++)
	{
		t_schema = (struct stellar_mq_topic_schema *)utarray_eltptr(mq_schema_array, i);
		if(strcmp(t_schema->topic_name, topic_name) == 0)
		{
			return i;
		}
	}	
	return -1;
}

int stellar_mq_update_topic(struct stellar *st, int topic_id, stellar_msg_free_cb_func *msg_free_cb, void *msg_free_arg)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	UT_array *mq_schema_array=plug_mgr->stellar_mq_schema_array;
	if(mq_schema_array == NULL)return -1;
	unsigned int len = utarray_len(mq_schema_array);
	if(len < (unsigned int)topic_id)return -1;
	struct stellar_mq_topic_schema *t_schema = (struct stellar_mq_topic_schema *)utarray_eltptr(mq_schema_array, (unsigned int)topic_id);
	if(t_schema == NULL)return -1;
	t_schema->free_cb=msg_free_cb;
	t_schema->free_cb_arg=msg_free_arg;
	return 0;
}

int stellar_mq_create_topic(struct stellar *st, const char *topic_name, stellar_msg_free_cb_func *msg_free_cb, void *msg_free_arg)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr->stellar_mq_schema_array == NULL)
	{
		utarray_new(plug_mgr->stellar_mq_schema_array, &stellar_mq_topic_schema_icd);
	}
	unsigned int len = utarray_len(plug_mgr->stellar_mq_schema_array);
	if(stellar_mq_get_topic_id(st, topic_name) >= 0)
	{
		return -1;
	}
	struct stellar_mq_topic_schema t_schema;
	memset(&t_schema, 0, sizeof(struct stellar_mq_topic_schema));
	t_schema.free_cb=msg_free_cb;
	t_schema.topic_name=(char *)topic_name;
	t_schema.topic_id=len;//topid_id equals arrary index
	t_schema.free_cb_arg=msg_free_arg;
	t_schema.subscribers=NULL;
	t_schema.subscriber_cnt=0;
	utarray_push_back(plug_mgr->stellar_mq_schema_array, &t_schema);
	plug_mgr->stellar_mq_topic_num+=1;
	return t_schema.topic_id;	
}

int stellar_mq_destroy_topic(struct stellar *st, int topic_id)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr->stellar_mq_schema_array==NULL)return -1;
    unsigned int len = utarray_len(plug_mgr->stellar_mq_schema_array);
    if (len <= (unsigned int)topic_id)
        return -1;
    struct stellar_mq_topic_schema *topic =
        (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->stellar_mq_schema_array, (unsigned int)topic_id);
    struct stellar_mq_subscriber *sub_elt, *sub_tmp;

	if(topic == NULL)return -1;

    if (topic->is_destroyed == 1)return 0;

    DL_FOREACH_SAFE(topic->subscribers, sub_elt, sub_tmp)
    {
        DL_DELETE(topic->subscribers, sub_elt);
        FREE(sub_elt);
    }
    topic->is_destroyed = 1;
	plug_mgr->stellar_mq_topic_num-=1;
    return 1; // success
}

UT_icd stellar_mq_subscriber_info_icd = {sizeof(struct stellar_mq_subscriber_info), NULL, NULL, NULL};

static int __stellar_mq_subscribe(struct plugin_manager_schema *plug_mgr, int topic_id, void *plugin_on_msg_cb, int plugin_idx, UT_array *registed_mq_subscriber_info)
{
	if(plug_mgr == NULL || plug_mgr->stellar_mq_schema_array==NULL || registed_mq_subscriber_info == NULL)return -1;

    unsigned int len = utarray_len(plug_mgr->stellar_mq_schema_array);
    if (len <= (unsigned int)topic_id)return -1;

    struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->stellar_mq_schema_array, (unsigned int)topic_id);
	if(topic==NULL)return -1;

	// if plugin already subscribe current topic, return 0
	struct stellar_mq_subscriber_info *p=NULL;
  	while( (p=(struct stellar_mq_subscriber_info *)utarray_next(registed_mq_subscriber_info,p))) 
	{
		if(p->topic_id==topic_id)
		{
			struct stellar_mq_subscriber *tmp_subscriber=topic->subscribers;
			int cnt=0;
			while(tmp_subscriber)
			{
				if(cnt==p->subscriber_idx)
				{
					tmp_subscriber->plugin_msg_cb=plugin_on_msg_cb;
					return 0;
				}
				cnt++;
				tmp_subscriber=tmp_subscriber->next;
			}
		}
	};

	struct stellar_mq_subscriber *new_subscriber = CALLOC(struct stellar_mq_subscriber,1);
	new_subscriber->topic_subscriber_idx = topic->subscriber_cnt;
	new_subscriber->plugin_idx = plugin_idx;
    new_subscriber->plugin_msg_cb = plugin_on_msg_cb;
	DL_APPEND(topic->subscribers, new_subscriber);

	struct stellar_mq_subscriber_info sub_info;
	memset(&sub_info, 0, sizeof(struct stellar_mq_subscriber_info));
	sub_info.topic_id=topic_id;
	sub_info.subscriber_idx=topic->subscriber_cnt;	
	utarray_push_back(registed_mq_subscriber_info, &sub_info);
	topic->subscriber_cnt+=1;
	plug_mgr->mq_topic_subscriber_num+=1;
	return 0;
}

static void stellar_mq_dispatch_one_message(struct stellar_message *mq_elt)
{
	struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)stellar_get_plugin_manager(mq_elt->st);
    struct stellar_mq_subscriber *sub_elt, *sub_tmp;
	struct registered_packet_plugin_schema *packet_plugin_schema;
    struct stellar_mq_topic_schema *topic = (struct stellar_mq_topic_schema *)utarray_eltptr(plug_mgr->stellar_mq_schema_array, 
																							(unsigned int)(mq_elt->header.topic_id));
    if (topic)
    {
        DL_FOREACH_SAFE(topic->subscribers, sub_elt, sub_tmp)
        {
            if (sub_elt->plugin_msg_cb)
            {
                packet_plugin_schema = (struct registered_packet_plugin_schema *)utarray_eltptr(
                    plug_mgr->registered_packet_plugin_array, (unsigned int)sub_elt->plugin_idx);
                if (packet_plugin_schema)
                {
                    sub_elt->plugin_msg_cb(mq_elt->header.topic_id, mq_elt->body, packet_plugin_schema->plugin_env);
                }
            }
        }
    }
}

static void stellar_mq_dispatch(struct stellar_message *priority_mq[], struct stellar_message ** dealth_letter_queue)
{
	struct stellar_message *mq_elt=NULL, *mq_tmp=NULL;
	int cur_priority = STELLAR_MQ_PRIORITY_HIGH;
	while(cur_priority >= STELLAR_MQ_PRIORITY_LOW)
	{
		if(priority_mq[cur_priority]==NULL)
		{
			cur_priority--;
			continue;
		}
        DL_FOREACH_SAFE(priority_mq[cur_priority], mq_elt, mq_tmp)
        {
			stellar_mq_dispatch_one_message(mq_elt);
			DL_DELETE(priority_mq[mq_elt->header.priority], mq_elt);
    		DL_APPEND(*dealth_letter_queue, mq_elt); // move to dlq list

			cur_priority=STELLAR_MQ_PRIORITY_HIGH;
			break;
        }		
	}
	return;
}

static void stellar_mq_free(struct stellar_message **head, UT_array *mq_schema_array)
{
	struct stellar_message *mq_elt, *tmp;
	struct stellar_mq_topic_schema *topic;
    DL_FOREACH_SAFE(*head, mq_elt, tmp)
    {
		topic = (struct stellar_mq_topic_schema *)utarray_eltptr(mq_schema_array,
                                                                    (unsigned int)(mq_elt->header.topic_id));
        if (topic && topic->free_cb)
        {
			topic->free_cb(mq_elt->body, topic->free_cb_arg);
        }
        DL_DELETE(*head, mq_elt);
        FREE(mq_elt);
    }
}

/*******************************
 * PACKET MQ *
 *******************************/

//return 0 if success, otherwise return -1.
int stellar_mq_subscribe(struct stellar *st, int topic_id, on_msg_cb_func *plugin_on_msg_cb, int plugin_id)
{
	int plugin_idx=plugin_id;

	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr == NULL || plug_mgr->registered_packet_plugin_array == NULL)return -1;

	struct registered_packet_plugin_schema *packet_plugin_schema = (struct registered_packet_plugin_schema *)utarray_eltptr(plug_mgr->registered_packet_plugin_array, (unsigned)plugin_idx);
	if(packet_plugin_schema==NULL)return -1;

	if(packet_plugin_schema->registed_packet_mq_subscriber_info==NULL)
	{
		utarray_new(packet_plugin_schema->registed_packet_mq_subscriber_info, &stellar_mq_subscriber_info_icd);
	}

	return __stellar_mq_subscribe(plug_mgr,topic_id, (void *)plugin_on_msg_cb, plugin_idx, packet_plugin_schema->registed_packet_mq_subscriber_info);
}

int stellar_mq_publish_message_with_priority(struct stellar *st, int topic_id, void *data, enum stellar_mq_priority priority)
{
	if(st==NULL)return -1;
	struct plugin_manager_schema *plug_mgr = (struct plugin_manager_schema *)stellar_get_plugin_manager(st);
	if(plug_mgr==NULL || plug_mgr->stellar_mq_schema_array == NULL)return -1;

	int tid = stellar_get_current_thread_index();
	if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt == -1)return -1;
	if(plug_mgr->per_thread_data[tid].pub_packet_msg_cnt >= plug_mgr->max_message_dispatch)return -1;

    unsigned int len = utarray_len(plug_mgr->stellar_mq_schema_array);
    if (len <= (unsigned int)topic_id)return -1;
	struct stellar_message *msg= CALLOC(struct stellar_message,1);
	msg->st=plug_mgr->st;
	msg->header.topic_id = topic_id;
	msg->header.priority = priority;
	msg->body = data;
	DL_APPEND(plug_mgr->per_thread_data[tid].priority_mq[priority], msg);
	plug_mgr->per_thread_data[tid].pub_packet_msg_cnt+=1;
	return 0;	
}

int stellar_mq_publish_message(struct stellar *st, int topic_id, void *data)
{
	return stellar_mq_publish_message_with_priority(st, topic_id, data, STELLAR_MQ_PRIORITY_NORMAL);
}

/*******************************
 * PLUGIN MANAGER SESSION RUNTIME *
 *******************************/
struct stellar_exdata *session_exdata_runtime_new(struct plugin_manager_schema *plug_mgr)
{
	struct stellar_exdata *exdata_rt = NULL;
	if(plug_mgr->stellar_exdata_schema_array==NULL)return NULL;
	unsigned int len = utarray_len(plug_mgr->stellar_exdata_schema_array);
	if(len > 0)
	{
		exdata_rt=CALLOC(struct stellar_exdata, len);
		exdata_rt->plug_mgr=plug_mgr; // TODO: temporarily set plug_mgr in exdata[0]
	}
	return exdata_rt;
}

void session_exdata_runtime_free(struct stellar_exdata *exdata_rt)
{
	if(exdata_rt==NULL || exdata_rt->plug_mgr == NULL)return;
	if(exdata_rt->plug_mgr->stellar_exdata_schema_array==NULL)return;
	unsigned int len=utarray_len(exdata_rt->plug_mgr->stellar_exdata_schema_array);
	for (unsigned int i = 0; i < len; i++)
	{
		void *exdata = (exdata_rt + i)->exdata;
		(exdata_rt + i)->state=EXIT;
		struct stellar_exdata_schema *schema = (struct stellar_exdata_schema *)utarray_eltptr(exdata_rt->plug_mgr->stellar_exdata_schema_array, i);
		if (exdata)
		{
			if (schema->free_func)
			{
				schema->free_func(i, exdata, schema->free_arg);
			}
		}
	}
	FREE(exdata_rt);
}


/*********************************************
 * PLUGIN MANAGER PACKET PLUGIN *
 *********************************************/
UT_icd registered_packet_plugin_array_icd = {sizeof(struct registered_packet_plugin_schema), NULL, NULL, NULL};

int stellar_plugin_register(struct stellar *st, unsigned char ip_proto, plugin_on_packet_func on_packet_input, plugin_on_packet_func on_packet_output, void *plugin_env)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr->registered_packet_plugin_array == NULL)
	{
		utarray_new(plug_mgr->registered_packet_plugin_array, &registered_packet_plugin_array_icd);
	}
	struct registered_packet_plugin_schema packet_plugin_schema;
	memset(&packet_plugin_schema, 0, sizeof(packet_plugin_schema));
	packet_plugin_schema.ip_protocol = ip_proto;
	packet_plugin_schema.on_packet[PACKET_STAGE_INPUT] = on_packet_input;
	packet_plugin_schema.on_packet[PACKET_STAGE_OUTPUT] = on_packet_output;
	packet_plugin_schema.plugin_env = plugin_env;
	utarray_push_back(plug_mgr->registered_packet_plugin_array, &packet_plugin_schema);
	return (utarray_len(plug_mgr->registered_packet_plugin_array)-1);// return packet plugin_id, equals to packet plugin arrary index 
}

static void plugin_manager_on_packet(struct plugin_manager_schema *plug_mgr, struct packet *pkt, enum packet_stage in_out)
{
	if(plug_mgr==NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
	struct registered_packet_plugin_schema *p=NULL;

	//TODO: get  innermost layer ip protocol by packet api
	struct tuple6 t6;
	packet_get_innermost_tuple6(pkt, &t6);
	unsigned char ip_proto=t6.ip_proto;
	
	int tid=stellar_get_current_thread_index();
	//TODO : provide public api to reset pub_msg_cnt
	plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=0;//reset pub_msg_cnt
	while ((p = (struct registered_packet_plugin_schema *)utarray_next(plug_mgr->registered_packet_plugin_array, p)))
	{
		if(p->ip_protocol == ip_proto && p->on_packet[in_out])
		{
			p->on_packet[in_out](pkt, ip_proto, p->plugin_env);
		}
	}
	stellar_mq_dispatch(plug_mgr->per_thread_data[tid].priority_mq, &plug_mgr->per_thread_data[tid].dealth_letter_queue);
	return;
}

void plugin_manager_on_packet_input(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
{
	plugin_manager_on_packet(plug_mgr, pkt, PACKET_STAGE_INPUT);
}

void plugin_manager_on_packet_output(struct plugin_manager_schema *plug_mgr, struct packet *pkt)
{
	if(plug_mgr == NULL || plug_mgr->registered_packet_plugin_array == NULL || pkt == NULL)return;
	plugin_manager_on_packet(plug_mgr, pkt, PACKET_STAGE_OUTPUT);
	int tid=stellar_get_current_thread_index();
	plug_mgr->per_thread_data[tid].pub_packet_msg_cnt=-1;//disable packet message publish
	stellar_mq_free(&plug_mgr->per_thread_data[tid].dealth_letter_queue, 
					plug_mgr->stellar_mq_schema_array);
	per_thread_packet_exdata_arrary_clean(plug_mgr);
}

/*********************************************
 * PLUGIN MANAGER POLLING PLUGIN *
 *********************************************/
UT_icd registered_polling_plugin_array_icd = {sizeof(struct registered_polling_plugin_schema), NULL, NULL, NULL};

int stellar_polling_plugin_register(struct stellar *st, plugin_on_polling_func on_polling, void *plugin_env)
{
	struct plugin_manager_schema *plug_mgr = stellar_get_plugin_manager(st);
	if(plug_mgr->registered_polling_plugin_array == NULL)
	{
		utarray_new(plug_mgr->registered_polling_plugin_array, &registered_polling_plugin_array_icd);
	}
	struct registered_polling_plugin_schema polling_plugin_schema;
	memset(&polling_plugin_schema, 0, sizeof(polling_plugin_schema));
	polling_plugin_schema.on_polling = on_polling;
	polling_plugin_schema.plugin_env = plugin_env;
	utarray_push_back(plug_mgr->registered_polling_plugin_array, &polling_plugin_schema);
	return (utarray_len(plug_mgr->registered_polling_plugin_array)-1);// return polling plugin_id, equals to polling plugin arrary index + POLLING_PULGIN_ID_BASE
}

int plugin_manager_on_polling(struct plugin_manager_schema *plug_mgr)
{
	if(plug_mgr==NULL || plug_mgr->registered_polling_plugin_array == NULL)return 0;
	struct registered_polling_plugin_schema *p=NULL;
	int polling_state=0;
	while ((p = (struct registered_polling_plugin_schema *)utarray_next(plug_mgr->registered_polling_plugin_array, p)))
	{
		if(p->on_polling)
		{
			if(p->on_polling(p->plugin_env)==1)
			{
				polling_state=1;
			}
		}
	}
	return polling_state;
}