Prefix replies on the wire with the packet size

1 files changed, 27 insertions, 0 deletions

diff --git a/proxy/src/gp_socket.c b/proxy/src/gp_socket.c
index 3c80142..ea8ae19 100644
--- a/proxy/src/gp_socket.c
+++ b/proxy/src/gp_socket.c
@@ -365,12 +365,39 @@ void gp_socket_send_data(verto_ctx *vctx, struct gp_conn *conn,
 static void gp_socket_write(verto_ctx *vctx, verto_ev *ev)
 {
     struct gp_buffer *wbuf;
+    uint32_t size;
     ssize_t wn;
     int fd;
 
     fd = verto_get_fd(ev);
     wbuf = verto_get_private(ev);
 
+    if (wbuf->pos == 0) {
+        /* first write, send the buffer size as packet header */
+        size = htonl(wbuf->size);
+
+        errno = 0;
+        wn = write(fd, &size, sizeof(uint32_t));
+        if (wn == -1) {
+            if (errno == EAGAIN || errno == EINTR) {
+                /* try again later */
+                gp_socket_schedule_write(vctx, wbuf);
+            } else {
+                /* error on socket, close and release it */
+                gp_conn_free(wbuf->conn);
+                gp_buffer_free(wbuf);
+            }
+            return;
+        }
+        if (wn != 4) {
+            /* don't bother trying to handle sockets that can't
+             * buffer even 4 bytes */
+            gp_conn_free(wbuf->conn);
+            gp_buffer_free(wbuf);
+            return;
+        }
+    }
+
     errno = 0;
     wn = write(fd, wbuf->data + wbuf->pos, wbuf->size - wbuf->pos);
     if (wn == -1) {