Abstract away communication into interface

* Create generic interface ufo_messenger_iface
* Move all zmq_* stuff into ufo_zmq_messenger that implements that
  interface
* also fixes a bug that happened when closing the zmq_socket from
  another thread by introducing a special TERMINATE message that
  will kill ufo-daemon

5 files changed, 133 insertions, 3 deletions

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 254bfa7..22a9813 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -7,6 +7,7 @@ set(TEST_SRCS
     test-graph.c
     test-profiler.c
     test-remote-node.c
+    test-zmq-messenger.c
     )
 
 set(SUITE_BIN "test-suite")
diff --git a/tests/test-remote-node.c b/tests/test-remote-node.c
index 59bd485..8642988 100644
--- a/tests/test-remote-node.c
+++ b/tests/test-remote-node.c
@@ -19,7 +19,6 @@
 
 #include <string.h>
 #include <ufo.h>
-#include <zmq.h>
 #include "test-suite.h"
 
 typedef struct {
@@ -44,8 +43,8 @@ static void
 teardown (Fixture *fixture, gconstpointer data)
 {
     g_object_unref (fixture->remote_node);
-
     ufo_daemon_stop (fixture->daemon);
+
     g_object_unref (fixture->daemon);
 }
 
@@ -58,9 +57,27 @@ test_remote_node_get_num_cpus (Fixture *fixture,
     g_assert (n_gpus > 0);
 }
 
+static void
+test_remote_node_get_structure (Fixture *fixture,
+                               gconstpointer unused)
+{
+    UfoTaskMode mode;
+    UfoInputParam *in_params;
+    guint n_inputs;
+    ufo_remote_node_get_structure (fixture->remote_node, &n_inputs, &in_params, &mode);
+    g_message ("received n_inputs == %d", n_inputs);
+    g_assert (n_inputs == 1);
+    g_message ("received n_dims == %d", in_params->n_dims);
+    g_assert (in_params->n_dims == 2);
+
+}
+
 void
 test_add_remote_node (void)
 {
+    g_test_add ("/remotenode/get_structure",
+                Fixture, NULL,
+                setup, test_remote_node_get_structure, teardown);
     g_test_add ("/remotenode/get_num_cpus",
                 Fixture, NULL,
                 setup, test_remote_node_get_num_cpus, teardown);
diff --git a/tests/test-suite.c b/tests/test-suite.c
index e6c813a..76a5eab 100644
--- a/tests/test-suite.c
+++ b/tests/test-suite.c
@@ -26,6 +26,7 @@ ignore_log (const gchar     *domain,
             const gchar     *message,
             gpointer         data)
 {
+    // g_print ("%s\n",message);
 }
 
 int main(int argc, char *argv[])
@@ -41,8 +42,8 @@ int main(int argc, char *argv[])
     test_add_config ();
     test_add_graph ();
     test_add_profiler ();
+    test_add_zmq_messenger ();
     test_add_remote_node ();
-
     g_test_run();
 
     return 0;
diff --git a/tests/test-suite.h b/tests/test-suite.h
index cc54ddb..afe699a 100644
--- a/tests/test-suite.h
+++ b/tests/test-suite.h
@@ -6,5 +6,6 @@ void test_add_config (void);
 void test_add_graph (void);
 void test_add_profiler (void);
 void test_add_remote_node (void);
+void test_add_zmq_messenger (void);
 
 #endif
diff --git a/tests/test-zmq-messenger.c b/tests/test-zmq-messenger.c
new file mode 100644
index 0000000..84576f7
--- /dev/null
+++ b/tests/test-zmq-messenger.c
@@ -0,0 +1,110 @@
+/*
+ * Copyright (C) 2011-2013 Karlsruhe Institute of Technology
+ *
+ * This file is part of Ufo.
+ *
+ * This library is free software: you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation, either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <ufo/ufo.h>
+#include "test-suite.h"
+
+typedef struct {
+    gchar *addr;
+} Fixture;
+
+static void
+setup (Fixture *fixture, gconstpointer data)
+{
+    fixture->addr = g_strdup ("tcp://127.0.0.1:5555");
+}
+
+static void
+teardown (Fixture *fixture, gconstpointer data)
+{
+    g_free (fixture->addr);
+}
+
+static void send_num_devices_request (gpointer unused)
+{
+    UfoMessenger *msger = UFO_MESSENGER (ufo_zmq_messenger_new ());
+    gchar *addr = g_strdup ("tcp://127.0.0.1:5555");
+    ufo_messenger_connect (msger, addr, UFO_MESSENGER_CLIENT);
+
+    guint x = 0;
+    while (x++ < 10) {
+        UfoMessage *request = ufo_message_new (UFO_MESSAGE_GET_NUM_DEVICES, 0);
+        UfoMessage *response;
+
+        response = ufo_messenger_send_blocking (msger, request, NULL);
+
+        guint16 num_devices = *(guint16 *) response->data;
+        g_assert (num_devices == x);
+
+        ufo_message_free (request);
+        ufo_message_free (response);
+    }
+    ufo_zmq_messenger_disconnect (msger);
+    g_object_unref (msger);
+}
+
+static void handle_num_devices (gpointer unused)
+{
+    UfoMessenger *msger = UFO_MESSENGER (ufo_zmq_messenger_new ());
+    gchar *addr = g_strdup ("tcp://127.0.0.1:5555");
+    ufo_messenger_connect (msger, addr, UFO_MESSENGER_SERVER);
+
+    guint16 x = 0;
+    GError *err = NULL;
+    while (x++ < 10) {
+        UfoMessage *msg = ufo_messenger_recv_blocking (UFO_MESSENGER (msger), &err);
+        if (err != NULL)
+            g_critical ("%s", err->message);
+
+        UfoMessage *resp;
+        switch (msg->type) {
+            case UFO_MESSAGE_GET_NUM_DEVICES:
+                resp = ufo_message_new (UFO_MESSAGE_ACK, sizeof (guint16));
+                *(guint16 *)resp->data = x;
+                ufo_zmq_messenger_send_blocking (msger, resp, NULL);
+                ufo_message_free (resp);
+                break;
+            default:
+                g_critical ("Unexpected message type: %d", msg->type);
+            break;
+        }
+        ufo_message_free (msg);
+    };
+
+    ufo_zmq_messenger_disconnect (msger);
+    g_object_unref (msger);
+}
+
+static void test_zmq_messenger (Fixture *fixture, gconstpointer unused)
+{
+    GThread *server = g_thread_create ((GThreadFunc) handle_num_devices, NULL, TRUE, NULL);
+    GThread *client = g_thread_create ((GThreadFunc) send_num_devices_request, NULL, TRUE, NULL);
+
+    g_thread_join (client);
+    g_thread_join (server);
+}
+
+
+void
+test_add_zmq_messenger (void)
+{
+    g_test_add ("/zmq_messenger/test_messenger",
+                Fixture, NULL,
+                setup, test_zmq_messenger, teardown);
+}