IPC pub sub test

sf-buffer/src/sf_replay.cpp

@@ -47,6 +47,7 @@ int main (int argc, char *argv[]) {
 
     auto ctx = zmq_ctx_new();
     auto socket = zmq_socket(ctx, ZMQ_PUSH);
+    auto meta_socket = zmq_socket(ctx, ZMQ_SUB);
 
     int status = 0;
 
@@ -66,6 +67,16 @@ int main (int argc, char *argv[]) {
         throw runtime_error(strerror (errno));
     }
 
+    //TODO: Use ipc?
+    if (zmq_connect(socket, "ipc://writer_metad") != 0) {
+        throw runtime_error(strerror (errno));
+    }
+
+    cout << "receiving " << endl;
+    uint64_t response;
+    zmq_recv(meta_socket, &response, sizeof(response), 0);
+    cout << "Done!! " << response << endl;
+
     for (const auto& suffix:path_suffixes) {
         metadata_buffer->start_pulse_id = suffix.start_pulse_id;
         metadata_buffer->stop_pulse_id = suffix.stop_pulse_id;

sf-writer/sf_h5_writer.cpp

@@ -7,6 +7,7 @@
 #include <BufferUtils.hpp>
 #include <jungfrau.hpp>
 #include <unordered_map>
+#include <thread>
 
 using namespace std;
 
@@ -35,6 +36,7 @@ int main (int argc, char *argv[])
     zmq_ctx_set (ctx, ZMQ_IO_THREADS, 16);
 
     auto socket = zmq_socket(ctx, ZMQ_PULL);
+    auto meta_socket = zmq_socket(ctx, ZMQ_PUB);
 
     int rcvhwm = 1000;
     if (zmq_setsockopt(socket, ZMQ_RCVHWM, &rcvhwm, sizeof(rcvhwm)) != 0) {
@@ -46,13 +48,19 @@ int main (int argc, char *argv[])
         throw runtime_error(strerror (errno));
     }
 
-    //TODO: Use ipc?
     if (zmq_bind(socket, "ipc://writer") != 0) {
         throw runtime_error(strerror (errno));
     }
 
+    if (zmq_bind(meta_socket, "ipc://writer_meta") != 0) {
+        throw runtime_error(strerror (errno));
+    }
 
+    this_thread::sleep_for(chrono::milliseconds(5000));
 
+    uint64_t test = 12;
+    zmq_send(meta_socket, &test, sizeof(test),0);
+    cout << "sent pub" << endl;
 
     auto metadata_buffer = make_unique<ModuleFrame>();