Imported Upstream version 2.1.8.20120216upstream/2.1.8.20120216

1 files changed, 1490 insertions, 0 deletions

diff --git a/dict/pdictclient.c b/dict/pdictclient.c
new file mode 100644
index 0000000..987d6bc
--- /dev/null
+++ b/dict/pdictclient.c
@@ -0,0 +1,1490 @@
+#include "../stdafx.h"
+#include <ctype.h>
+#include <stdlib.h>
+#include <stdio.h>
+#ifndef _MSC_EXTENSIONS
+#include <unistd.h>
+#endif
+#include <assert.h>
+#if !defined(_WINDOWS) || defined(_CYGWIN)
+#include <pthread.h>
+#include <signal.h>
+#else
+#include "wincompat.h"
+#endif
+#include "pdictclient.h"
+#include "ptree.h"
+#include "utils.h"
+#include "md5.h"
+
+/*
+ * TODO XXX:
+ * - notification arguments and reason
+ */
+
+#ifdef DEBUG_PROTOCOL
+int debug_protocol = 0;
+#endif
+
+#define ABORT() ((*(int *)0 = 1), abort())
+#define PENDING_PATTERN "report 200-lid([0-9]*) is pending, key (.*)" \
+    " latest value \".*\" \\((.*)\\)"
+
+extern const char *ws_protocol_ver;
+
+typedef struct result {
+	int r_tag;
+	void (*r_notify)(pdc_session_t *, struct result *, int code,
+	    int final, const char *line);
+	void *r_arg;
+} result_t;
+
+struct pdc_session {
+	int pdcs_rfd;
+	int pdcs_wfd;
+	int (*pdcs_read)(int, void *, unsigned int, char *, int);
+	int (*pdcs_write)(int, const void *, unsigned int, char *, int);
+	int (*pdcs_close)(int, char *, int);
+	void (*pdcs_cleanup)(void *);
+	void *pdcs_cleanup_ptr;
+	ptree_node_t *pdcs_listeners;
+	char pdcs_readbuf[2048];
+	int pdcs_bufcur;
+	int pdcs_buflen;
+	pthread_mutex_t pdcs_listeners_lock;
+	pthread_mutex_t pdcs_pending_lock;
+	ptree_node_t *pdcs_pending;
+	pthread_t pdcs_resultreader;
+	char pdcs_resultreader_errdesc[2048];
+};
+
+typedef struct {
+	int l_id;
+	void (*l_cb)(const char *key, const char *val, unsigned int len,
+	    pdict_reason_t, void *);
+	void *l_arg;
+} listener_t;
+
+typedef struct {
+	pthread_mutex_t grcba_lock;
+	pthread_cond_t grcba_cv;
+	int grcba_code;
+	int grcba_desired;
+	char *grcba_line;
+} getresult_cb_arg_t;
+
+typedef struct {
+	pthread_mutex_t grcba_lock;
+	pthread_cond_t grcba_cv;
+	int grcba_code;
+	int grcba_reply_len;
+	char *grcba_reply;
+	char *grcba_reply_ptr;
+} jgetresult_cb_arg_t;
+
+static int events;
+regex_t pendingex;
+static int initialized;
+
+static void *read_results(void *);
+static pdict_reason_t _pdict_reason_from_str(const char *s);
+static int cmd_timeout = 10000; //ms
+
+/*
+ * Stops the read thread and frees the pdc session
+ */
+void pdc_session_free(pdc_session_t *pdcs)
+{
+	if(pdcs)
+	{
+		//make sure the thread is not running...
+		pdc_readthread_join(pdcs, NULL);
+		
+		//free
+		pthread_mutex_destroy(&pdcs->pdcs_pending_lock);
+		pthread_mutex_destroy(&pdcs->pdcs_listeners_lock);
+		free(pdcs); pdcs = NULL;
+	}
+	return;
+}
+
+/*
+ * Allocates a new pdc session and starts a thread reading on readfd
+ */
+pdc_session_t * CCONV
+pdc_session_alloc(int readfd, int(*readfunc)(int, void *, unsigned int, char *,
+    int), int writefd, int(*writefunc)(int, const void *, unsigned int, char *,
+    int), int (*closefunc)(int, char *, int), void *cleanupPtr, void (*cleanupFunc)(void *))
+{
+	pdc_session_t *pdcs;
+#ifndef _WINDOWS
+	sigset_t new, old;
+#endif
+#ifdef _WINDOWS
+	int opt = 1;
+	int ssoret;
+#endif
+
+	if (!initialized)
+		pdc_init();
+
+	if (!(pdcs = malloc(sizeof (*pdcs))))
+		return NULL;
+	memset(pdcs, 0, sizeof (*pdcs));
+	pdcs->pdcs_rfd = readfd;
+	pdcs->pdcs_read = readfunc;
+	pdcs->pdcs_wfd = writefd;
+	pdcs->pdcs_write = writefunc;
+	pdcs->pdcs_close = closefunc;
+	pdcs->pdcs_cleanup_ptr = cleanupPtr;
+	pdcs->pdcs_cleanup = cleanupFunc;
+	if (pthread_mutex_init(&pdcs->pdcs_pending_lock, 0) != 0
+		|| pthread_mutex_init(&pdcs->pdcs_listeners_lock, 0) != 0) {
+		free(pdcs);
+		return NULL;
+	}
+#ifndef _WINDOWS
+	sigfillset(&new);
+	pthread_sigmask(SIG_BLOCK, &new, &old);
+#endif
+
+#ifdef _WINDOWS
+	//Disable Nagle
+	opt = TRUE;
+	ssoret = setsockopt(readfd, IPPROTO_TCP, TCP_NODELAY, (void *)&opt, sizeof (opt));
+	//set send buffer to 0 (send data right away)
+	opt = 0;
+	ssoret = setsockopt(readfd, SOL_SOCKET, SO_SNDBUF, (void *)&opt, sizeof (opt));
+#endif
+
+	if (pthread_create(&pdcs->pdcs_resultreader, 0, read_results, pdcs) !=
+	    0) {
+		pthread_mutex_destroy(&pdcs->pdcs_pending_lock);
+		pthread_mutex_destroy(&pdcs->pdcs_listeners_lock);
+		free(pdcs); pdcs = NULL;
+		return NULL;
+	}
+#ifdef _WINDOWS
+	//pthread_up_priority(pdcs->pdcs_resultreader);
+#endif
+#ifndef _WINDOWS
+	pthread_sigmask(SIG_SETMASK, &old, NULL);
+#endif
+
+	return pdcs;
+}
+
+/*
+ * Sets up regex for PENDING_PATTERN
+ */
+int CCONV
+pdc_init(void)
+{
+	int res;
+
+	if ((res = regcomp(&pendingex, PENDING_PATTERN, REG_EXTENDED)) != 0) {
+		LOG_STDERR(PHIDGET_LOG_CRITICAL, "pending report pattern compilation error %d", res);
+		ABORT();
+	}
+	initialized = 1;
+	return 1;
+}
+
+/*
+ * Compares 2 listener IDs
+ */
+static int
+lcmp(const void *sv, const void *tv)
+{
+	return ((const listener_t *)sv)->l_id - ((const listener_t *)tv)->l_id;
+}
+
+/*
+ * Compares 2 tag IDs
+ */
+static int
+tagcmp(const void *sv, const void *tv)
+{
+	return ((result_t *)sv)->r_tag - ((result_t *)tv)->r_tag;
+}
+
+/*
+ * Handles a report from the read thread
+ */
+static void
+handle_report(pdc_session_t *pdcs, char *line)
+{
+	regmatch_t pmatch[7];
+	listener_t *lp;
+	listener_t l;
+	char *idstr = NULL;
+	char *key = NULL;
+	char *val = NULL;
+	char *eoval = NULL; //end of val
+	char *soval = NULL; //start of val
+	char *reason_str = NULL;
+	pdict_reason_t r;
+	int res;
+	//char err[1024];
+
+	events++;
+
+//Disable this, I think it's causing bad problems!
+#if 0
+#ifdef _WINDOWS
+	//Send something back if we're on Windows, so this report gets ACKed right away
+	// otherwise, we face 200ms delayed ack -> clumps of events every 200ms
+	pdcs->pdcs_write(pdcs->pdcs_wfd, "report ack\n", strlen("report ack\n"), err, sizeof (err));
+#endif
+#endif
+
+	/* handle value without regex, due to crappy regex implementations */
+	/* val is bounded by double quotes */
+
+	if (!(val = strchr(line, '\"')))
+		goto ignore;
+	soval = ++val;
+	if (!(eoval = strchr(val, '\"')))
+		goto ignore;
+	
+	/* end of val \" = NULL */
+	*eoval = '\0';
+	
+	/* make sure there are no more double quotes */
+	if (strchr(eoval + 1, '\"') != NULL)
+		goto ignore;
+	
+	/* create a copy of val, or undo my changes if there is not enough memory */
+	if (!(val = strdup(val))) {
+		*eoval = '\"';
+		pu_log(PUL_WARN, pdcs->pdcs_rfd,
+		    "report dropped due to low memory");
+		goto ignore;
+	}
+	
+	/* put back what I changed and shift anything after val to the start of val */
+	*eoval = '\"';
+	memmove(soval, eoval, strlen(eoval) + 1);
+
+	if ((res = regexec(&pendingex, line, 6, pmatch, 0)) != 0) {
+ignore:
+#ifdef DEBUG_PROTOCOL
+		pu_log(PUL_DEBUG, pdcs->pdcs_rfd,
+		    "ignoring invalid report (see next line):");
+		pu_log(PUL_DEBUG, pdcs->pdcs_rfd, line);
+#endif
+		goto end;
+	}
+	if (!getmatchsub(line, &idstr, pmatch, 1) || !idstr)
+	{
+		goto end;
+	}
+	if (!getmatchsub(line, &key, pmatch, 2) || !key)
+	{
+		goto end;
+	}
+	if (!getmatchsub(line, &reason_str, pmatch, 3) || !reason_str ||
+	    !(r = _pdict_reason_from_str(reason_str)))
+	{
+		goto end;
+	}
+	free(reason_str); reason_str = NULL;
+
+	l.l_id = atoi(idstr);
+	pthread_mutex_lock(&pdcs->pdcs_listeners_lock);
+	if (ptree_contains(&l, (ptree_node_t *)pdcs->pdcs_listeners, lcmp,
+	    (void **)&lp) && lp->l_cb) {
+		char *ueval;
+		unsigned int uevlen;
+
+		pthread_mutex_unlock(&pdcs->pdcs_listeners_lock);
+		if (!unescape(val, &ueval, &uevlen)) {
+			/* XXX log */
+			goto end;
+		}
+		lp->l_cb(key, ueval, uevlen, r, lp->l_arg);
+		free(ueval);
+	}
+	else
+	{
+		pthread_mutex_unlock(&pdcs->pdcs_listeners_lock);
+		LOG(PHIDGET_LOG_WARNING, "No callback found for: %s",line);
+	}
+end:
+	free(val);
+	free(idstr);
+	free(key);
+}
+
+/*
+ * Called by read thread for any pending commands that didn't complete before the socket was closed.
+ */
+static ptree_walk_res_t
+finish_pending_async(const void *node, int level, void *arg, void *pwra)
+{
+	pdc_session_t *pdcs = arg;
+	result_t *r;
+	char msg[1024];
+
+	ptree_inorder_walk_remove(&pdcs->pdcs_pending, (void **)&r, pwra, tagcmp);
+	assert(node == r);
+
+	snprintf(msg, 1024, "Socket was closed before command T%d finished.",r->r_tag);
+
+	//notify frees r
+	if(r->r_notify)
+		r->r_notify(pdcs, r, 500, PTRUE, msg);
+
+	//free(r);
+
+	return PTREE_WALK_CONTINUE;
+}
+static ptree_walk_res_t
+free_node(const void *node, int level, void *arg, void *pwra)
+{
+	pdc_session_t *pdcs = arg;
+	void *ov;
+	ptree_inorder_walk_remove(&pdcs->pdcs_listeners, &ov, pwra, lcmp);
+	assert(node == ov);
+	free(ov);	
+	return PTREE_WALK_CONTINUE;
+}
+
+/*
+ * Runs in a thread started by pdc_session_alloc
+ */
+static void *
+read_results(void *arg)
+{
+	pdc_session_t *pdcs = arg;
+	char *line = 0;
+	result_t *r;
+	result_t rk;
+	char *startp;
+	int linelen;
+	int final;
+	int tag;
+
+next:
+	if (line) {
+		free(line); line = NULL;
+	}
+	pdcs->pdcs_resultreader_errdesc[0] = 0;
+	if (!pd_getline((char *)pdcs->pdcs_readbuf, sizeof (pdcs->pdcs_readbuf),
+	    &pdcs->pdcs_bufcur, &pdcs->pdcs_buflen, pdcs->pdcs_read, pdcs->pdcs_close,
+	    pdcs->pdcs_rfd, &line, pdcs->pdcs_resultreader_errdesc,
+	    sizeof (pdcs->pdcs_resultreader_errdesc)))
+	{
+		/* socket closed */
+		free(line); line=NULL;
+		
+		//respond for any pending commands (failed)...
+		pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+		ptree_walk(pdcs->pdcs_pending, PTREE_POSTORDER, finish_pending_async, tagcmp, pdcs);
+		assert(!pdcs->pdcs_pending);
+		//ptree_clear(&pdcs->pdcs_pending);
+		//pdcs->pdcs_pending = NULL;
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		
+		//remove listeners
+		pthread_mutex_lock(&pdcs->pdcs_listeners_lock);
+		ptree_walk(pdcs->pdcs_listeners, PTREE_POSTORDER, free_node, lcmp, pdcs);
+		assert(!pdcs->pdcs_listeners);
+		pthread_mutex_unlock(&pdcs->pdcs_listeners_lock);
+		//ptree_clear(&pdcs->pdcs_listeners);
+		
+		//do cleanup - this handles stuff outside on the dictionary code, etc.
+		if (pdcs->pdcs_cleanup)
+			pdcs->pdcs_cleanup(pdcs->pdcs_cleanup_ptr);
+		return (void *)(size_t)(-1 * errno); // XXX pdcs errno function
+	}
+
+	//LOG(PHIDGET_LOG_DEBUG, "read_results: %s",line);
+
+	if (line[0] == 'r') {
+		handle_report(pdcs, line);
+		goto next;
+	} else if (line[0] == 'T') {
+		tag = atoi(line + 1);
+		startp = strchr(line, ' ');
+		while (startp && *startp) {
+			if (isdigit(*startp))
+				goto notify;
+			startp++;
+		}
+		// XXX log malformed line
+		goto next;
+	} else {
+		startp = line;
+		tag = 0;
+	}
+notify:
+	if ((linelen = (int)strlen(line)) < 4) {
+		// XXX log malformed line
+		goto next;
+	}
+	final = (startp[3] == ' ');
+	pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	rk.r_tag = tag;
+	if (final) {
+		if (!ptree_remove(&rk, &pdcs->pdcs_pending, tagcmp,
+		    (void **)&r)) {
+			//ABORT();
+			pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+			// XXX log unanticipated result
+			goto next;
+		}
+	} else {
+		if (!ptree_contains(&rk, pdcs->pdcs_pending, tagcmp,
+		    (void **)&r)) {
+			//ABORT();
+			pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+			// XXX log unanticipated result
+			goto next;
+		}
+	}
+	pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+	r->r_notify(pdcs, r, atoi(startp), final, startp + 4);
+	r = NULL;
+	goto next;
+//	return  (void *)(0);
+}
+
+/*
+ * Callback from cmd
+ */
+static void
+getresult_cb(pdc_session_t *pdcs, result_t *r, int code, int final,
+    const char *line)
+{
+	getresult_cb_arg_t *grcba = r->r_arg;
+
+	if (final) {
+		if (pthread_mutex_lock(&grcba->grcba_lock))
+			abort();
+		grcba->grcba_code = code;
+		if (grcba->grcba_code != grcba->grcba_desired)
+			grcba->grcba_line = strdup(line);
+		if (pthread_mutex_unlock(&grcba->grcba_lock))
+			abort();
+		if (pthread_cond_signal(&grcba->grcba_cv))
+			abort();
+	}
+}
+
+/*
+ * Callback from jcmd
+ */
+static void
+jgetresult_cb(pdc_session_t *pdcs, result_t *r, int code, int final,
+    const char *line)
+{
+	jgetresult_cb_arg_t *grcba = r->r_arg;
+	int spaceleft = grcba->grcba_reply_len - (grcba->grcba_reply_ptr-grcba->grcba_reply);
+
+	if (pthread_mutex_lock(&grcba->grcba_lock))
+		abort();
+
+	if (final) 
+	{
+		grcba->grcba_code = code;
+	}
+
+	//add more space if needed
+	if((size_t)spaceleft < strlen(line)+1)
+	{
+		int offset = grcba->grcba_reply_ptr-grcba->grcba_reply;
+		if(!(grcba->grcba_reply = (char *)realloc(grcba->grcba_reply, strlen(line) + offset + 100)))
+		{
+			LOG(PHIDGET_LOG_WARNING,"Couldn't realloc!");
+			return;
+		}
+		grcba->grcba_reply_len = strlen(line) + offset + 100;
+		grcba->grcba_reply_ptr = grcba->grcba_reply + offset;
+		spaceleft = grcba->grcba_reply_len - (grcba->grcba_reply_ptr-grcba->grcba_reply);
+	}
+
+	strncpy(grcba->grcba_reply_ptr, line,
+	    spaceleft - 1);
+	if(!final)
+		grcba->grcba_reply_ptr[strlen(line)] = '\n';
+	grcba->grcba_reply_ptr+=(strlen(line)+1);
+
+	if (pthread_mutex_unlock(&grcba->grcba_lock))
+		abort();
+		
+	if (final) 
+	{
+		if (pthread_cond_signal(&grcba->grcba_cv))
+			abort();
+	}
+}
+
+static int wait_timeout(pthread_mutex_t *grcba_lock, pthread_cond_t *grcba_cv, int timeout_ms)
+{
+	int retval;
+#ifdef _WINDOWS
+	int timeout = timeout_ms;
+#else
+	struct timespec timeout;
+	struct timeval now;
+	
+    // Lock the mutex.
+    //pthread_mutex_lock(&grcba->grcba_lock);
+	
+	gettimeofday(&now,0);
+	timeout.tv_sec = now.tv_sec + timeout_ms/1000;
+	timeout.tv_nsec = now.tv_usec*1000 + (timeout_ms%1000 * 1000000);
+	if(timeout.tv_nsec >= 1000000000)
+	{
+		timeout.tv_sec++;
+		timeout.tv_nsec -= 1000000000;
+	}
+#endif
+	retval = pthread_cond_timedwait(grcba_cv, grcba_lock, &timeout);
+	//pthread_mutex_unlock(&grcba->grcba_lock); 
+	return retval;
+}
+
+static int
+cmd(pdc_session_t *pdcs, 
+	int desired, 
+	const char *cmd, 
+	char *errdesc,
+    int errlen)
+{
+	int len = (int)strlen(cmd);
+	getresult_cb_arg_t *grcba;
+	result_t r;
+	result_t *or;
+
+	if (!(grcba = malloc(sizeof (*grcba)))) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	r.r_tag = 0;
+	r.r_notify = getresult_cb;
+	r.r_arg = grcba;
+	memset(grcba, 0, sizeof (*grcba));
+	grcba->grcba_code = -1;
+	grcba->grcba_desired = desired;
+		
+	if (pthread_mutex_init(&grcba->grcba_lock, NULL))
+		abort();
+	if (pthread_cond_init(&grcba->grcba_cv, 0))
+		abort();
+	if (pthread_mutex_lock(&pdcs->pdcs_pending_lock))
+		abort();
+	if (!ptree_replace(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or)) {
+		pthread_mutex_destroy(&grcba->grcba_lock);
+		pthread_cond_destroy(&grcba->grcba_cv);
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		free(grcba); grcba = NULL;
+		if (errdesc)
+			(void) snprintf(errdesc, errlen,
+			    "result replacement failure");
+		return 0;
+	}
+	
+	// if or != NULL then pdcs_pending already contained a node with tag == r.r_tag
+	assert(!or);
+	if (pthread_mutex_lock(&grcba->grcba_lock))
+		abort();
+	if (pthread_mutex_unlock(&pdcs->pdcs_pending_lock))
+		abort();
+
+	if (!pdcs->pdcs_write(pdcs->pdcs_wfd, cmd, len, errdesc, errlen)) {
+		pthread_mutex_unlock(&grcba->grcba_lock);
+		pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+		ptree_remove(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		pthread_mutex_destroy(&grcba->grcba_lock);
+		pthread_cond_destroy(&grcba->grcba_cv);
+		free(grcba);
+		return 0;
+	}
+	
+	while (grcba->grcba_code == -1)
+	{
+		int retval;
+		//5 second timeout
+		if((retval = wait_timeout(&grcba->grcba_lock, &grcba->grcba_cv, cmd_timeout)) != 0)
+		{
+			switch (retval)
+			{
+				case ETIMEDOUT:
+				default:
+					if (errdesc)
+						(void) snprintf(errdesc, errlen,"cmd: \"%s\" Timed out after %dms.",cmd,cmd_timeout);
+					pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+					ptree_remove(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+					pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+					pthread_mutex_destroy(&grcba->grcba_lock);
+					pthread_cond_destroy(&grcba->grcba_cv);
+					if (grcba->grcba_line)
+						free(grcba->grcba_line);
+					free(grcba);
+					return 0; //Timeout
+			}
+		}
+	}
+	
+	pthread_mutex_destroy(&grcba->grcba_lock);
+	pthread_cond_destroy(&grcba->grcba_cv);
+	
+	if (grcba->grcba_code != desired)
+	{
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "protocol error: %d%s%s", grcba->grcba_code, grcba->grcba_line ? " " : "", grcba->grcba_line ? grcba->grcba_line : "");
+		if (grcba->grcba_line)
+			free(grcba->grcba_line);
+		free(grcba);
+		return 0;
+	}
+	
+	if (grcba->grcba_line)
+		free(grcba->grcba_line);
+	free(grcba);
+	return 1;
+}
+
+/*
+ * jcmd is for a command that returns results, like getKey
+ */
+static int
+jcmd(pdc_session_t *pdcs, 
+	 const char *cmd, 
+	 int desired, 
+	 char **reply,
+	 int replylen, 
+	 char *errdesc, 
+	 int errlen)
+{
+	int len = (int)strlen(cmd);
+	jgetresult_cb_arg_t *grcba;
+	result_t r;
+	result_t *or;
+
+	if (!(grcba = malloc(sizeof (*grcba)))) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	r.r_tag = 0;
+	r.r_notify = jgetresult_cb;
+	r.r_arg = grcba;
+	memset(grcba, 0, sizeof (*grcba));
+	grcba->grcba_code = -1;
+	grcba->grcba_reply = *reply;
+	grcba->grcba_reply_ptr = *reply;
+	grcba->grcba_reply_len = replylen;
+		
+	if (pthread_mutex_init(&grcba->grcba_lock, NULL))
+		abort();
+	if (pthread_cond_init(&grcba->grcba_cv, 0))
+		abort();
+	if (pthread_mutex_lock(&pdcs->pdcs_pending_lock))
+		abort();
+	if (!ptree_replace(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or)) {
+		pthread_mutex_destroy(&grcba->grcba_lock);
+		pthread_cond_destroy(&grcba->grcba_cv);
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		free(grcba); grcba = NULL;
+		if (errdesc)
+			(void) snprintf(errdesc, errlen,
+			    "result replacement failure");
+		return 0;
+	}
+	// if or != NULL then pdcs_pending already contained a node with tag == r.r_tag
+	assert(!or);
+	if (pthread_mutex_lock(&grcba->grcba_lock))
+		abort();
+	if (pthread_mutex_unlock(&pdcs->pdcs_pending_lock))
+		abort();
+
+#ifndef _MSC_EXTENSIONS
+#ifdef DEBUG_PROTOCOL
+	if (debug_protocol) {
+		write(1, "L: ", 3);
+		write(1, cmd, len);
+	}
+#endif
+#endif
+	if (!pdcs->pdcs_write(pdcs->pdcs_wfd, cmd, len, errdesc, errlen)) {
+		pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+		ptree_remove(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		pthread_mutex_destroy(&grcba->grcba_lock);
+		pthread_cond_destroy(&grcba->grcba_cv);
+		free(grcba); grcba = NULL;
+		return 0;
+	}
+	while (grcba->grcba_code == -1)
+	{
+		int retval;
+		//5 second timeout
+		if((retval = wait_timeout(&grcba->grcba_lock, &grcba->grcba_cv, cmd_timeout)) != 0)
+		{
+			switch (retval)
+			{
+				case ETIMEDOUT:
+				default:
+					if (errdesc)
+						(void) snprintf(errdesc, errlen,"jcmd: \"%s\" Timed out after %dms.",cmd,cmd_timeout);
+					pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+					ptree_remove(&r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+					pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+					pthread_mutex_destroy(&grcba->grcba_lock);
+					pthread_cond_destroy(&grcba->grcba_cv);
+					free(grcba); grcba = NULL;
+					return 0;
+			}
+		}
+		//if (pthread_cond_wait(&grcba->grcba_cv, &grcba->grcba_lock))
+		//	abort();
+	}
+	pthread_mutex_destroy(&grcba->grcba_lock);
+	pthread_cond_destroy(&grcba->grcba_cv);
+	if (grcba->grcba_code != desired) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen,
+			    "protocol error: %d%s%s", grcba->grcba_code,
+			    grcba->grcba_reply ? " " : "", grcba->grcba_reply ?
+			    grcba->grcba_reply : "");
+		*reply = grcba->grcba_reply;
+		free(grcba); grcba = NULL;
+		return 0;
+	}
+	*reply = grcba->grcba_reply;
+	free(grcba); grcba = NULL;
+	return 1;
+}
+
+//int CCONV
+//pdc_set(pdc_session_t *pdcs, const char *key, const char *val,
+//    int len, int remove_on_close, char *errdesc, int errlen)
+//{
+//	char *escval;
+//	char *buf;
+//	int res;
+//
+//	if(val[0] == '\0')
+//	{
+//		if (!escape("\001", len, &escval)) {
+//			if (errdesc)
+//				(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+//			return 0;
+//		}
+//	}
+//	else
+//	{
+//		if (!escape(val, len, &escval)) {
+//			if (errdesc)
+//				(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+//			return 0;
+//		}
+//	}
+//	if ((len = pasprintf(&buf, "set %s=\"%s\"%s\n", key, escval,
+//	    remove_on_close ? " for session" : "")) < 0) {
+//		free((void *)escval); escval = NULL;
+//		if (errdesc)
+//			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+//		return 0;
+//	}
+//	res = cmd(pdcs, 200, buf, errdesc, errlen);
+//	free((void *)escval); escval = NULL;
+//	free(buf); buf = NULL;
+//	return res;
+//}
+
+/*
+ * Returns a nonzero listen ID on success, 0 otherwise.
+ */
+pdc_listen_id_t CCONV
+pdc_listen(pdc_session_t *pdcs, const char *pattern,
+    void (*cb)(const char *, const char *, unsigned int, pdict_reason_t,
+    void *), void *arg, char *errdesc, int errlen)
+{
+	listener_t *l;
+	static int lid = 1;
+	char *buf;
+	int len;
+
+	if(!pdcs)
+		return 0;
+
+	if (!(l = malloc(sizeof (*l)))) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	l->l_id = lid;
+	l->l_cb = cb;
+	l->l_arg = arg;
+	if ((len = pasprintf(&buf, "listen \"%s\" lid%d\n", pattern, lid++)) < 0)
+	{
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+
+	pthread_mutex_lock(&pdcs->pdcs_listeners_lock);
+	if (!ptree_replace(l, (ptree_node_t **)&pdcs->pdcs_listeners, lcmp, NULL)) {
+		free(buf); buf = NULL;
+		free(l); l = NULL;
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		pthread_mutex_unlock(&pdcs->pdcs_listeners_lock);
+		return 0;
+	}
+	pthread_mutex_unlock(&pdcs->pdcs_listeners_lock);
+
+	if (!cmd(pdcs, 200, buf, errdesc, errlen)) {
+		free(buf); buf = NULL;
+		return 0;
+	}
+
+	free(buf); buf = NULL;
+	return l->l_id;
+}
+
+int CCONV
+pdc_enable_periodic_reports(pdc_session_t *pdcs, int periodms,
+    char *errdesc, int errlen)
+{
+	char *buf;
+	int res;
+	if(!pdcs)
+		return 0;
+
+	if (periodms <= 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "invalid period");
+		return 0;
+	}
+	if (pasprintf(&buf, "report %d report\n", periodms) < 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	res = cmd(pdcs, 200, buf, errdesc, errlen);
+	free(buf); buf = NULL;
+	return res;
+}
+
+int CCONV
+pdc_disable_periodic_reports(pdc_session_t *pdcs, char *errdesc,
+    int errlen)
+{
+	if(!pdcs)
+		return 0;
+	return cmd(pdcs, 200, "report 0 report\n", errdesc, errlen);
+}
+
+int CCONV
+pdc_flush(pdc_session_t *pdc, char *errdesc, int errlen)
+{
+	if(!pdc)
+		return 0;
+	return cmd(pdc, 200, "flush\n", errdesc, errlen);
+}
+
+int CCONV
+pdc_ignore(pdc_session_t *pdcs, pdc_listen_id_t id, char *errdesc,
+    int errlen)
+{
+	char *buf;
+	int res;
+
+	if(!pdcs)
+		return 0;
+
+	if (pasprintf(&buf, "ignore lid%d\n", id) < 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	res = cmd(pdcs, 200, buf, errdesc, errlen);
+	free(buf); buf = NULL;
+	return res;
+}
+
+//Async stuff starts here
+static void
+async_cmd_callback(pdc_session_t *pdcs, const char *cmd, 
+	void (*r_notify)(pdc_session_t *, struct result *, int code,int final, const char *line), void *r_arg,
+    void (*error)(const char *errdesc, void *arg), void *arg);
+
+typedef struct {
+	int aca_desired;
+	void (*aca_error)(const char *errdesc, void *arg);
+	void *aca_error_arg;
+} async_cmd_arg_t;
+
+/*
+ * Default callback for async commands.
+ * If the code matches what we wanted, then we don't hear anything about it.
+ * Otherwise, we get the error callback.
+ */
+static void
+async_cmd_cb(pdc_session_t *pdcs, result_t *r, int code, int final,
+    const char *line)
+{
+	async_cmd_arg_t *aca;
+	char *buf;
+
+	assert(final);
+	aca = (async_cmd_arg_t *)r->r_arg;
+	if (aca->aca_desired != code) {
+		if (pasprintf(&buf, "protocol error: %s", line) > 0) {
+			if(aca->aca_error)
+				aca->aca_error(buf, aca->aca_error_arg);
+			free(buf); buf = NULL;
+		} else {
+			/* XXX warn about low-memory condition */
+			if(aca->aca_error)
+				aca->aca_error("protocol error (insufficient memory"
+			    " to describe)", aca->aca_error_arg);
+		}
+	}
+	free(aca); aca = NULL;
+	free(r); r = NULL;
+}
+
+typedef struct {
+	void (*aaa_success)(void *arg, void (*error)(const char *errdesc, void *arg));
+	void (*aaa_error)(const char *errdesc, void *arg);
+	void *aaa_arg;
+	char *aaa_pass;
+} async_auth_arg_t;
+
+static void
+async_auth_cb(pdc_session_t *pdcs, result_t *r, int code, int final,
+    const char *line)
+{
+	async_auth_arg_t *aaa;
+
+	assert(final);
+	aaa = (async_auth_arg_t *)r->r_arg;
+
+	switch(code)
+	{
+		//Authentication needed
+		case 999:
+			{		
+				int randomHash;
+				char buf[100];
+				md5_state_t state;
+				md5_byte_t digest[16];
+
+				int di;
+
+				randomHash = strtol(line, NULL, 0);
+				
+				LOG(PHIDGET_LOG_DEBUG,"Got hash: %d",randomHash);
+				
+				if(aaa->aaa_pass)
+				{
+					snprintf(buf,100,"%d%s",randomHash,aaa->aaa_pass);
+				}
+
+				md5_init(&state);
+				md5_append(&state, (const md5_byte_t *)buf, (int)strlen(buf));
+				md5_finish(&state, digest);
+				
+				memset(buf, 0, sizeof(buf));
+				
+				snprintf(buf, 5, "997 ");
+				
+				for (di = 0; di < 16; ++di)
+					snprintf((buf+4) + di * 2, 3, "%02x", digest[di]);
+					
+				buf[strlen(buf)] = '\n';
+
+				//send out the next async command
+				async_cmd_callback(pdcs, buf, async_auth_cb, aaa, aaa->aaa_error, aaa->aaa_arg);
+
+				goto exit_pending;
+			}
+		// Authentication failed
+		case 998:		
+			LOG(PHIDGET_LOG_INFO,"Authentication Failed - bad password");
+			//send out the error event
+			if(aaa->aaa_error)
+				aaa->aaa_error("Authentication Failed - bad password", aaa->aaa_arg);
+			goto exit_done;
+		// Authentication is good, or not needed
+		case 996:
+			//make sure there is a verison - otherwise this is a really old webservice
+			{
+				char *versionstring = strstr(line, "version=");
+				if(!versionstring)
+				{
+					LOG(PHIDGET_LOG_ERROR,"Trying to connect to an old webservice. Webservice needs to be updated.");
+					//send out an error event
+					if(aaa->aaa_error)
+						aaa->aaa_error("Trying to connect to an old webservice. Webservice needs to be updated.", aaa->aaa_arg);
+					goto exit_done;
+				}
+				//It's good - this is the end of authentication
+				else
+				{
+					//send out auth_finished event
+					if(aaa->aaa_success)
+						aaa->aaa_success(aaa->aaa_arg, aaa->aaa_error);
+				}
+			}
+			goto exit_done;
+		// Version mismatch
+		case 994:
+			{
+				char buf[100];
+				snprintf(buf, sizeof(buf), "Version Mismatch; webservice:%s, expecting:%s",strlen(line)>24?(line+24):line,ws_protocol_ver);
+				LOG(PHIDGET_LOG_ERROR, buf);
+				//send out error event
+				if(aaa->aaa_error)
+					aaa->aaa_error(buf, aaa->aaa_arg);
+			}
+			goto exit_done;
+		case 500:
+			{
+				//callback an error
+				char buf[100];
+				snprintf(buf, sizeof(buf), "Network Error (probably a timeout occured): %s", line);
+				LOG(PHIDGET_LOG_ERROR, buf);
+				//send out error event
+				if(aaa->aaa_error)
+					aaa->aaa_error(buf, aaa->aaa_arg);
+			}
+			goto exit_done;
+		default:
+			//callback an error
+			if(aaa->aaa_error)
+				aaa->aaa_error("Unexpected response during authorization. Probably a version mismatch - upgrade the Webservice.", aaa->aaa_arg);
+			goto exit_done;
+	}
+
+exit_done:
+	free(aaa); aaa = NULL;
+exit_pending:
+	free(r); r = NULL;
+}
+
+/*
+ * Sends out an async command. Response comes back on the r_notify callback.
+ */
+static void
+async_cmd_callback(pdc_session_t *pdcs, const char *cmd, 
+	void (*r_notify)(pdc_session_t *, struct result *, int code,int final, const char *line), void *r_arg,
+    void (*error)(const char *errdesc, void *arg), void *arg)
+{
+	static int tag = 1;
+	char errdesc[256];
+	result_t *r;
+	result_t *or;
+	char *buf;
+	int len;
+
+	if ((len = pasprintf(&buf, "T%d %s", tag, cmd)) < 0) {
+		if(error)
+			error(strerror(errno), arg);
+		return;
+	}
+	if (!(r = malloc(sizeof (*r)))) {
+		if(error)
+			error(strerror(errno), arg);
+		return;
+	}
+	/* add pending */
+	pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	r->r_tag = tag++;
+	r->r_notify = r_notify;
+	r->r_arg = r_arg;
+	ptree_replace(r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+	// if or != NULL then pdcs_pending already contained a node with tag == r.r_tag
+	assert(!or);
+
+	if (!pdcs->pdcs_write(pdcs->pdcs_wfd, buf, (int)strlen(buf), errdesc,
+		(unsigned int)sizeof (errdesc))) {
+		if(error)
+			error(errdesc, arg);
+		ptree_remove(r, &pdcs->pdcs_pending, tagcmp, (void **)&or);
+		if(r == or && r)
+		{
+			if(r->r_arg)
+				free(r->r_arg); r->r_arg = NULL;
+			free(r); r = NULL;
+		}
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		return;
+	}
+
+	pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+	free(buf); buf = NULL;
+}
+
+/*
+ * Sends out an async command. Errors come back on the error callback if desired does not match the response.
+ */
+static void
+async_cmd(pdc_session_t *pdcs, int desired, const char *cmd,
+    void (*error)(const char *errdesc, void *arg), void *arg)
+{
+	async_cmd_arg_t *aca;
+
+	if (!(aca = malloc(sizeof (*aca)))) {
+		if(error)
+			error(strerror(errno), arg);
+		return;
+	}
+
+	aca->aca_desired = desired;
+	aca->aca_error = error;
+	aca->aca_error_arg = arg;
+
+	async_cmd_callback(pdcs, cmd, async_cmd_cb, aca, error, arg);
+}
+
+typedef struct _pdc_session_and_arg_t{
+	pdc_session_t *pdcs;
+	void *r_arg;
+} pdc_session_and_arg_t;
+
+static ptree_walk_res_t
+clean_pending_async(const void *node, int level, void *arg, void *pwra)
+{
+	result_t *rk = (result_t *)node;
+	pdc_session_and_arg_t *a = (pdc_session_and_arg_t *)arg;
+	result_t *r;
+	char msg[1024];
+
+	if(rk->r_notify == &async_cmd_cb && ((async_cmd_arg_t *)rk->r_arg)->aca_error_arg == a->r_arg)
+	{
+		ptree_inorder_walk_remove(&a->pdcs->pdcs_pending, (void **)&r, pwra, tagcmp);
+		assert(rk == r);
+
+		snprintf(msg, 1024, "Socket was closed before command T%d finished.",r->r_tag);
+
+		//notify frees r
+		if(r->r_notify)
+			r->r_notify(a->pdcs, r, 500, PTRUE, msg);
+	}
+
+	return PTREE_WALK_CONTINUE;
+}
+
+static int
+async_pending_cmp(const void *arg, const void *tv)
+{
+	result_t *rk = (result_t *)tv;
+
+	if(rk->r_notify == &async_cmd_cb && ((async_cmd_arg_t *)rk->r_arg)->aca_error_arg == arg)
+		return 0;
+	return -1;
+}
+
+//If there are pending keys for this arg, then 1st wait up to 1 second for them to finish, then report the errors
+void
+cleanup_pending(pdc_session_t *pdcs, void *arg)
+{
+	int timeout = 500; //ms
+	pdc_session_and_arg_t a = {pdcs, arg};
+
+	pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	//TODO: Do this differently - can't use a different compare then the one used to add to the list!
+	while(ptree_contains(arg, pdcs->pdcs_pending, async_pending_cmp, NULL) && timeout>0)
+	{
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		SLEEP(10);
+		timeout-=10;
+		pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	}
+	if(timeout<=0)
+	{
+		ptree_walk(pdcs->pdcs_pending, PTREE_POSTORDER, clean_pending_async, tagcmp, &a);
+	}
+	pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+}
+
+//loop so long as there are any pending
+void
+wait_pending(pdc_session_t *pdcs)
+{
+	int timeout = 500; //ms
+
+	pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	while(pdcs->pdcs_pending && timeout>0)
+	{
+		pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+		SLEEP(10);
+		timeout-=10;
+		pthread_mutex_lock(&pdcs->pdcs_pending_lock);
+	}
+	pthread_mutex_unlock(&pdcs->pdcs_pending_lock);
+}
+
+void CCONV
+pdc_async_enable_periodic_reports(pdc_session_t *pdcs, int periodms,
+								  void (*error)(const char *errdesc, void *arg),
+								  void *arg)
+{
+	char *buf;
+	
+	if(!pdcs)
+		return;
+
+	if (periodms <= 0) {
+		if (error)
+			error("invalid period", arg);
+		return;
+	}
+	if (pasprintf(&buf, "report %d report\n", periodms) < 0) {
+		if (error)
+			error(strerror(errno), arg);
+		return;
+	}
+	async_cmd(pdcs, 200, buf, error, arg);
+	free(buf); buf = NULL;
+}
+
+void CCONV
+pdc_async_disable_periodic_reports(pdc_session_t *pdcs, void (*error)(const char *errdesc, void *arg),
+								   void *arg)
+{
+	if(!pdcs)
+		return;
+	async_cmd(pdcs, 200, "report 0 report\n", error, arg);
+}
+
+void CCONV
+pdc_async_set(pdc_session_t *pdcs, const char *key, const char *val,
+    int len, int remove_on_close, void (*error)(const char *errdesc, void *arg),
+    void *arg)
+{
+	char *escval;
+	char *buf;
+	if(!pdcs)
+		return;
+		
+	if(val[0] == '\0')
+	{
+		if (!escape("\001", len, &escval)) {
+			if (error)
+				error(strerror(errno), arg);
+			return;
+		}
+	}
+	else
+	{
+		if (!escape(val, len, &escval)) {
+			if (error)
+				error(strerror(errno), arg);
+			return;
+		}
+	}
+	if (pasprintf(&buf, "set %s=\"%s\"%s\n", key, escval,
+	    remove_on_close ? " for session" : "") < 0) {
+		free((void *)escval); escval = NULL;
+		if (error)
+			error(strerror(errno), arg);
+		return;
+	}
+	async_cmd(pdcs, 200, buf, error, arg);
+	free(buf); buf = NULL;
+	free((void *)escval); escval = NULL;
+}
+
+void CCONV
+pdc_async_ignore(pdc_session_t *pdcs, pdc_listen_id_t id, 
+	void (*error)(const char *errdesc, void *arg), void *arg)
+{
+	char *buf;
+	if(!pdcs)
+		return;
+	
+	if (pasprintf(&buf, "ignore lid%d\n", id) < 0) {
+		if (error)
+			error(strerror(errno), arg);
+		return;
+	}
+	async_cmd(pdcs, 200, buf, error, arg);
+	free(buf); buf = NULL;
+}
+
+void CCONV
+pdc_async_authorize(pdc_session_t *pdcs, const char *version, char *password, 
+	void (*success) (void *arg, void (*error)(const char *errdesc, void *arg)),
+	void (*error)(const char *errdesc, void *arg), void *arg)
+{	
+	char *buf;
+	async_auth_arg_t *aaa;
+
+	if(!pdcs)
+		return;
+
+	if (!(aaa = malloc(sizeof (*aaa)))) {
+		if(error)
+			error(strerror(errno), arg);
+		return;
+	}
+
+	aaa->aaa_success = success;
+	aaa->aaa_error = error;
+	aaa->aaa_arg = arg;
+	aaa->aaa_pass = password;
+	
+	if (pasprintf(&buf, "995 authenticate, version=%s\n", version) < 0) {
+		if (error)
+			error(strerror(errno), arg);
+		return;
+	}
+
+	//200 is bogus - I want to get the callback on any response
+	async_cmd_callback(pdcs, buf, async_auth_cb, aaa, error, arg);
+	free(buf); buf = NULL;
+}
+
+void CCONV
+pdc_async_remove(pdc_session_t *pdcs, const char *pattern, void (*error)(const char *errdesc, void *arg), void *arg)
+{
+	char *buf;
+	
+	if(!pdcs)
+		return;
+	
+	if (pasprintf(&buf, "remove %s\n", pattern) < 0) {
+		if (error)
+			error(strerror(errno), arg);
+		return;
+	}
+	async_cmd(pdcs, 200, buf, error, arg);
+	free(buf); buf = NULL;
+}
+
+int CCONV
+pdc_quit(pdc_session_t *pdc, char *errdesc, int errlen)
+{
+	return cmd(pdc, 200, "quit\n", errdesc, errlen);
+}
+
+int CCONV
+pdc_remove(pdc_session_t *pdcs, const char *pattern, char *errdesc, int errlen)
+{
+	char *buf;
+	int res;
+	
+	if(!pdcs)
+		return 0;
+
+	if (pasprintf(&buf, "remove %s\n", pattern) < 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	res = cmd(pdcs, 200, buf, errdesc, errlen);
+	free(buf); buf = NULL;
+	return res;
+}
+
+int CCONV
+pdc_get(pdc_session_t *pdcs, const char *key, char *val, int vallen, char *errdesc, int errlen)
+{
+	char *buf, *results;
+	int res, resultsize;
+
+	if(!pdcs)
+		return 0;
+	
+	resultsize = vallen + 30;
+
+	results = (char *)malloc(resultsize);
+
+	if (pasprintf(&buf, "get %s\n", key) < 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		free(results);
+		return 0;
+	}
+	res = jcmd(pdcs, buf, 200, &results, resultsize, errdesc, errlen);
+	if(res)
+	{
+		char *ueval, *eol, *resultval;
+		unsigned int uevlen;
+
+		if ((eol = strchr(results, '\n')))
+		{
+			eol[0] = '\0';
+		}
+		if(!(resultval = strstr(results, "value ")))
+		{
+			val[0] = 0;
+			goto end;
+		}
+		resultval += 6;
+
+		if (!unescape(resultval, &ueval, &uevlen)) {
+			/* XXX log */
+			goto end;
+		}
+		strncpy(val, ueval, vallen-1);
+		val[vallen-1]='\0';
+		free(ueval);
+	}
+
+end:
+	free(buf);
+	free(results);
+	return res;
+}
+
+static pdict_reason_t
+_pdict_reason_from_str(const char *s)
+{
+	if (strcmp(s, "changed") == 0)
+		return PDR_VALUE_CHANGED;
+	if (strcmp(s, "current") == 0)
+		return PDR_CURRENT_VALUE;
+	if (strcmp(s, "added") == 0)
+		return PDR_ENTRY_ADDED;
+	if (strcmp(s, "removing") == 0)
+		return PDR_ENTRY_REMOVING;
+	return 0;
+}
+
+int CCONV
+pdc_get_server_session_id(pdc_session_t *pdc, int *id, char *errdesc,
+    int errlen)
+{
+	char reply[80];
+	char *buf;
+	int res;
+
+	if(!pdc)
+		return 0;
+
+	if (pasprintf(&buf, "get session id\n") < 0) {
+		if (errdesc)
+			(void) snprintf(errdesc, errlen, "%s", strerror(errno));
+		return 0;
+	}
+	res = jcmd(pdc, buf, 200, (char **)&reply, sizeof (reply), errdesc, errlen);
+	free(buf); buf = NULL;
+
+	if (res && id)
+		*id = atoi(reply);
+	return res;
+}
+
+int CCONV
+pdc_readthread_join(pdc_session_t *pdcs, void **status)
+{
+	int res=0;
+	if(pdcs)
+	{
+		if(pdcs->pdcs_resultreader)
+			res = pthread_join(pdcs->pdcs_resultreader, status);
+		pdcs->pdcs_resultreader = 0;
+	}
+	return res;
+}