Monitor implementation

tests/common.h

@@ -98,4 +98,23 @@ void onEchoEvent(const void* data, const size_t size)
     BOOST_CHECK_EQUAL(size, message.size());
     BOOST_CHECK_EQUAL(echoMessage, message);
 }
+
+class Monitor : public zeroeq::Monitor
+{
+public:
+    explicit Monitor(zeroeq::Sender& sender)
+        : zeroeq::Monitor(sender)
+        , connections(0)
+    {
+    }
+
+    Monitor(zeroeq::Sender& sender, zeroeq::Receiver& shared)
+        : zeroeq::Monitor(sender, shared)
+        , connections(0)
+    {
+    }
+
+    void notifyNewConnection() final { ++connections; }
+    size_t connections;
+};
 }

tests/connection/broker.cpp

@@ -43,6 +43,7 @@ class Subscriber
 
         // Using the connection broker in place of zeroconf
         BrokerPtr broker = createBroker(subscriber);
+        test::Monitor monitor(*broker, subscriber);
         BOOST_REQUIRE(broker.get());
         if (!broker)
             return;
@@ -55,8 +56,12 @@ class Subscriber
         }
 
         // test receive of data for echo event
-        for (size_t i = 0; i < 100 && !received; ++i)
+        for (size_t i = 0; i < 100 && !received && monitor.connections == 0;
+             ++i)
+        {
             subscriber.receive(100);
+        }
+        BOOST_CHECK_EQUAL(monitor.connections, 1);
     }
 
     void waitStarted() const

tests/pubSub.cpp

@@ -22,18 +22,31 @@ BOOST_AUTO_TEST_CASE(publish_receive_serializable)
 
     zeroeq::Publisher publisher(zeroeq::NULL_SESSION);
     zeroeq::Subscriber subscriber(zeroeq::URI(publisher.getURI()));
+    test::Monitor monitor(publisher);
+
     BOOST_CHECK(subscriber.subscribe(echoIn));
 
     for (size_t i = 0; i < 10; ++i)
     {
         BOOST_CHECK(publisher.publish(echoOut));
-
+        std::cout << monitor.receive(0) << std::endl;
         if (subscriber.receive(100))
         {
             BOOST_CHECK_EQUAL(echoIn.getMessage(), echoOut.getMessage());
+
+            for (size_t j = 0; j < 10; ++j)
+            {
+                if (monitor.receive(100))
+                {
+                    BOOST_CHECK_EQUAL(monitor.connections, 1);
+                    return;
+                }
+            }
+            BOOST_CHECK(!"reachable");
             return;
         }
     }
+    BOOST_CHECK_EQUAL(monitor.connections, 1);
     BOOST_CHECK(!"reachable");
 }
 
@@ -42,6 +55,8 @@ BOOST_AUTO_TEST_CASE(publish_receive_event)
     const std::string echoString("The quick brown fox");
     zeroeq::Publisher publisher(zeroeq::NULL_SESSION);
     zeroeq::Subscriber subscriber(zeroeq::URI(publisher.getURI()));
+    test::Monitor monitor(publisher, subscriber);
+
     bool received = false;
     BOOST_CHECK(subscriber.subscribe(
         zeroeq::make_uint128("Echo"),
@@ -59,6 +74,15 @@ BOOST_AUTO_TEST_CASE(publish_receive_event)
         if (subscriber.receive(100))
         {
             BOOST_CHECK(received);
+            for (size_t j = 0; j < 10; ++j)
+            {
+                if (subscriber.receive(100))
+                {
+                    BOOST_CHECK_EQUAL(monitor.connections, 1);
+                    return;
+                }
+            }
+            BOOST_CHECK(!"reachable");
             return;
         }
     }
@@ -208,7 +232,7 @@ BOOST_AUTO_TEST_CASE(publish_receive_filters)
         while (subscriber.receive(0)) /* NOP to drain */
             ;
     }
-    const auto& noEchoTime =
+    const auto noEchoTime =
         std::chrono::high_resolution_clock::now() - startTime;
 
     // Benchmark with echo handler, now should send data

