myvtm/spbase/sp_var.c

#include "precompile.h"

#include "sp_var.h"
#include "sp_def.h"
#include "sp_svc.h"
#include "sp_dbg_export.h"
#include "sp_iom.h"
#include "sp_cfg.h"
#include "sp_env.h"
#include "sp_rsn.h"

#include "memutil.h"
#include "strutil.h"
#include "list.h"
#include "array.h"
#include "refcnt.h"
#include "iniutil.h"

#define VAR_CMD_CHANGE		0x01
#define VAR_CMD_SUBSCRIBE	0x02
#define VAR_CMD_UNSUBSCRIBE	0x03

struct sp_var_client_t
{
	sp_svc_t *svc;
};

static int sp_var_can_write(sp_var_client_t *client, const char *key, sp_cfg_shell_sysevent_t **p_sysevent)
{
	sp_iom_t *iom = sp_svc_get_iom(client->svc);
	sp_cfg_t *cfg = sp_get_env(iom)->cfg;
	sp_cfg_shell_sysevent_t *sysevent;

	sysevent = sp_cfg_get_sysevent(cfg, key);
	if (sysevent) {
		int i;
		*p_sysevent = sysevent;
		for (i = 0; i < sysevent->arr_owner_entity->nelts; ++i) {
			sp_cfg_shell_entity_t *ent = ARRAY_IDX(sysevent->arr_owner_entity, i, sp_cfg_shell_entity_t*);
			if (ent->idx == sp_svc_get_id(client->svc))
				return TRUE;
		}
	} else {
		*p_sysevent = NULL;
	}

	return FALSE;
}

int sp_var_client_create(sp_svc_t *svc, sp_var_client_t **p_client)
{
	sp_var_client_t *client = MALLOC_T(sp_var_client_t);
	client->svc = svc;
	*p_client = client;
	return 0;
}

void sp_var_client_destroy(sp_var_client_t *client)
{
	free(client);
}

// str == NULL for delete
int sp_var_client_set(sp_var_client_t *client, const char *key, const char *str, int persist)
{
	sp_iom_t *iom = sp_svc_get_iom(client->svc);
	sp_cfg_t *cfg = sp_get_env(iom)->cfg;
	int svc_id = sp_svc_get_id(client->svc);
	int rc;
	sp_cfg_shell_sysevent_t *sysevent;
	char old_value[SP_CFG_MAX_SYSEVT_BUF];

	if (!key ) {
		sp_dbg_warn("%s invalid argument", __FUNCTION__);
		return Error_Null;
	}

	if (str && strlen(str) >= SP_CFG_MAX_SYSEVT_BUF) {
		sp_dbg_warn("%s has exceed max sysevent length!", key);
		return Error_Param;
	}

	if (!sp_var_can_write(client, key, &sysevent)) {
		sp_dbg_warn("%s has no write privilege!", __FUNCTION__);
		return Error_NoPrivilege;
	}

	// same value, not trigger change event
	if ((str != NULL && strcmp(str, sysevent->init_value) == 0)
		|| (str == NULL && sysevent->init_value[0] == 0))
		return Error_Succeed;
	
	sp_cfg_lock(cfg);
	if (str) {
		strcpy(old_value, sysevent->init_value);
		strcpy(sysevent->init_value, str);

		if (persist)
			inifile_format_write(cfg->shellvar_ini_path != NULL ? cfg->shellvar_ini_path : cfg->shell_ini_path,
				"SysEvent", key, "%s,\"%s\"", sp_cfg_get_entity_by_idx(cfg, svc_id)->name, str);
	} else {
		sysevent->init_value[0] = 0;

		if (persist)
			inifile_format_write(cfg->shellvar_ini_path != NULL ? cfg->shellvar_ini_path : cfg->shell_ini_path,
				"SysEvent", key, "%s,\"\"", sp_cfg_get_entity_by_idx(cfg, svc_id)->name);
	}
	sp_cfg_unlock(cfg);

	{
		iobuffer_t *pkt = iobuffer_create(-1, -1);
		iobuffer_write(pkt, IOBUF_T_I4, &svc_id, 0);
		iobuffer_write(pkt, IOBUF_T_STR, key, -1);
		iobuffer_write(pkt, IOBUF_T_STR, old_value, -1);
		iobuffer_write(pkt, IOBUF_T_STR, str, -1);
		rc = sp_svc_post(client->svc, SP_SHELL_MOD_ID, SP_SHELL_SVC_ID, SP_PKT_VAR|VAR_CMD_CHANGE, 0, &pkt);
		if (rc != 0) {
			sp_dbg_warn("%s post SP_PKT_VAR|VAR_CMD_CHANGE failed!", __FUNCTION__);
		}
		if (pkt)
			iobuffer_dec_ref(pkt);
	}

	return rc;
}

