Move async code to RedsStream

The AsyncRead structure in reds.h wraps an async read + callback to
be done on a stream. Moving it to reds_stream.h is needed in order
to move SASL authentication there.

server/reds.c

@@ -121,15 +121,6 @@ static bool exit_on_disconnect = FALSE;
 
 static RedsState *reds = NULL;
 
-typedef struct AsyncRead {
-    RedsStream *stream;
-    void *opaque;
-    uint8_t *now;
-    uint8_t *end;
-    void (*done)(void *opaque);
-    void (*error)(void *opaque, int err);
-} AsyncRead;
-
 typedef struct RedLinkInfo {
     RedsStream *stream;
     AsyncRead async_read;
@@ -1880,57 +1871,6 @@ end:
     g_free(password);
 }
 
-static inline void async_read_clear_handlers(AsyncRead *obj)
-{
-    if (!obj->stream->watch) {
-        return;
-    }
-
-    reds_stream_remove_watch(obj->stream);
-}
-
-static void async_read_handler(int fd, int event, void *data)
-{
-    AsyncRead *obj = (AsyncRead *)data;
-
-    for (;;) {
-        int n = obj->end - obj->now;
-
-        spice_assert(n > 0);
-        n = reds_stream_read(obj->stream, obj->now, n);
-        if (n <= 0) {
-            if (n < 0) {
-                switch (errno) {
-                case EAGAIN:
-                    if (!obj->stream->watch) {
-                        obj->stream->watch = core->watch_add(obj->stream->socket,
-                                                           SPICE_WATCH_EVENT_READ,
-                                                           async_read_handler, obj);
-                    }
-                    return;
-                case EINTR:
-                    break;
-                default:
-                    async_read_clear_handlers(obj);
-                    obj->error(obj->opaque, errno);
-                    return;
-                }
-            } else {
-                async_read_clear_handlers(obj);
-                obj->error(obj->opaque, 0);
-                return;
-            }
-        } else {
-            obj->now += n;
-            if (obj->now == obj->end) {
-                async_read_clear_handlers(obj);
-                obj->done(obj->opaque);
-                return;
-            }
-        }
-    }
-}
-
 static void reds_get_spice_ticket(RedLinkInfo *link)
 {
     AsyncRead *obj = &link->async_read;

server/reds_stream.c

@@ -246,6 +246,8 @@ RedsStream *reds_stream_new(int socket)
     stream->write = stream_write_cb;
     stream->writev = stream_writev_cb;
 
+    stream->async_read.stream = stream;
+
     return stream;
 }
 
@@ -303,6 +305,68 @@ int reds_stream_enable_ssl(RedsStream *stream, SSL_CTX *ctx)
     return reds_stream_ssl_accept(stream);
 }
 
+void async_read_set_error_handler(AsyncRead *async,
+                                  AsyncReadError error_handler,
+                                 void *opaque)
+{
+    async->error = error_handler;
+}
+
+static inline void async_read_clear_handlers(AsyncRead *obj)
+{
+    if (!obj->stream->watch) {
+        return;
+    }
+
+    reds_stream_remove_watch(obj->stream);
+}
+
+void async_read_handler(int fd, int event, void *data)
+{
+    AsyncRead *obj = (AsyncRead *)data;
+
+    for (;;) {
+        int n = obj->end - obj->now;
+
+        spice_assert(n > 0);
+        n = reds_stream_read(obj->stream, obj->now, n);
+        if (n <= 0) {
+            if (n < 0) {
+                switch (errno) {
+                case EAGAIN:
+                    if (!obj->stream->watch) {
+                        obj->stream->watch = core->watch_add(obj->stream->socket,
+                                                           SPICE_WATCH_EVENT_READ,
+                                                           async_read_handler, obj);
+                    }
+                    return;
+                case EINTR:
+                    break;
+                default:
+                    async_read_clear_handlers(obj);
+		    if (obj->error) {
+                        obj->error(obj->opaque, errno);
+		    }
+                    return;
+                }
+            } else {
+                async_read_clear_handlers(obj);
+		if (obj->error) {
+		    obj->error(obj->opaque, 0);
+		}
+                return;
+            }
+        } else {
+            obj->now += n;
+            if (obj->now == obj->end) {
+                async_read_clear_handlers(obj);
+                obj->done(obj->opaque);
+                return;
+            }
+        }
+    }
+}
+
 #if HAVE_SASL
 bool reds_stream_write_u8(RedsStream *s, uint8_t n)
 {

server/reds_stream.h

@@ -28,6 +28,9 @@
 #if HAVE_SASL
 #include <sasl/sasl.h>
 
+typedef void (*AsyncReadDone)(void *opaque);
+typedef void (*AsyncReadError)(void *opaque, int err);
+
 typedef struct RedsSASL {
     sasl_conn_t *conn;
 
@@ -57,6 +60,19 @@ typedef struct RedsSASL {
 #endif
 
 typedef struct RedsStream RedsStream;
+typedef struct AsyncRead {
+    RedsStream *stream;
+    void *opaque;
+    uint8_t *now;
+    uint8_t *end;
+    AsyncReadDone done;
+    AsyncReadError error;
+} AsyncRead;
+
+void async_read_handler(int fd, int event, void *data);
+void async_read_set_error_handler(AsyncRead *async,
+                                  AsyncReadError error_handler,
+                                  void *opaque);
 
 struct RedsStream {
     int socket;
@@ -67,6 +83,8 @@ struct RedsStream {
     int shutdown;
     SSL *ssl;
 
+    AsyncRead async_read;
+
 #if HAVE_SASL
     RedsSASL sasl;
 #endif