zeroeq/CMakeLists.txt

@@ -7,8 +7,10 @@ set(ZEROEQ_PUBLIC_HEADERS
   connection/broker.h
   connection/service.h
   log.h
+  monitor.h
   publisher.h
   receiver.h
+  sender.h
   subscriber.h
   types.h
   uri.h)
@@ -26,6 +28,7 @@ set(ZEROEQ_SOURCES
   connection/service.cpp
   detail/port.cpp
   detail/sender.cpp
+  monitor.cpp
   publisher.cpp
   receiver.cpp
   subscriber.cpp

zeroeq/connection/broker.cpp

@@ -1,5 +1,5 @@
 
-/* Copyright (c) 2014-2015, Human Brain Project
+/* Copyright (c) 2014-2017, Human Brain Project
  *                          [email protected]
  */
 
@@ -130,5 +130,10 @@ std::string Broker::getAddress() const
 {
     return _impl->getAddress();
 }
+
+void* Broker::getSocket()
+{
+    return _impl->socket;
+}
 }
 }

zeroeq/connection/broker.h

@@ -1,5 +1,5 @@
 
-/* Copyright (c) 2014-2015, Human Brain Project
+/* Copyright (c) 2014-2017, Human Brain Project
  *                          [email protected]
  */
 
@@ -8,6 +8,7 @@
 
 #include <zeroeq/log.h>
 #include <zeroeq/receiver.h> // base class
+#include <zeroeq/sender.h>   // base class
 
 namespace zeroeq
 {
@@ -35,7 +36,7 @@ class Broker;
  *
  * Example: @include tests/connection/broker.cpp
  */
-class Broker : public Receiver
+class Broker : public Receiver, public Sender
 {
 public:
     enum PortSelection
@@ -94,6 +95,9 @@ class Broker : public Receiver
     // Receiver API
     void addSockets(std::vector<zeroeq::detail::Socket>& entries) final;
     void process(zeroeq::detail::Socket& socket, uint32_t timeout) final;
+
+    // Sender API
+    void* getSocket() final;
 };
 }
 }

zeroeq/monitor.cpp

@@ -0,0 +1,118 @@
+
+/* Copyright (c) 2017, Human Brain Project
+ *                          [email protected]
+ */
+
+#include "monitor.h"
+
+#include "detail/socket.h"
+#include "log.h"
+#include "sender.h"
+
+#include <zmq.h>
+
+namespace zeroeq
+{
+class Monitor::Impl
+{
+public:
+    Impl(Sender& sender, void* context)
+    {
+        const auto inproc = std::string("inproc://zeroeq.monitor.") +
+                            servus::make_UUID().getString();
+
+        if (::zmq_socket_monitor(sender.getSocket(), inproc.c_str(),
+                                 ZMQ_EVENT_ALL) != 0)
+        {
+            ZEROEQTHROW(
+                std::runtime_error(std::string("Cannot monitor socket: ") +
+                                   zmq_strerror(zmq_errno())));
+        }
+
+        _socket = ::zmq_socket(context, ZMQ_PAIR);
+        if (!_socket)
+            ZEROEQTHROW(std::runtime_error(
+                std::string("Cannot create inproc socket: ") +
+                zmq_strerror(zmq_errno())));
+
+        if (::zmq_connect(_socket, inproc.c_str()) != 0)
+        {
+            ZEROEQTHROW(std::runtime_error(
+                std::string("Cannot connect inproc socket: ") +
+                zmq_strerror(zmq_errno())));
+        }
+    }
+
+    ~Impl()
+    {
+        if (_socket)
+            ::zmq_close(_socket);
+    }
+
+    void addSockets(std::vector<zeroeq::detail::Socket>& entries)
+    {
+        zeroeq::detail::Socket entry;
+        entry.socket = _socket;
+        entry.events = ZMQ_POLLIN;
+        entries.push_back(entry);
+    }
+
+private:
+    void* _socket;
+};
+
+Monitor::Monitor(Sender& sender)
+    : Receiver()
+    , _impl(new Impl(sender, getZMQContext()))
+{
+}
+
+Monitor::Monitor(Sender& sender, Receiver& shared)
+    : Receiver(shared)
+    , _impl(new Impl(sender, getZMQContext()))
+{
+}
+
+Monitor::~Monitor()
+{
+}
+
+void Monitor::addSockets(std::vector<zeroeq::detail::Socket>& entries)
+{
+    _impl->addSockets(entries);
+}
+
+void Monitor::process(zeroeq::detail::Socket& socket, uint32_t)
+{
+    // Messages consist of 2 Frames, the first containing the event-id and the
+    // associated value. The second frame holds the affected endpoint as string.
+    zmq_msg_t msg;
+    zmq_msg_init(&msg);
+
+    //  The layout of the first Frame is: 16 bit event id 32 bit event value
+    if (zmq_msg_recv(&msg, socket.socket, 0) == -1)
+        return;
+
+    const uint16_t event = *(uint16_t*)zmq_msg_data(&msg);
+    if (!zmq_msg_more(&msg))
+        return;
+    zmq_msg_close(&msg);
+
+    //  Second frame in message contains event address, skip
+    zmq_msg_init(&msg);
+    if (zmq_msg_recv(&msg, socket.socket, 0) == -1)
+        return;
+    zmq_msg_close(&msg);
+
+    switch (event)
+    {
+    case ZMQ_EVENT_CONNECTED:
+    case ZMQ_EVENT_ACCEPTED:
+        notifyNewConnection();
+        break;
+
+    default:
+        ZEROEQWARN << "Unhandled monitor event " << event << std::endl;
+    }
+}
+}

