wrapper: add a local buffer for client

2 files changed, 168 insertions, 0 deletions

diff --git a/wrapper/ppswrapper-client.c b/wrapper/ppswrapper-client.c
index 540b868..3f5fd98 100644
--- a/wrapper/ppswrapper-client.c
+++ b/wrapper/ppswrapper-client.c
@@ -128,6 +128,7 @@ ppsw_client_create(int port)
 	close (fds[1]);
     }
 
+    self->buf = NULL;
     self->categories = NULL;
     self->channel = NULL;
     self->callback = NULL;
@@ -136,6 +137,7 @@ ppsw_client_create(int port)
     self->evt_quit = 0;
 
     pthread_mutex_init (&self->mutex, NULL);
+    pthread_mutex_init (&self->packet_mutex, NULL);
 
     return &self->base;
  spawn_server_failed:
@@ -217,6 +219,7 @@ ppsw_client_destroy(ppsw_t* super)
 
     ppsw_client_stop_event_thread(super);
     pthread_mutex_destroy (&self->mutex);
+    pthread_mutex_destroy (&self->packet_mutex);
 
     free_categories(self->categories);
     ppsw_free_channel(self->channel);
@@ -224,6 +227,157 @@ ppsw_client_destroy(ppsw_t* super)
     free (self);
 }
 
+static int
+ppsw_client_check_local (ppsw_t *super, uint16_t id,
+			 ppsw_msg_t *retmsg,
+			 ppsw_packet_t *retp)
+{
+    ppsw_client_t *self = (ppsw_client_t*)super;
+    ppsw_buf_t *buf, *prev = NULL;
+
+    for (buf = self->buf; buf; buf = buf->next, prev = buf) {
+	if (buf->id == id) {
+	    if (!prev)
+		self->buf = buf->next;
+	    else
+		prev->next = buf->next;
+	    *retmsg = buf->msg;
+	    *retp = buf->packet;
+	    free (buf);
+	    return 0;
+	}
+    }
+
+    return -1;
+}
+
+static int
+ppsw_client_push_buf (ppsw_t *super,
+		      ppsw_msg_t *msg,
+		      ppsw_packet_t *packet,
+		      uint16_t id)
+{
+    ppsw_client_t *self = (ppsw_client_t*)super;
+    ppsw_buf_t *buf = malloc (sizeof (*buf));
+    ppsw_buf_t *prev = NULL;
+
+    if (!buf)
+	return -1;
+
+    buf->msg = *msg;
+    buf->packet = *packet;
+    buf->id = id;
+    buf->next = NULL;
+
+    if (!self->buf) {
+	self->buf = buf;
+	return 0;
+    }
+
+    for (prev = self->buf; prev; prev = prev->next) {
+	if (!prev->next) {
+	    prev->next = buf;
+	    break;
+	}
+    }
+    return 0;
+}
+
+static int
+ppsw_client_recv (ppsw_t *super, uint16_t id,
+		  ppsw_msg_t *retmsg,
+		  ppsw_packet_t *retp)
+{
+    ppsw_client_t *self = (ppsw_client_t*)super;
+    ppsw_msg_t msg;
+    ppsw_packet_t p;
+    uint32_t packet_id;
+    fd_set rfs;
+    fd_set efs;
+    int ret;
+    struct timeval tv;
+
+    if (!ppsw_client_check_local (super, id, retmsg, retp))
+	return 0;
+
+    //printf ("wait for a message from client\n");
+    pthread_mutex_lock (&self->packet_mutex);
+    if (self->error) {
+	ret = -1;
+	goto done;
+    }
+    if (!ppsw_client_check_local (super, id, retmsg, retp)) {
+	ret = 0;
+	goto done;
+    }
+
+    for (;;) {
+	FD_ZERO(&rfs);
+	FD_SET(self->base.sock, &rfs);
+	FD_ZERO(&efs);
+	FD_SET(self->base.sock, &efs);
+
+	memset (&p, 0, sizeof(p));
+
+	tv.tv_sec = 1;
+	tv.tv_usec = 0;
+
+	ret = select(self->base.sock + 1, &rfs, NULL, &efs, &tv);
+	if (ret < 0 && errno == EINTR)
+	    continue;
+
+	if (FD_ISSET (self->base.sock, &efs)) {
+	    self->error = 1;
+	    ret = -1;
+	    goto done;
+	}
+
+	if (!FD_ISSET (self->base.sock, &rfs))
+	    continue;
+
+	if (ppsw_recv_packet (super, &msg, &p) < 0) {
+	    self->error = 1;
+	    ret = -1;
+	    goto done;
+	}
+
+	if (ppsw_packet_readl (&p, &packet_id) < 0) {
+	    self->error = 1;
+	    ret = -1;
+	    goto done;
+	}
+
+	if (packet_id == id) {
+	    ret = 0;
+	    goto done;
+	}
+
+	/* others packet, push it in queue */
+	if (ppsw_client_push_buf (super, &msg, &p, packet_id) < 0) {
+	    self->error = 1;
+	    ret = -1;
+	    goto done;
+	}
+	pthread_mutex_unlock (&self->packet_mutex);
+
+	/* ugly */
+	usleep (1000);
+	if (self->error) {
+	    ret = -1;
+	    goto done;
+	}
+	if (!ppsw_client_check_local (super, id, retmsg, retp)) {
+	    ret = 0;
+	    goto done;
+	}
+	pthread_mutex_lock (&self->packet_mutex);
+    }
+
+ done:
+     pthread_mutex_unlock (&self->packet_mutex);
+     return ret;
+}
+
 int ppsw_client_send_create(ppsw_t* super,
 			    const char *oem_name,
 			    const char *terminal_type,
diff --git a/wrapper/ppswrapper.h b/wrapper/ppswrapper.h
index f5c9605..9ac0319 100644
--- a/wrapper/ppswrapper.h
+++ b/wrapper/ppswrapper.h
@@ -143,6 +143,15 @@ typedef struct {
     volatile uint16_t seq;
 } ppsw_t;
 
+typedef struct ppsw_buf {
+    ppsw_msg_t msg;
+    ppsw_packet_t packet;
+
+    uint8_t id;
+
+    struct ppsw_buf *next;
+} ppsw_buf_t;
+
 typedef struct {
     ppsw_t          base;
 
@@ -154,7 +163,12 @@ typedef struct {
     ppscategory*    categories;
     ppschannel*     channel;
 
+    ppsw_buf_t     *buf;
+
     pthread_mutex_t mutex;
+    pthread_mutex_t packet_mutex;
+
+    int             error;
 
     int             has_thread;
     int             evt_quit;