Convert Dispatcher and MainDispatcher to GObjects

Allows more explicit inheritance relationship, and numerous other
advantages.

Acked-by: Christophe Fergeau <cfergeau@redhat.com>

5 files changed, 399 insertions, 136 deletions

diff --git a/server/dispatcher.c b/server/dispatcher.c
index d6c03ca7..7e71ea2d 100644
--- a/server/dispatcher.c
+++ b/server/dispatcher.c
@@ -1,6 +1,5 @@
-/* -*- Mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
 /*
-   Copyright (C) 2009-2012 Red Hat, Inc.
+   Copyright (C) 2009-2016 Red Hat, Inc.
 
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
@@ -39,6 +38,156 @@
 #include <signal.h>
 #endif
 
+G_DEFINE_TYPE(Dispatcher, dispatcher, G_TYPE_OBJECT)
+
+#define DISPATCHER_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), TYPE_DISPATCHER, DispatcherPrivate))
+
+struct DispatcherPrivate {
+    int recv_fd;
+    int send_fd;
+    pthread_t self;
+    pthread_mutex_t lock;
+    DispatcherMessage *messages;
+    int stage;  /* message parser stage - sender has no stages */
+    guint max_message_type;
+    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;
+    dispatcher_handle_any_message any_handler;
+};
+
+enum {
+    PROP_0,
+    PROP_MAX_MESSAGE_TYPE,
+    PROP_OPAQUE
+};
+
+static void
+dispatcher_get_property(GObject    *object,
+                        guint       property_id,
+                        GValue     *value,
+                        GParamSpec *pspec)
+{
+    Dispatcher *self = DISPATCHER(object);
+
+    switch (property_id)
+    {
+        case PROP_MAX_MESSAGE_TYPE:
+            g_value_set_uint(value, self->priv->max_message_type);
+            break;
+        case PROP_OPAQUE:
+            g_value_set_pointer(value, self->priv->opaque);
+            break;
+        default:
+            G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
+    }
+}
+
+static void
+dispatcher_set_property(GObject      *object,
+                        guint         property_id,
+                        const GValue *value,
+                        GParamSpec   *pspec)
+{
+    Dispatcher *self = DISPATCHER(object);
+
+    switch (property_id)
+    {
+        case PROP_MAX_MESSAGE_TYPE:
+            self->priv->max_message_type = g_value_get_uint(value);
+            break;
+        case PROP_OPAQUE:
+            dispatcher_set_opaque(self, g_value_get_pointer(value));
+            break;
+        default:
+            G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
+    }
+}
+
+static void
+dispatcher_finalize(GObject *object)
+{
+    Dispatcher *self = DISPATCHER(object);
+    g_free(self->priv->messages);
+    close(self->priv->send_fd);
+    close(self->priv->recv_fd);
+    pthread_mutex_destroy(&self->priv->lock);
+    free(self->priv->payload);
+    G_OBJECT_CLASS(dispatcher_parent_class)->finalize(object);
+}
+
+static void dispatcher_constructed(GObject *object)
+{
+    Dispatcher *self = DISPATCHER(object);
+    int channels[2];
+
+    G_OBJECT_CLASS(dispatcher_parent_class)->constructed(object);
+
+#ifdef DEBUG_DISPATCHER
+    setup_dummy_signal_handler();
+#endif
+    if (socketpair(AF_LOCAL, SOCK_STREAM, 0, channels) == -1) {
+        spice_error("socketpair failed %s", strerror(errno));
+        return;
+    }
+    pthread_mutex_init(&self->priv->lock, NULL);
+    self->priv->recv_fd = channels[0];
+    self->priv->send_fd = channels[1];
+    self->priv->self = pthread_self();
+
+    self->priv->messages = g_new0(DispatcherMessage,
+                                  self->priv->max_message_type);
+}
+
+static void
+dispatcher_class_init(DispatcherClass *klass)
+{
+    GObjectClass *object_class = G_OBJECT_CLASS(klass);
+
+    g_type_class_add_private(klass, sizeof (DispatcherPrivate));
+
+    object_class->get_property = dispatcher_get_property;
+    object_class->set_property = dispatcher_set_property;
+    object_class->constructed = dispatcher_constructed;
+    object_class->finalize = dispatcher_finalize;
+
+    g_object_class_install_property(object_class,
+                                    PROP_MAX_MESSAGE_TYPE,
+                                    g_param_spec_uint("max-message-type",
+                                                      "max-message-type",
+                                                      "Maximum message type",
+                                                      0, G_MAXUINT, 0,
+                                                      G_PARAM_STATIC_STRINGS |
+                                                      G_PARAM_READWRITE |
+                                                      G_PARAM_CONSTRUCT_ONLY));
+    g_object_class_install_property(object_class,
+                                    PROP_OPAQUE,
+                                    g_param_spec_pointer("opaque",
+                                                         "opaque",
+                                                         "User data to pass to callbacks",
+                                                         G_PARAM_STATIC_STRINGS |
+                                                         G_PARAM_READWRITE |
+                                                         G_PARAM_CONSTRUCT));
+
+}
+
+static void
+dispatcher_init(Dispatcher *self)
+{
+    self->priv = DISPATCHER_PRIVATE(self);
+}
+
+Dispatcher *
+dispatcher_new(size_t max_message_type, void *opaque)
+{
+    return g_object_new(TYPE_DISPATCHER,
+                        "max-message-type", (guint) max_message_type,
+                        "opaque", opaque,
+                        NULL);
+}
+
+
 #define ACK 0xffffffff
 
 /*
@@ -118,10 +267,10 @@ static int dispatcher_handle_single_read(Dispatcher *dispatcher)
     int ret;
     uint32_t type;
     DispatcherMessage *msg = NULL;
-    uint8_t *payload = dispatcher->payload;
+    uint8_t *payload = dispatcher->priv->payload;
     uint32_t ack = ACK;
 
-    if ((ret = read_safe(dispatcher->recv_fd, (uint8_t*)&type, sizeof(type), 0)) == -1) {
+    if ((ret = read_safe(dispatcher->priv->recv_fd, (uint8_t*)&type, sizeof(type), 0)) == -1) {
         spice_printerr("error reading from dispatcher: %d", errno);
         return 0;
     }
@@ -129,28 +278,28 @@ static int dispatcher_handle_single_read(Dispatcher *dispatcher)
         /* no messsage */
         return 0;
     }