int sp_var_client_get(sp_var_client_t *client, const char *key, char *str, int *n)
{
	sp_iom_t *iom = sp_svc_get_iom(client->svc);
	sp_cfg_t *cfg = sp_get_env(iom)->cfg;
	int rc = 0;
	sp_cfg_shell_sysevent_t *sysevent;

	sysevent = sp_cfg_get_sysevent(cfg, key);

	if (!sysevent) {
		sp_dbg_warn("%s cannot find sysevent !", key);
		return Error_NotExist;
	}
	
	if (str) {
		int slen;		
		sp_cfg_lock(cfg);
		slen = (int)strlen(sysevent->init_value);
		if (*n > slen) {
			strcpy(str, sysevent->init_value);
		} else {
			rc = Error_Bug;
		}
		sp_cfg_unlock(cfg);
	} else {
		if (n) {
			*n = SP_CFG_MAX_SYSEVT_BUF;
		} else {
			rc = Error_Bug;
		}
	}

	return rc;
}

int sp_var_client_lock(sp_var_client_t *client)
{
	sp_iom_t *iom = sp_svc_get_iom(client->svc);
	sp_cfg_t *cfg = sp_get_env(iom)->cfg;
	sp_cfg_lock(cfg);
	return 0;
}

int sp_var_client_unlock(sp_var_client_t *client)
{
	sp_iom_t *iom = sp_svc_get_iom(client->svc);
	sp_cfg_t *cfg = sp_get_env(iom)->cfg;
	sp_cfg_unlock(cfg);
	return 0;
}

struct sp_var_listener_t
{
	int client_id;
	int subcribed;
	int enabled;
	char *key;
	strand_t *strand;
	sp_var_on_change on_change;
	void *user_data;
	void *tag;
	sp_svc_t *svc;
	DECLARE_REF_COUNT_MEMBER(ref_cnt);
};

DECLARE_REF_COUNT_STATIC(sp_var_listener, sp_var_listener_t)

