server/dispatcher: add dispatcher_register_async_done_callback

2 files changed, 29 insertions, 2 deletions

diff --git a/server/dispatcher.c b/server/dispatcher.c
index 81abbdf3..90e46e36 100644
--- a/server/dispatcher.c
+++ b/server/dispatcher.c
@@ -126,6 +126,9 @@ static int dispatcher_handle_single_read(Dispatcher *dispatcher)
             red_printf("error writing ack for message %d\n", type);
             /* TODO: close socketpair? */
         }
+    } else if (msg->ack == DISPATCHER_ASYNC && dispatcher->handle_async_done) {
+        dispatcher->handle_async_done(dispatcher->opaque, type,
+                                      (void *)payload);
     }
     return 1;
 }
@@ -174,9 +177,17 @@ unlock:
     pthread_mutex_unlock(&dispatcher->lock);
 }
 
+void dispatcher_register_async_done_callback(
+                                        Dispatcher *dispatcher,
+                                        dispatcher_handle_async_done handler)
+{
+    assert(dispatcher->handle_async_done == NULL);
+    dispatcher->handle_async_done = handler;
+}
+
 void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type,
-                                 dispatcher_handle_message handler, size_t size,
-                                 int ack)
+                                 dispatcher_handle_message handler,
+                                 size_t size, int ack)
 {
     DispatcherMessage *msg;
 
diff --git a/server/dispatcher.h b/server/dispatcher.h
index 95b6bfc6..a468c58f 100644
--- a/server/dispatcher.h
+++ b/server/dispatcher.h
@@ -8,6 +8,11 @@ typedef struct Dispatcher Dispatcher;
 typedef void (*dispatcher_handle_message)(void *opaque,
                                           void *payload);
 
+typedef void (*dispatcher_handle_async_done)(void *opaque,
+                                             uint32_t message_type,
+                                             void *payload);
+
+
 typedef struct DispatcherMessage {
     size_t size;
     int ack;
@@ -26,6 +31,7 @@ struct Dispatcher {
     void *payload; /* allocated as max of message sizes */
     size_t payload_size; /* used to track realloc calls */
     void *opaque;
+    dispatcher_handle_async_done handle_async_done;
 };
 
 /*
@@ -69,6 +75,16 @@ void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type,
                                  int ack);
 
 /*
+ * dispatcher_register_async_done_callback
+ * @dispatcher:     dispatcher
+ * @handler:        callback on the receiver side called *after* the
+ *                  message callback in case ack == DISPATCHER_ASYNC.
+ */
+void dispatcher_register_async_done_callback(
+                                    Dispatcher *dispatcher,
+                                    dispatcher_handle_async_done handler);
+
+/*
  *  dispatcher_handle_recv_read
  *  @dispatcher: Dispatcher instance
  */