zeroeq/monitor.h

@@ -0,0 +1,38 @@
+
+/* Copyright (c) 2017, Human Brain Project
+ *                          [email protected]
+ */
+
+#pragma once
+
+#include <zeroeq/receiver.h> // base class
+
+namespace zeroeq
+{
+/** Monitors a Sender and notifies on events on its socket. */
+class Monitor : public Receiver
+{
+public:
+    /** Monitor the given sender. */
+    ZEROEQ_API explicit Monitor(Sender& sender);
+
+    /** Monitor the given sender and notify with the given shared group. */
+    ZEROEQ_API Monitor(Sender& sender, Receiver& shared);
+
+    /** Destroy this monitor*/
+    ZEROEQ_API ~Monitor();
+
+    /** Notify of a new connection to the sender. */
+    virtual void notifyNewConnection() {}
+private:
+    class Impl;
+    std::unique_ptr<Impl> _impl;
+
+    Monitor(const Monitor&) = delete;
+    Monitor& operator=(const Monitor&) = delete;
+
+    // Receiver API
+    void addSockets(std::vector<zeroeq::detail::Socket>& entries) final;
+    void process(zeroeq::detail::Socket& socket, uint32_t timeout) final;
+};
+}

zeroeq/publisher.cpp

@@ -178,4 +178,9 @@ const URI& Publisher::getURI() const
 {
     return _impl->uri;
 }
+
+void* Publisher::getSocket()
+{
+    return _impl->socket;
+}
 }

zeroeq/publisher.h

@@ -1,5 +1,5 @@
 
-/* Copyright (c) 2014-2016, Human Brain Project
+/* Copyright (c) 2014-2017, Human Brain Project
  *                          Daniel Nachbaur <[email protected]>
  *                          [email protected]
  */
@@ -8,6 +8,7 @@
 #define ZEROEQ_PUBLISHER_H
 
 #include <zeroeq/api.h>
+#include <zeroeq/sender.h> // base class
 #include <zeroeq/types.h>
 
 #include <memory>
@@ -22,7 +23,7 @@ namespace zeroeq
  *
  * Example: @include tests/publisher.cpp
  */
-class Publisher
+class Publisher : public Sender
 {
 public:
     /**
@@ -136,6 +137,8 @@ class Publisher
 
     Publisher(const Publisher&) = delete;
     Publisher& operator=(const Publisher&) = delete;
+
+    ZEROEQ_API void* getSocket() final;
 };
 }