static void listener_on_pkt_threadpool(threadpool_t *threadpool, void *arg)
{
	iobuffer_t *pkt = (iobuffer_t*)arg;
	sp_var_listener_t *listener;
	sp_svc_t *svc;
	int epid;
	int pkt_type;
	int svc_id;
	int pkt_id;
	sp_uid_t rsn;
	sp_rsn_context_t rsn_ctx;

	iobuffer_read(pkt, IOBUF_T_I4, &listener, NULL);
	iobuffer_read(pkt, IOBUF_T_I4, &epid, NULL);
	iobuffer_read(pkt, IOBUF_T_I4, &svc_id, NULL);
	iobuffer_read(pkt, IOBUF_T_I4, &pkt_type, NULL);
	iobuffer_read(pkt, IOBUF_T_I4, &pkt_id, NULL);

	svc = listener->svc;

	rsn = sp_svc_new_runserial(svc);
	sp_rsn_context_init_original(rsn, SP_ORIGINAL_T_CALLBACK, &rsn_ctx);
	sp_svc_push_runserial_context(svc, &rsn_ctx);
	
	if (listener->subcribed && listener->enabled) {
		int var_client_id = pkt_id;
		int var_cmd = SP_GET_TYPE(pkt_type);
		if (var_cmd == VAR_CMD_CHANGE && listener->subcribed) {
			int from_client_id;
			char *key = NULL;
			iobuffer_format_read(pkt, "4s", &from_client_id, &key);
			if (listener->key[0] == '*' || _stricmp(key, listener->key) == 0) {
				char *old_value = NULL;
				char *new_value = NULL;
				iobuffer_format_read(pkt, "ss", &old_value, &new_value);
				listener->on_change(listener, key, old_value, new_value, from_client_id, listener->user_data);
				if (old_value)
					toolkit_free(old_value);
				if (new_value)
					toolkit_free(new_value);
			}
			if (key)
				toolkit_free(key);
		}
	}

	if (pkt)
		iobuffer_dec_ref(pkt);

	sp_var_listener_dec_ref(listener); //@

	sp_svc_pop_runserial_context(svc);
}

static int listener_on_pkt(sp_svc_t *svc,int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt, void *user_data)
{
	sp_var_listener_t *listener = (sp_var_listener_t *)user_data;
	int var_cmd = SP_GET_TYPE(pkt_type);
	if (var_cmd == VAR_CMD_CHANGE && listener->subcribed && listener->enabled) {
		iobuffer_t *pkt = iobuffer_clone(*p_pkt);
		iobuffer_write_head(pkt, IOBUF_T_I4, &pkt_id, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &pkt_type, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &svc_id, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &epid, 0);
		iobuffer_write_head(pkt, IOBUF_T_I4, &listener, 0);
		sp_var_listener_inc_ref(listener);//@
		if (threadpool_queue_workitem(sp_svc_get_threadpool(svc), listener->strand, &listener_on_pkt_threadpool, pkt) < 0) {
			sp_var_listener_dec_ref(listener);//@
			iobuffer_dec_ref(pkt);
		}
	}
	return TRUE; // continue 
}

int sp_var_listener_create(sp_svc_t *svc, const char *key, sp_var_on_change on_change, void *user_data, sp_var_listener_t **p_listener)
{
	sp_var_listener_t *listener = MALLOC_T(sp_var_listener_t);
	listener->on_change = on_change;
	listener->user_data = user_data;
	listener->subcribed = 0;
	listener->enabled = 1;
	listener->strand = strand_create();
	listener->key = _strdup(key);
	listener->svc = svc;
	*p_listener = listener;
	return 0;
}

static void __sp_var_listener_destroy(sp_var_listener_t *listener)
{
	strand_destroy(listener->strand);
	free(listener->key);
	free(listener);
}
IMPLEMENT_REF_COUNT_MT_STATIC(sp_var_listener, sp_var_listener_t, ref_cnt, __sp_var_listener_destroy)

void sp_var_listener_destroy(sp_var_listener_t *listener)
{
	sp_var_listener_dec_ref(listener);
}

int sp_var_listener_subscribe(sp_var_listener_t *listener)
{
	iobuffer_t *pkt;
	int rc;

	if (listener->subcribed)
		return Error_Duplication;

	pkt = iobuffer_create(-1, -1);

	sp_svc_add_pkt_handler(listener->svc, (int)listener, SP_PKT_VAR, &listener_on_pkt, listener);

	iobuffer_write(pkt, IOBUF_T_STR, listener->key, -1);
	rc = sp_svc_post(listener->svc, SP_SHELL_MOD_ID, SP_SHELL_SVC_ID, 
		SP_PKT_VAR|VAR_CMD_SUBSCRIBE, 0, &pkt);
	if (rc == 0) {
		listener->subcribed = 1;		
	}
	if (pkt)
		iobuffer_dec_ref(pkt);

	return rc;
}