-    msg = &dispatcher->messages[type];
-    if (read_safe(dispatcher->recv_fd, payload, msg->size, 1) == -1) {
+    msg = &dispatcher->priv->messages[type];
+    if (read_safe(dispatcher->priv->recv_fd, payload, msg->size, 1) == -1) {
         spice_printerr("error reading from dispatcher: %d", errno);
         /* TODO: close socketpair? */
         return 0;
     }
-    if (dispatcher->any_handler) {
-        dispatcher->any_handler(dispatcher->opaque, type, payload);
+    if (dispatcher->priv->any_handler) {
+        dispatcher->priv->any_handler(dispatcher->priv->opaque, type, payload);
     }
     if (msg->handler) {
-        msg->handler(dispatcher->opaque, (void *)payload);
+        msg->handler(dispatcher->priv->opaque, payload);
     } else {
         spice_printerr("error: no handler for message type %d", type);
     }
     if (msg->ack == DISPATCHER_ACK) {
-        if (write_safe(dispatcher->recv_fd,
+        if (write_safe(dispatcher->priv->recv_fd,
                        (uint8_t*)&ack, sizeof(ack)) == -1) {
             spice_printerr("error writing ack for message %d", type);
             /* TODO: close socketpair? */
         }
-    } else if (msg->ack == DISPATCHER_ASYNC && dispatcher->handle_async_done) {
-        dispatcher->handle_async_done(dispatcher->opaque, type,
+    } else if (msg->ack == DISPATCHER_ASYNC && dispatcher->priv->handle_async_done) {
+        dispatcher->priv->handle_async_done(dispatcher->priv->opaque, type,
                                       (void *)payload);
     }
     return 1;
@@ -171,12 +320,12 @@ void dispatcher_send_message(Dispatcher *dispatcher, uint32_t message_type,
 {
     DispatcherMessage *msg;
     uint32_t ack;
-    int send_fd = dispatcher->send_fd;
+    int send_fd = dispatcher->priv->send_fd;
 
-    assert(dispatcher->max_message_type > message_type);
-    assert(dispatcher->messages[message_type].handler);
-    msg = &dispatcher->messages[message_type];
-    pthread_mutex_lock(&dispatcher->lock);
+    assert(dispatcher->priv->max_message_type > message_type);
+    assert(dispatcher->priv->messages[message_type].handler);
+    msg = &dispatcher->priv->messages[message_type];
+    pthread_mutex_lock(&dispatcher->priv->lock);
     if (write_safe(send_fd, (uint8_t*)&message_type, sizeof(message_type)) == -1) {
         spice_printerr("error: failed to send message type for message %d",
                    message_type);
@@ -197,15 +346,15 @@ void dispatcher_send_message(Dispatcher *dispatcher, uint32_t message_type,
         }
     }
 unlock:
-    pthread_mutex_unlock(&dispatcher->lock);
+    pthread_mutex_unlock(&dispatcher->priv->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;
+    assert(dispatcher->priv->handle_async_done == NULL);
+    dispatcher->priv->handle_async_done = handler;
 }
 
 void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type,
@@ -214,15 +363,15 @@ void dispatcher_register_handler(Dispatcher *dispatcher, uint32_t message_type,
 {
     DispatcherMessage *msg;
 
-    assert(message_type < dispatcher->max_message_type);
-    assert(dispatcher->messages[message_type].handler == 0);
-    msg = &dispatcher->messages[message_type];
+    assert(message_type < dispatcher->priv->max_message_type);
+    assert(dispatcher->priv->messages[message_type].handler == 0);
+    msg = &dispatcher->priv->messages[message_type];
     msg->handler = handler;
     msg->size = size;
     msg->ack = ack;
-    if (msg->size > dispatcher->payload_size) {
-        dispatcher->payload = realloc(dispatcher->payload, msg->size);
-        dispatcher->payload_size = msg->size;
+    if (msg->size > dispatcher->priv->payload_size) {
+        dispatcher->priv->payload = realloc(dispatcher->priv->payload, msg->size);
+        dispatcher->priv->payload_size = msg->size;
     }
 }
 
@@ -230,7 +379,7 @@ void dispatcher_register_universal_handler(
                                Dispatcher *dispatcher,
                                dispatcher_handle_any_message any_handler)
 {
-    dispatcher->any_handler = any_handler;
+    dispatcher->priv->any_handler = any_handler;
 }
 
 #ifdef DEBUG_DISPATCHER
@@ -257,35 +406,18 @@ static void setup_dummy_signal_handler(void)
 }
 #endif
 
-void dispatcher_init(Dispatcher *dispatcher, size_t max_message_type,
-                     void *opaque)
+void dispatcher_set_opaque(Dispatcher *self, void *opaque)
 {
-    int channels[2];
-
-#ifdef DEBUG_DISPATCHER
-    setup_dummy_signal_handler();
-#endif
-    dispatcher->opaque = opaque;
-    if (socketpair(AF_LOCAL, SOCK_STREAM, 0, channels) == -1) {
-        spice_error("socketpair failed %s", strerror(errno));
-        return;
-    }
-    pthread_mutex_init(&dispatcher->lock, NULL);
-    dispatcher->recv_fd = channels[0];
-    dispatcher->send_fd = channels[1];
-    dispatcher->self = pthread_self();
-
-    dispatcher->messages = spice_malloc0_n(max_message_type,
-                                           sizeof(dispatcher->messages[0]));
-    dispatcher->max_message_type = max_message_type;
+    self->priv->opaque = opaque;
+    g_object_notify(G_OBJECT(self), "opaque");
 }
 
-void dispatcher_set_opaque(Dispatcher *dispatcher, void *opaque)
+int dispatcher_get_recv_fd(Dispatcher *dispatcher)
 {
-    dispatcher->opaque = opaque;
+    return dispatcher->priv->recv_fd;
 }
 
-int dispatcher_get_recv_fd(Dispatcher *dispatcher)
+pthread_t dispatcher_get_thread_id(Dispatcher *self)
 {
-    return dispatcher->recv_fd;
+    return self->priv->self;
 }
diff --git a/server/dispatcher.h b/server/dispatcher.h
index 78ef663a..caaebc5a 100644
--- a/server/dispatcher.h
+++ b/server/dispatcher.h
@@ -18,9 +18,37 @@
 #ifndef DISPATCHER_H
 #define DISPATCHER_H
 
+#include <glib-object.h>
 #include "red-common.h"
 
+#define TYPE_DISPATCHER dispatcher_get_type()
+
+#define DISPATCHER(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), TYPE_DISPATCHER, Dispatcher))
+#define DISPATCHER_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), TYPE_DISPATCHER, DispatcherClass))
+#define IS_DISPATCHER(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), TYPE_DISPATCHER))
+#define IS_DISPATCHER_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), TYPE_DISPATCHER))
+#define DISPATCHER_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS((obj), TYPE_DISPATCHER, DispatcherClass))
+
 typedef struct Dispatcher Dispatcher;
+typedef struct DispatcherClass DispatcherClass;
+typedef struct DispatcherPrivate DispatcherPrivate;
+
+struct Dispatcher
+{
+    GObject parent;
+
+    DispatcherPrivate *priv;
+};
+
+struct DispatcherClass
+{
+    GObjectClass parent_class;
+};
+
+GType dispatcher_get_type(void) G_GNUC_CONST;
+
+Dispatcher *dispatcher_new(size_t max_message_type, void *opaque);
+
 
 typedef void (*dispatcher_handle_message)(void *opaque,
                                           void *payload);
@@ -40,20 +68,6 @@ typedef struct DispatcherMessage {
     dispatcher_handle_message handler;
 } DispatcherMessage;
 
-struct Dispatcher {
-    int recv_fd;
-    int send_fd;
-    pthread_t self;
-    pthread_mutex_t lock;
-    DispatcherMessage *messages;
-    int stage;  /* message parser stage - sender has no stages */
-    size_t max_message_type;
-    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;
-    dispatcher_handle_any_message any_handler;
-};
 
 /*
  * dispatcher_send_message
@@ -63,15 +77,6 @@ struct Dispatcher {
 void dispatcher_send_message(Dispatcher *dispatcher, uint32_t message_type,
                              void *payload);
 
-/*
- * dispatcher_init
- * @max_message_type: number of message types. Allows upfront allocation
- *  of a DispatcherMessage list.
- * up front, and registration in any order wanted.
- */
-void dispatcher_init(Dispatcher *dispatcher, size_t max_message_type,
-                     void *opaque);
-
 enum {
     DISPATCHER_NONE = 0,
     DISPATCHER_ACK,
@@ -131,4 +136,6 @@ int dispatcher_get_recv_fd(Dispatcher *);
  */
 void dispatcher_set_opaque(Dispatcher *dispatcher, void *opaque);
 
+pthread_t dispatcher_get_thread_id(Dispatcher *self);
+
 #endif //DISPATCHER_H
diff --git a/server/main-dispatcher.c b/server/main-dispatcher.c
index 298a961a..bc0de241 100644
--- a/server/main-dispatcher.c
+++ b/server/main-dispatcher.c
@@ -47,12 +47,99 @@
  *   seperate from self because it may send an ack or do other work in the future.
  */
 
-struct MainDispatcher {
-    Dispatcher base;
-    SpiceCoreInterfaceInternal *core;
-    RedsState *reds;
+G_DEFINE_TYPE(MainDispatcher, main_dispatcher, TYPE_DISPATCHER)
+
+#define MAIN_DISPATCHER_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE((o), TYPE_MAIN_DISPATCHER, MainDispatcherPrivate))
+
+struct MainDispatcherPrivate
+{
+    SpiceCoreInterfaceInternal *core; /* weak */
+    RedsState *reds; /* weak */
 };
 
+
+enum {
+    PROP0,
+    PROP_SPICE_SERVER,
+    PROP_CORE_INTERFACE
+};
+
+static void
+main_dispatcher_get_property(GObject    *object,
+                                  guint       property_id,
+                                  GValue     *value,
+                                  GParamSpec *pspec)
+{
+    MainDispatcher *self = MAIN_DISPATCHER(object);
+
+    switch (property_id) {
+        case PROP_SPICE_SERVER:
+             g_value_set_pointer(value, self->priv->reds);
+            break;
+        case PROP_CORE_INTERFACE:
+             g_value_set_pointer(value, self->priv->core);
+            break;
+        default:
+            G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
+    }
+}
+
+static void
+main_dispatcher_set_property(GObject      *object,
+                                  guint         property_id,
+                                  const GValue *value,
+                                  GParamSpec   *pspec)
+{
+    MainDispatcher *self = MAIN_DISPATCHER(object);
+
+    switch (property_id) {
+        case PROP_SPICE_SERVER:
+            self->priv->reds = g_value_get_pointer(value);
+            break;
+        case PROP_CORE_INTERFACE:
+            self->priv->core = g_value_get_pointer(value);
+            break;
+        default:
+            G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
+    }
+}
+
+static void main_dispatcher_constructed(GObject *object);
+
+static void
+main_dispatcher_class_init(MainDispatcherClass *klass)
+{
+    GObjectClass *object_class = G_OBJECT_CLASS(klass);
+
+    g_type_class_add_private(klass, sizeof(MainDispatcherPrivate));
+
+    object_class->constructed = main_dispatcher_constructed;
+    object_class->get_property = main_dispatcher_get_property;
+    object_class->set_property = main_dispatcher_set_property;
+
+    g_object_class_install_property(object_class,
+                                    PROP_SPICE_SERVER,
+                                    g_param_spec_pointer("spice-server",
+                                                         "spice-server",
+                                                         "The spice server associated with this dispatcher",
+                                                         G_PARAM_READWRITE |
+                                                         G_PARAM_CONSTRUCT_ONLY));
+
+    g_object_class_install_property(object_class,
+                                    PROP_CORE_INTERFACE,
+                                    g_param_spec_pointer("core-interface",
+                                                         "core-interface",
+                                                         "The SpiceCoreInterface server associated with this dispatcher",
+                                                         G_PARAM_READWRITE |
+                                                         G_PARAM_CONSTRUCT_ONLY));
+}
+
+static void
+main_dispatcher_init(MainDispatcher *self)
+{
+    self->priv = MAIN_DISPATCHER_PRIVATE(self);
+}
+
 enum {
     MAIN_DISPATCHER_CHANNEL_EVENT = 0,
     MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
@@ -85,7 +172,7 @@ static void main_dispatcher_self_handle_channel_event(MainDispatcher *self,
                                                       int event,
                                                       SpiceChannelEventInfo *info)
 {
-    reds_handle_channel_event(self->reds, event, info);
+    reds_handle_channel_event(self->priv->reds, event, info);
 }
 
 static void main_dispatcher_handle_channel_event(void *opaque,
@@ -103,13 +190,13 @@ void main_dispatcher_channel_event(MainDispatcher *self, int event, SpiceChannel
 {
     MainDispatcherChannelEventMessage msg = {0,};
 
-    if (pthread_self() == self->base.self) {
+    if (pthread_self() == dispatcher_get_thread_id(DISPATCHER(self))) {
         main_dispatcher_self_handle_channel_event(self, event, info);
         return;
     }
     msg.event = event;
     msg.info = info;
-    dispatcher_send_message(&self->base, MAIN_DISPATCHER_CHANNEL_EVENT,
+    dispatcher_send_message(DISPATCHER(self), MAIN_DISPATCHER_CHANNEL_EVENT,
                             &msg);
 }
 
@@ -120,7 +207,7 @@ static void main_dispatcher_handle_migrate_complete(void *opaque,
     MainDispatcher *self = opaque;
     MainDispatcherMigrateSeamlessDstCompleteMessage *mig_complete = payload;
 
-    reds_on_client_seamless_migrate_complete(self->reds, mig_complete->client);
+    reds_on_client_seamless_migrate_complete(self->priv->reds, mig_complete->client);
     red_client_unref(mig_complete->client);
 }
 
@@ -129,7 +216,7 @@ static void main_dispatcher_handle_mm_time_latency(void *opaque,
 {
     MainDispatcher *self = opaque;
     MainDispatcherMmTimeLatencyMessage *msg = payload;
-    reds_set_client_mm_time_latency(self->reds, msg->client, msg->latency);
+    reds_set_client_mm_time_latency(self->priv->reds, msg->client, msg->latency);
     red_client_unref(msg->client);
 }
 
@@ -140,7 +227,7 @@ static void main_dispatcher_handle_client_disconnect(void *opaque,
     MainDispatcherClientDisconnectMessage *msg = payload;
 
     spice_debug("client=%p", msg->client);
-    reds_client_disconnect(self->reds, msg->client);
+    reds_client_disconnect(self->priv->reds, msg->client);
     red_client_unref(msg->client);
 }
 
@@ -149,13 +236,13 @@ void main_dispatcher_seamless_migrate_dst_complete(MainDispatcher *self,
 {
     MainDispatcherMigrateSeamlessDstCompleteMessage msg;
 
-    if (pthread_self() == self->base.self) {
-        reds_on_client_seamless_migrate_complete(self->reds, client);
+    if (pthread_self() == dispatcher_get_thread_id(DISPATCHER(self))) {
+        reds_on_client_seamless_migrate_complete(self->priv->reds, client);
         return;
     }
 
     msg.client = red_client_ref(client);
-    dispatcher_send_message(&self->base, MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
+    dispatcher_send_message(DISPATCHER(self), MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
                             &msg);
 }
 
@@ -163,14 +250,14 @@ void main_dispatcher_set_mm_time_latency(MainDispatcher *self, RedClient *client
 {
     MainDispatcherMmTimeLatencyMessage msg;
 
-    if (pthread_self() == self->base.self) {
-        reds_set_client_mm_time_latency(self->reds, client, latency);
+    if (pthread_self() == dispatcher_get_thread_id(DISPATCHER(self))) {
+        reds_set_client_mm_time_latency(self->priv->reds, client, latency);
         return;
     }
 
     msg.client = red_client_ref(client);
     msg.latency = latency;
-    dispatcher_send_message(&self->base, MAIN_DISPATCHER_SET_MM_TIME_LATENCY,
+    dispatcher_send_message(DISPATCHER(self), MAIN_DISPATCHER_SET_MM_TIME_LATENCY,
                             &msg);
 }
 
@@ -181,7 +268,7 @@ void main_dispatcher_client_disconnect(MainDispatcher *self, RedClient *client)
     if (!client->disconnecting) {
         spice_debug("client %p", client);
         msg.client = red_client_ref(client);
-        dispatcher_send_message(&self->base, MAIN_DISPATCHER_CLIENT_DISCONNECT,
+        dispatcher_send_message(DISPATCHER(self), MAIN_DISPATCHER_CLIENT_DISCONNECT,
                                 &msg);
     } else {
         spice_debug("client %p already during disconnection", client);
@@ -192,7 +279,7 @@ static void dispatcher_handle_read(int fd, int event, void *opaque)
 {
     MainDispatcher *self = opaque;
 
-    dispatcher_handle_recv_read(&self->base);
+    dispatcher_handle_recv_read(DISPATCHER(self));
 }
 
 /*
@@ -202,23 +289,35 @@ static void dispatcher_handle_read(int fd, int event, void *opaque)
  */
 MainDispatcher* main_dispatcher_new(RedsState *reds, SpiceCoreInterfaceInternal *core)
 {
-    MainDispatcher *main_dispatcher = g_new0(MainDispatcher, 1);
-    main_dispatcher->core = core;
-    main_dispatcher->reds = reds;
-    dispatcher_init(&main_dispatcher->base, MAIN_DISPATCHER_NUM_MESSAGES, main_dispatcher);
-    core->watch_add(core, main_dispatcher->base.recv_fd, SPICE_WATCH_EVENT_READ,
-                    dispatcher_handle_read, main_dispatcher);
-    dispatcher_register_handler(&main_dispatcher->base, MAIN_DISPATCHER_CHANNEL_EVENT,
+    MainDispatcher *self = g_object_new(TYPE_MAIN_DISPATCHER,
+                                        "spice-server", reds,
+                                        "core-interface", core,
+                                        "max-message-type", MAIN_DISPATCHER_NUM_MESSAGES,
+                                        NULL);
+    return self;
+}
+
+void main_dispatcher_constructed(GObject *object)
+{
+    MainDispatcher *self = MAIN_DISPATCHER(object);
+
+    G_OBJECT_CLASS(main_dispatcher_parent_class)->constructed(object);
+    dispatcher_set_opaque(DISPATCHER(self), self);
+
+    self->priv->core->watch_add(self->priv->core,
+                                dispatcher_get_recv_fd(DISPATCHER(self)),
+                                SPICE_WATCH_EVENT_READ, dispatcher_handle_read,
+                                self);
+    dispatcher_register_handler(DISPATCHER(self), MAIN_DISPATCHER_CHANNEL_EVENT,
                                 main_dispatcher_handle_channel_event,
                                 sizeof(MainDispatcherChannelEventMessage), 0 /* no ack */);
-    dispatcher_register_handler(&main_dispatcher->base, MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
+    dispatcher_register_handler(DISPATCHER(self), MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
                                 main_dispatcher_handle_migrate_complete,
                                 sizeof(MainDispatcherMigrateSeamlessDstCompleteMessage), 0 /* no ack */);
-    dispatcher_register_handler(&main_dispatcher->base, MAIN_DISPATCHER_SET_MM_TIME_LATENCY,
+    dispatcher_register_handler(DISPATCHER(self), MAIN_DISPATCHER_SET_MM_TIME_LATENCY,
                                 main_dispatcher_handle_mm_time_latency,
                                 sizeof(MainDispatcherMmTimeLatencyMessage), 0 /* no ack */);
-    dispatcher_register_handler(&main_dispatcher->base, MAIN_DISPATCHER_CLIENT_DISCONNECT,
+    dispatcher_register_handler(DISPATCHER(self), MAIN_DISPATCHER_CLIENT_DISCONNECT,
                                 main_dispatcher_handle_client_disconnect,
                                 sizeof(MainDispatcherClientDisconnectMessage), 0 /* no ack */);
-    return main_dispatcher;
 }
diff --git a/server/main-dispatcher.h b/server/main-dispatcher.h
index cbc3657a..ffe82c89 100644
--- a/server/main-dispatcher.h
+++ b/server/main-dispatcher.h
@@ -19,9 +19,34 @@
 #define MAIN_DISPATCHER_H
 
 #include <spice.h>
+#include "dispatcher.h"
 #include "red-channel.h"
 
+#define TYPE_MAIN_DISPATCHER main_dispatcher_get_type()
+
+#define MAIN_DISPATCHER(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), TYPE_MAIN_DISPATCHER, MainDispatcher))
+#define MAIN_DISPATCHER_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), TYPE_MAIN_DISPATCHER, MainDispatcherClass))
+#define IS_MAIN_DISPATCHER(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), TYPE_MAIN_DISPATCHER))
+#define IS_MAIN_DISPATCHER_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), TYPE_MAIN_DISPATCHER))
+#define MAIN_DISPATCHER_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS((obj), TYPE_MAIN_DISPATCHER, MainDispatcherClass))
+
 typedef struct MainDispatcher MainDispatcher;
+typedef struct MainDispatcherClass MainDispatcherClass;
+typedef struct MainDispatcherPrivate MainDispatcherPrivate;
+
+struct MainDispatcher
+{
+    Dispatcher parent;
+
+    MainDispatcherPrivate *priv;
+};
+
+struct MainDispatcherClass
+{
+    DispatcherClass parent_class;
+};
+
+GType main_dispatcher_get_type(void) G_GNUC_CONST;
 
 void main_dispatcher_channel_event(MainDispatcher *self, int event, SpiceChannelEventInfo *info);
 void main_dispatcher_seamless_migrate_dst_complete(MainDispatcher *self, RedClient *client);
diff --git a/server/red-qxl.c b/server/red-qxl.c
index 7015144a..abde0bab 100644
--- a/server/red-qxl.c
+++ b/server/red-qxl.c
@@ -48,7 +48,7 @@ struct AsyncCommand {
 struct QXLState {
     QXLWorker qxl_worker;
     QXLInstance *qxl;
-    Dispatcher dispatcher;
+    Dispatcher *dispatcher;
     uint32_t pending;
     int primary_active;
     int x_res;
@@ -203,7 +203,7 @@ static void red_qxl_update_area(QXLState *qxl_state, uint32_t surface_id,
     payload.qxl_dirty_rects = qxl_dirty_rects;
     payload.num_dirty_rects = num_dirty_rects;
     payload.clear_dirty_region = clear_dirty_region;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_UPDATE,
                             &payload);
 }
@@ -248,7 +248,7 @@ static void red_qxl_update_area_async(QXLState *qxl_state,
     payload.surface_id = surface_id;
     payload.qxl_area = *qxl_area;
     payload.clear_dirty_region = clear_dirty_region;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             message,
                             &payload);
 }
@@ -267,7 +267,7 @@ static void red_qxl_add_memslot(QXLState *qxl_state, QXLDevMemSlot *mem_slot)
     RedWorkerMessageAddMemslot payload;
 
     payload.mem_slot = *mem_slot;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_ADD_MEMSLOT,
                             &payload);
 }
@@ -285,7 +285,7 @@ static void red_qxl_add_memslot_async(QXLState *qxl_state, QXLDevMemSlot *mem_sl
 
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
     payload.mem_slot = *mem_slot;
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void red_qxl_del_memslot(QXLState *qxl_state, uint32_t slot_group_id, uint32_t slot_id)
@@ -295,7 +295,7 @@ static void red_qxl_del_memslot(QXLState *qxl_state, uint32_t slot_group_id, uin
 
     payload.slot_group_id = slot_group_id;
     payload.slot_id = slot_id;
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void qxl_worker_del_memslot(QXLWorker *qxl_worker, uint32_t slot_group_id, uint32_t slot_id)
@@ -308,7 +308,7 @@ static void red_qxl_destroy_surfaces(QXLState *qxl_state)
 {
     RedWorkerMessageDestroySurfaces payload;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_DESTROY_SURFACES,
                             &payload);
 }
@@ -325,7 +325,7 @@ static void red_qxl_destroy_surfaces_async(QXLState *qxl_state, uint64_t cookie)
     RedWorkerMessage message = RED_WORKER_MESSAGE_DESTROY_SURFACES_ASYNC;
 
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void red_qxl_destroy_primary_surface_complete(QXLState *qxl_state)
@@ -344,7 +344,7 @@ red_qxl_destroy_primary_surface_sync(QXLState *qxl_state,
 {
     RedWorkerMessageDestroyPrimarySurface payload;
     payload.surface_id = surface_id;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_DESTROY_PRIMARY_SURFACE,
                             &payload);
     red_qxl_destroy_primary_surface_complete(qxl_state);
@@ -359,7 +359,7 @@ red_qxl_destroy_primary_surface_async(QXLState *qxl_state,
 
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
     payload.surface_id = surface_id;
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void
@@ -403,7 +403,7 @@ red_qxl_create_primary_surface_async(QXLState *qxl_state, uint32_t surface_id,
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
     payload.surface_id = surface_id;
     payload.surface = *surface;
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void
@@ -415,7 +415,7 @@ red_qxl_create_primary_surface_sync(QXLState *qxl_state, uint32_t surface_id,
     qxl_state->surface_create = *surface;
     payload.surface_id = surface_id;
     payload.surface = *surface;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_CREATE_PRIMARY_SURFACE,
                             &payload);
     red_qxl_create_primary_surface_complete(qxl_state);
@@ -443,7 +443,7 @@ static void red_qxl_reset_image_cache(QXLState *qxl_state)
 {
     RedWorkerMessageResetImageCache payload;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_RESET_IMAGE_CACHE,
                             &payload);
 }
@@ -458,7 +458,7 @@ static void red_qxl_reset_cursor(QXLState *qxl_state)
 {
     RedWorkerMessageResetCursor payload;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_RESET_CURSOR,
                             &payload);
 }
@@ -475,7 +475,7 @@ static void red_qxl_destroy_surface_wait_sync(QXLState *qxl_state,
     RedWorkerMessageDestroySurfaceWait payload;
 
     payload.surface_id = surface_id;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_DESTROY_SURFACE_WAIT,
                             &payload);
 }
@@ -489,7 +489,7 @@ static void red_qxl_destroy_surface_wait_async(QXLState *qxl_state,
 
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
     payload.surface_id = surface_id;
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void red_qxl_destroy_surface_wait(QXLState *qxl_state,
@@ -513,7 +513,7 @@ static void red_qxl_reset_memslots(QXLState *qxl_state)
 {
     RedWorkerMessageResetMemslots payload;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_RESET_MEMSLOTS,
                             &payload);
 }
@@ -542,7 +542,7 @@ static void red_qxl_wakeup(QXLState *qxl_state)
     if (red_qxl_set_pending(qxl_state, RED_DISPATCHER_PENDING_WAKEUP))
         return;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_WAKEUP,
                             &payload);
 }
@@ -560,7 +560,7 @@ static void red_qxl_oom(QXLState *qxl_state)
     if (red_qxl_set_pending(qxl_state, RED_DISPATCHER_PENDING_OOM))
         return;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_OOM,
                             &payload);
 }
@@ -575,7 +575,7 @@ void red_qxl_start(QXLInstance *qxl)
 {
     RedWorkerMessageStart payload;
 
-    dispatcher_send_message(&qxl->st->dispatcher,
+    dispatcher_send_message(qxl->st->dispatcher,
                             RED_WORKER_MESSAGE_START,
                             &payload);
 }
@@ -592,7 +592,7 @@ static void red_qxl_flush_surfaces_async(QXLState *qxl_state, uint64_t cookie)
     RedWorkerMessage message = RED_WORKER_MESSAGE_FLUSH_SURFACES_ASYNC;
 
     payload.base.cmd = async_command_alloc(qxl_state, message, cookie);
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void red_qxl_monitors_config_async(QXLState *qxl_state,
@@ -608,14 +608,14 @@ static void red_qxl_monitors_config_async(QXLState *qxl_state,
     payload.group_id = group_id;
     payload.max_monitors = qxl_state->max_monitors;
 
-    dispatcher_send_message(&qxl_state->dispatcher, message, &payload);
+    dispatcher_send_message(qxl_state->dispatcher, message, &payload);
 }
 
 static void red_qxl_driver_unload(QXLState *qxl_state)
 {
     RedWorkerMessageDriverUnload payload;
 
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_DRIVER_UNLOAD,
                             &payload);
 }
@@ -624,7 +624,7 @@ void red_qxl_stop(QXLInstance *qxl)
 {
     RedWorkerMessageStop payload;
 
-    dispatcher_send_message(&qxl->st->dispatcher,
+    dispatcher_send_message(qxl->st->dispatcher,
                             RED_WORKER_MESSAGE_STOP,
                             &payload);
 }
@@ -644,7 +644,7 @@ static void red_qxl_loadvm_commands(QXLState *qxl_state,
     spice_printerr("");
     payload.count = count;
     payload.ext = ext;
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_LOADVM_COMMANDS,
                             &payload);
 }
@@ -884,7 +884,7 @@ void spice_qxl_gl_scanout(QXLInstance *qxl,
     pthread_mutex_unlock(&qxl_state->scanout_mutex);
 
     /* FIXME: find a way to coallesce all pending SCANOUTs */
-    dispatcher_send_message(&qxl_state->dispatcher,
+    dispatcher_send_message(qxl_state->dispatcher,
                             RED_WORKER_MESSAGE_GL_SCANOUT, &payload);
 }
 
@@ -909,7 +909,7 @@ void spice_qxl_gl_draw_async(QXLInstance *qxl,
     spice_return_if_fail(qxl_state->gl_draw_async == NULL);
 
     qxl_state->gl_draw_async = async_command_alloc(qxl_state, message, cookie);
-    dispatcher_send_message(&qxl_state->dispatcher, message, &draw);
+    dispatcher_send_message(qxl_state->dispatcher, message, &draw);
 }
 
 void red_qxl_async_complete(QXLInstance *qxl, AsyncCommand *async_command)
@@ -960,7 +960,7 @@ void red_qxl_init(RedsState *reds, QXLInstance *qxl)
     qxl_state->qxl = qxl;
     pthread_mutex_init(&qxl_state->scanout_mutex, NULL);
     qxl_state->scanout.drm_dma_buf_fd = -1;
-    dispatcher_init(&qxl_state->dispatcher, RED_WORKER_MESSAGE_COUNT, NULL);
+    qxl_state->dispatcher = dispatcher_new(RED_WORKER_MESSAGE_COUNT, NULL);
     qxl_state->qxl_worker.major_version = SPICE_INTERFACE_QXL_MAJOR;
     qxl_state->qxl_worker.minor_version = SPICE_INTERFACE_QXL_MINOR;
     qxl_state->qxl_worker.wakeup = qxl_worker_wakeup;
@@ -999,9 +999,9 @@ void red_qxl_init(RedsState *reds, QXLInstance *qxl)
     red_worker_run(worker);
 }
 
-struct Dispatcher *red_qxl_get_dispatcher(QXLInstance *qxl)
+Dispatcher *red_qxl_get_dispatcher(QXLInstance *qxl)
 {
-    return &qxl->st->dispatcher;
+    return qxl->st->dispatcher;
 }
 
 void red_qxl_clear_pending(QXLState *qxl_state, int pending)
@@ -1031,7 +1031,7 @@ void red_qxl_on_ic_change(QXLInstance *qxl, SpiceImageCompression ic)
 {
     RedWorkerMessageSetCompression payload;
     payload.image_compression = ic;
-    dispatcher_send_message(&qxl->st->dispatcher,
+    dispatcher_send_message(qxl->st->dispatcher,
                             RED_WORKER_MESSAGE_SET_COMPRESSION,
                             &payload);
 }
@@ -1040,7 +1040,7 @@ void red_qxl_on_sv_change(QXLInstance *qxl, int sv)
 {
     RedWorkerMessageSetStreamingVideo payload;
     payload.streaming_video = sv;
-    dispatcher_send_message(&qxl->st->dispatcher,
+    dispatcher_send_message(qxl->st->dispatcher,
                             RED_WORKER_MESSAGE_SET_STREAMING_VIDEO,
                             &payload);
 }
@@ -1049,7 +1049,7 @@ void red_qxl_set_mouse_mode(QXLInstance *qxl, uint32_t mode)
 {
     RedWorkerMessageSetMouseMode payload;
     payload.mode = mode;
-    dispatcher_send_message(&qxl->st->dispatcher,
+    dispatcher_send_message(qxl->st->dispatcher,
                             RED_WORKER_MESSAGE_SET_MOUSE_MODE,
                             &payload);
 }