X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=src%2Fafb-ws.c;h=c48a5e242576e157bc4dd8883421d1fbeede2adc;hb=f15ea770dd9b13a20331853a026091316984f9ca;hp=4e0138a83d4758d39958f7a2cc055fc9cc2d6e56;hpb=c6b4fa9e2c35d5c8bdbf929b6f1622f31afd35c9;p=src%2Fapp-framework-binder.git

diff --git a/src/afb-ws.c b/src/afb-ws.c
index 4e0138a8..c48a5e24 100644
--- a/src/afb-ws.c
+++ b/src/afb-ws.c
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2016 "IoT.bzh"
+ * Copyright (C) 2016, 2017 "IoT.bzh"
  * Author: José Bollo <jose.bollo@iot.bzh>
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -99,11 +99,21 @@ struct afb_ws
 static inline struct buf aws_pick_buffer(struct afb_ws *ws)
 {
 	struct buf result = ws->buffer;
+	if (result.buffer)
+		result.buffer[result.size] = 0;
 	ws->buffer.buffer = NULL;
 	ws->buffer.size = 0;
 	return result;
 }
 
+/*
+ * Clear the current buffer
+ */
+static inline void aws_clear_buffer(struct afb_ws *ws)
+{
+	ws->buffer.size = 0;
+}
+
 /*
  * Disconnect the websocket 'ws' and calls on_hangup if
  * 'call_on_hangup' is not null.
@@ -116,7 +126,7 @@ static void aws_disconnect(struct afb_ws *ws, int call_on_hangup)
 		sd_event_source_unref(ws->evsrc);
 		ws->evsrc = NULL;
 		websock_destroy(wsi);
-		free(aws_pick_buffer(ws).buffer);
+		free(ws->buffer.buffer);
 		ws->state = waiting;
 		if (call_on_hangup && ws->itf->on_hangup)
 			ws->itf->on_hangup(ws->closure);
@@ -337,21 +347,58 @@ int afb_ws_binary_v(struct afb_ws *ws, const struct iovec *iovec, int count)
  */
 static ssize_t aws_writev(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 {
-	ssize_t rc;
+	int i;
+	ssize_t rc, sz, dsz;
+	struct iovec *iov2;
+	struct pollfd pfd;
+
+	/* compute the size */
+	dsz = 0;
+	i = 0;
+	while (i < iovcnt) {
+		dsz += iov[i++].iov_len;
+		if (dsz < 0) {
+			errno = EINVAL;
+			return -1;
+		}
+	}
+	if (dsz == 0)
+		return 0;
+
+	/* write the data */
+	iov2 = (struct iovec*)iov;
+	sz = dsz;
 	for (;;) {
-		rc = writev(ws->fd, iov, iovcnt);
-		if (rc == -1) {
+		rc = writev(ws->fd, iov2, iovcnt);
+		if (rc < 0) {
 			if (errno == EINTR)
 				continue;
-			else if (errno == EAGAIN) {
-				struct pollfd pfd;
-				pfd.fd = ws->fd;
-				pfd.events = POLLOUT;
-				poll(&pfd, 1, 10);
-				continue;
+			if (errno != EAGAIN)
+				return -1;
+		} else {
+			dsz -= rc;
+			if (dsz == 0)
+				return sz;
+
+			i = 0;
+			while (rc >= (ssize_t)iov2[i].iov_len)
+				rc -= (ssize_t)iov2[i++].iov_len;
+
+			iovcnt -= i;
+			if (iov2 != iov)
+				iov2 += i;
+			else {
+				iov += i;
+				iov2 = alloca(iovcnt * sizeof *iov2);
+				for (i = 0 ; i < iovcnt ; i++)
+					iov2[i] = iov[i];
 			}
+			iov2->iov_base += rc;
+			iov2->iov_len -= rc;
 		}
-		return rc;
+		pfd.fd = ws->fd;
+		pfd.events = POLLOUT;
+		poll(&pfd, 1, 10);
 	}
 }
 
@@ -364,9 +411,7 @@ static ssize_t aws_readv(struct afb_ws *ws, const struct iovec *iov, int iovcnt)
 	do {
 		rc = readv(ws->fd, iov, iovcnt);
 	} while(rc == -1 && errno == EINTR);
-	if (rc == -1 && errno == EAGAIN) {
-		rc = 0;
-	} else if (rc == 0) {
+	if (rc == 0) {
 		errno = EPIPE;
 		rc = -1;
 	}
@@ -381,7 +426,7 @@ static void aws_on_readable(struct afb_ws *ws)
 	int rc;
 
 	assert(ws->ws != NULL);
-	rc = websock_dispatch(ws->ws);
+	rc = websock_dispatch(ws->ws, 0);
 	if (rc < 0 && errno == EPIPE)
 		afb_ws_hangup(ws);
 }
@@ -393,18 +438,28 @@ static void aws_on_readable(struct afb_ws *ws)
  */
 static int aws_read(struct afb_ws *ws, size_t size)
 {
+	struct pollfd pfd;
 	ssize_t sz;
 	char *buffer;
 
-	if (size != 0) {
+	if (size != 0 || ws->buffer.buffer == NULL) {
 		buffer = realloc(ws->buffer.buffer, ws->buffer.size + size + 1);
 		if (buffer == NULL)
 			return 0;
 		ws->buffer.buffer = buffer;
-		sz = websock_read(ws->ws, &buffer[ws->buffer.size], size);
-		if ((size_t)sz != size)
-			return 0;
-		ws->buffer.size += size;
+		while (size != 0) {
+			sz = websock_read(ws->ws, &buffer[ws->buffer.size], size);
+			if (sz < 0) {
+				if (errno != EAGAIN)
+					return 0;
+				pfd.fd = ws->fd;
+				pfd.events = POLLIN;
+				poll(&pfd, 1, 10); /* TODO: make fully asynchronous websockets */
+			} else {
+				ws->buffer.size += (size_t)sz;
+				size -= (size_t)sz;
+			}
+		}
 	}
 	return 1;
 }
@@ -417,7 +472,7 @@ static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size)
 	struct buf b;
 
 	ws->state = waiting;
-	free(aws_pick_buffer(ws).buffer);
+	aws_clear_buffer(ws);
 	if (ws->itf->on_close == NULL) {
 		websock_drop(ws->ws);
 		afb_ws_hangup(ws);
@@ -435,7 +490,7 @@ static void aws_on_close(struct afb_ws *ws, uint16_t code, size_t size)
 static void aws_drop_error(struct afb_ws *ws, uint16_t code)
 {
 	ws->state = waiting;
-	free(aws_pick_buffer(ws).buffer);
+	aws_clear_buffer(ws);
 	websock_drop(ws->ws);
 	websock_error(ws->ws, code, NULL, 0);
 }
@@ -454,7 +509,6 @@ static void aws_continue(struct afb_ws *ws, int last, size_t size)
 		istxt = ws->state == reading_text;
 		ws->state = waiting;
 		b = aws_pick_buffer(ws);
-		b.buffer[b.size] = 0;
 		(istxt ? ws->itf->on_text : ws->itf->on_binary)(ws->closure, b.buffer, b.size);
 	}
 }
@@ -490,7 +544,7 @@ static void aws_on_binary(struct afb_ws *ws, int last, size_t size)
 }
 
 /*
- * Callback when 'close' command received from 'ws' with 'code' and 'size'.
+ * Callback when 'continue' command received from 'ws' with 'code' and 'size'.
  */
 static void aws_on_continue(struct afb_ws *ws, int last, size_t size)
 {