int sp_var_listener_unsubscribe(sp_var_listener_t *listener)
{
	int rc;	

	if (listener->subcribed) {
		iobuffer_t *pkt = iobuffer_create(-1, -1);
		iobuffer_write(pkt, IOBUF_T_STR, listener->key, -1);
		rc = sp_svc_post(listener->svc, SP_SHELL_MOD_ID, SP_SHELL_SVC_ID, 
			SP_PKT_VAR|VAR_CMD_UNSUBSCRIBE, 0, &pkt);
		if (rc == 0) {
			listener->subcribed = 0;
			sp_svc_remove_pkt_handler(listener->svc, (int)listener, SP_PKT_VAR);
		}
		if (pkt)
			iobuffer_dec_ref(pkt);
	} else {
		rc = Error_NotInit;
	}

	return rc;
}

void sp_var_listener_set_tag(sp_var_listener_t *listener, void *tag)
{
	listener->tag = tag;
}

void* sp_var_listener_get_tag(sp_var_listener_t *listener)
{
	return listener->tag;
}

int sp_var_listener_enable(sp_var_listener_t *listener, int enabled)
{
	int old = listener->enabled;
	listener->enabled = enabled;
	return old;
}

typedef struct var_listener_entry {
	int epid;
	int svc_id;
	char *key;
	int instance;
}var_listener_entry;

struct sp_var_daemon_t
{
	CRITICAL_SECTION lock;
	array_header_t *arr_listener;
	sp_svc_t *svc;
};

static void daemon_lock(sp_var_daemon_t *daemon)
{
	EnterCriticalSection(&daemon->lock);
}

static void daemon_unlock(sp_var_daemon_t *daemon)
{
	LeaveCriticalSection(&daemon->lock);
}

static int daemon_on_pkt(sp_svc_t *svc,int epid, int svc_id, int pkt_type, int pkt_id, iobuffer_t **p_pkt, void *user_data)
{
	sp_var_daemon_t *daemon = (sp_var_daemon_t *)user_data;
	int var_cmd = SP_GET_TYPE(pkt_type);

	if (var_cmd == VAR_CMD_CHANGE) {
		iobuffer_t *pkt = *p_pkt;
		int i;
		int client_id;
		int read_state = iobuffer_get_read_state(pkt);
		int write_state = iobuffer_get_write_state(pkt);
		char *key = NULL;
		iobuffer_format_read(pkt, "4s", &client_id, &key);
		iobuffer_restore_read_state(pkt, read_state);
		iobuffer_restore_write_state(pkt, write_state);
		daemon_lock(daemon);
		for (i = 0; i < daemon->arr_listener->nelts; ++i) {
			var_listener_entry* tmp = ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*);
			if (_stricmp(tmp->key, "*")==0 || _stricmp(key, tmp->key) == 0) {
				iobuffer_t *copy_pkt;
				if (i == daemon->arr_listener->nelts-1) {
					copy_pkt = pkt;
					pkt =NULL;
				} else {
					copy_pkt = iobuffer_clone(pkt);
				}
				sp_svc_post(daemon->svc, tmp->epid, tmp->svc_id, SP_PKT_VAR|VAR_CMD_CHANGE, 0, &copy_pkt);
				if (copy_pkt)
					iobuffer_dec_ref(copy_pkt);
			}
		}
		daemon_unlock(daemon);
		toolkit_free(key);
		if (pkt)
			iobuffer_dec_ref(pkt);
		*p_pkt = NULL;
	} else if (var_cmd == VAR_CMD_SUBSCRIBE) {
		char *key = NULL;
		int i;
		int slen;
		iobuffer_read(*p_pkt, IOBUF_T_STR, NULL, &slen);
		key = malloc(slen+1);
		iobuffer_read(*p_pkt, IOBUF_T_STR, key, NULL);
		daemon_lock(daemon);
		for (i = 0; i < daemon->arr_listener->nelts; ++i) {
			var_listener_entry* tmp = ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*);
			if (tmp->epid == epid && tmp->svc_id == svc_id && _stricmp(key, tmp->key) == 0) {
				tmp->instance++;
				break;
			}
		}
		if (i == daemon->arr_listener->nelts) {
			var_listener_entry* tmp = MALLOC_T(var_listener_entry);
			tmp->epid = epid;
			tmp->svc_id = svc_id;
			tmp->key = key;
			key = NULL;
			tmp->instance = 1;
			ARRAY_PUSH(daemon->arr_listener, var_listener_entry*) = tmp;
		}
		daemon_unlock(daemon);
		if (key)
			free(key);
	} else if (var_cmd == VAR_CMD_UNSUBSCRIBE) {
		char *key = NULL;
		int i;
		int slen;
		iobuffer_read(*p_pkt, IOBUF_T_STR, NULL, &slen);
		key = malloc(slen+1);
		iobuffer_read(*p_pkt, IOBUF_T_STR, key, NULL);
		daemon_lock(daemon);
		for (i = 0; i < daemon->arr_listener->nelts; ++i) {
			var_listener_entry* tmp = ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*);
			if (tmp->epid == epid && tmp->svc_id == svc_id && _stricmp(key, tmp->key) == 0) {
				tmp->instance--;
				if (tmp->instance == 0) {
					if (i < daemon->arr_listener->nelts-1)
						ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*) = ARRAY_IDX(daemon->arr_listener, daemon->arr_listener->nelts-1, var_listener_entry*);
					array_pop(daemon->arr_listener);
					free(tmp->key);
					free(tmp);
				}
				break;
			}
		}
		daemon_unlock(daemon);
		if (key)
			free(key);
	} else {
		assert(0);
	}

	return TRUE;
}

static void daemon_on_sys(sp_svc_t *svc,int epid, int state, void *user_data)
{
	sp_var_daemon_t *daemon = (sp_var_daemon_t *)user_data;

	if (state == BUS_STATE_OFF) {
		int i = 0;
		daemon_lock(daemon);
		while (i < daemon->arr_listener->nelts) {
			var_listener_entry* tmp = ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*);
			if (tmp->epid == epid) {
				if (i < daemon->arr_listener->nelts-1)
					ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*) = ARRAY_IDX(daemon->arr_listener, daemon->arr_listener->nelts-1, var_listener_entry*);
				array_pop(daemon->arr_listener);
				free(tmp->key);
				free(tmp);
			} else {
				++i;
			}
		}
		daemon_unlock(daemon);
	}
}

int sp_var_daemon_create(sp_svc_t *svc, sp_var_daemon_t **p_daemon)
{
	sp_var_daemon_t *daemon = ZALLOC_T(sp_var_daemon_t);

	daemon->svc = svc;
	daemon->arr_listener = array_make(3, sizeof(var_listener_entry*));
	InitializeCriticalSection(&daemon->lock);
	sp_svc_add_pkt_handler(svc, (int)daemon, SP_PKT_VAR, &daemon_on_pkt, daemon);
	sp_svc_add_sys_handler(svc, (int)daemon, &daemon_on_sys, daemon);
	*p_daemon = daemon;

	return 0;
}

void sp_var_daemon_destroy(sp_var_daemon_t *daemon)
{
	int i;

	for (i = 0; i < daemon->arr_listener->nelts; ++i) {
		var_listener_entry *tmp = ARRAY_IDX(daemon->arr_listener, i, var_listener_entry*);
		free(tmp->key);
		free(tmp);
	}
	array_free(daemon->arr_listener);
	DeleteCriticalSection(&daemon->lock);
	sp_svc_remove_pkt_handler(daemon->svc, (int)daemon, SP_PKT_VAR);
	sp_svc_remove_sys_handler(daemon->svc, (int)daemon);
	free(daemon);
}