Merge pull request #281 from tne-lab/eb-scopedpointer

Event Broadcaster socket reorganization

Author: aacuevas

Source/Plugins/EventBroadcaster/EventBroadcaster.cpp

@@ -14,6 +14,8 @@
 EventBroadcaster::ZMQContext::ZMQContext()
 #ifdef ZEROMQ
     : context(zmq_ctx_new())
+#else
+    : context(nullptr)
 #endif
 {}
 
@@ -31,72 +33,91 @@ void* EventBroadcaster::ZMQContext::createZMQSocket()
 #ifdef ZEROMQ
     jassert(context != nullptr);
     return zmq_socket(context, ZMQ_PUB);
+#else
+    return nullptr;
 #endif
 }
 
-EventBroadcaster::ZMQSocketPtr::ZMQSocketPtr()
-    : std::unique_ptr<void, decltype(&closeZMQSocket)>(nullptr, &closeZMQSocket)
+EventBroadcaster::ZMQSocket::ZMQSocket()
+    : socket    (nullptr)
+    , boundPort (0)
 {
 #ifdef ZEROMQ
-    reset(context->createZMQSocket());
+    socket = context->createZMQSocket();
 #endif
 }
 
-EventBroadcaster::ZMQSocketPtr::~ZMQSocketPtr()
+EventBroadcaster::ZMQSocket::~ZMQSocket()
 {
-    // close the socket before the context might get destroyed.
-    reset(nullptr);
+#ifdef ZEROMQ
+    unbind(); // do this explicitly to free the port immediately
+    zmq_close(socket);
+#endif
 }
 
-int EventBroadcaster::unbindZMQSocket()
+bool EventBroadcaster::ZMQSocket::isValid() const
+{
+    return socket != nullptr;
+}
+
+int EventBroadcaster::ZMQSocket::getBoundPort() const
+{
+    return boundPort;
+}
+
+int EventBroadcaster::ZMQSocket::send(const void* buf, size_t len, int flags)
 {
 #ifdef ZEROMQ
-    void* socket = zmqSocket.get();
-    if (socket != nullptr && listeningPort != 0)
-    {
-        return zmq_unbind(socket, getEndpoint(listeningPort).toRawUTF8());
-    }
+    return zmq_send(socket, buf, len, flags);
 #endif
     return 0;
 }
 
-int EventBroadcaster::rebindZMQSocket()
+int EventBroadcaster::ZMQSocket::bind(int port)
 {
 #ifdef ZEROMQ
-    void* socket = zmqSocket.get();
-    if (socket != nullptr && listeningPort != 0)
+    if (isValid() && port != 0)
     {
-        return zmq_bind(socket, getEndpoint(listeningPort).toRawUTF8());
+        int status = unbind();
+        if (status == 0)
+        {
+            status = zmq_bind(socket, getEndpoint(port).toRawUTF8());
+            if (status == 0)
+            {
+                boundPort = port;
+            }
+        }
+        return status;
     }
 #endif
     return 0;
 }
 
-void EventBroadcaster::closeZMQSocket(void* socket)
+int EventBroadcaster::ZMQSocket::unbind()
 {
 #ifdef ZEROMQ
-    zmq_close(socket);
+    if (isValid() && boundPort != 0)
+    {
+        int status = zmq_unbind(socket, getEndpoint(boundPort).toRawUTF8());
+        if (status == 0)
+        {
+            boundPort = 0;
+        }
+        return status;
+    }
 #endif
+    return 0;
 }
 
+
 String EventBroadcaster::getEndpoint(int port)
 {
     return String("tcp://*:") + String(port);
 }
 
-void EventBroadcaster::reportActualListeningPort(int port)
-{
-    listeningPort = port;
-    auto editor = static_cast<EventBroadcasterEditor*>(getEditor());
-    if (editor)
-    {
-        editor->setDisplayedPort(port);
-    }
-}
 
 EventBroadcaster::EventBroadcaster()
     : GenericProcessor  ("Event Broadcaster")
-    , listeningPort     (0)
 {
     setProcessorType (PROCESSOR_TYPE_SINK);
 
@@ -118,59 +139,65 @@ AudioProcessorEditor* EventBroadcaster::createEditor()
 
 int EventBroadcaster::getListeningPort() const
 {
-    return listeningPort;
+    if (zmqSocket == nullptr)
+    {
+        return 0;
+    }
+    return zmqSocket->getBoundPort();
 }
 
 
 int EventBroadcaster::setListeningPort(int port, bool forceRestart)
 {
-    if ((listeningPort != port) || forceRestart)
+    int status = 0;
+    int currPort = getListeningPort();
+    if ((currPort != port) || forceRestart)
     {
 #ifdef ZEROMQ
         // unbind current socket (if any) to free up port
-        unbindZMQSocket();
-        ZMQSocketPtr newSocket;
-        auto editor = static_cast<EventBroadcasterEditor*>(getEditor());
-        int status = 0;
+        if (zmqSocket != nullptr)
+        {
+            zmqSocket->unbind();
+        }
 
-        if (!newSocket.get())
+        ScopedPointer<ZMQSocket> newSocket = new ZMQSocket();
+
+        if (!newSocket->isValid())
         {
             status = zmq_errno();
             std::cout << "Failed to create socket: " << zmq_strerror(status) << std::endl;
         }
         else
         {
-            if (0 != zmq_bind(newSocket.get(), getEndpoint(port).toRawUTF8()))
+            if (0 != newSocket->bind(port))
             {
                 status = zmq_errno();
-                std::cout << "Failed to open socket: " << zmq_strerror(status) << std::endl;
+                std::cout << "Failed to bind to port " << port << ": "
+                    << zmq_strerror(status) << std::endl;
             }
             else
             {
                 // success
-                zmqSocket.swap(newSocket);
-                reportActualListeningPort(port);
-                return status;
+                zmqSocket = newSocket;
             }
         }
 
-        // failure, try to rebind current socket to previous port
-        if (0 == rebindZMQSocket())
+        if (status != 0 && zmqSocket != nullptr)
         {
-            reportActualListeningPort(listeningPort);
+            // try to rebind current socket to previous port
+            zmqSocket->bind(currPort);
         }
-        else
-        {
-            reportActualListeningPort(0);
-        }
-        return status;
 
-#else
-        reportActualListeningPort(port);
-        return 0;
 #endif
     }
-    return -1;
+
+    // update editor
+    auto editor = static_cast<EventBroadcasterEditor*>(getEditor());
+    if (editor != nullptr)
+    {
+        editor->setDisplayedPort(getListeningPort());
+    }
+    return status;
 }
 
 
@@ -183,13 +210,17 @@ void EventBroadcaster::process(AudioSampleBuffer& continuousBuffer)
 //IMPORTANT: The structure of the event buffers has changed drastically, so we need to find a better way of doing this
 void EventBroadcaster::sendEvent(const MidiMessage& event, float eventSampleRate) const
 {
+#ifdef ZEROMQ
 	double timestampSeconds = double(Event::getTimestamp(event)) / eventSampleRate;
 	uint16 type = Event::getBaseType(event);
 
-#ifdef ZEROMQ
-	if (-1 == zmq_send(zmqSocket.get(), &type, sizeof(type), ZMQ_SNDMORE) ||
-		-1 == zmq_send(zmqSocket.get(), &timestampSeconds, sizeof(timestampSeconds), ZMQ_SNDMORE) ||
-		-1 == zmq_send(zmqSocket.get(), event.getRawData(), event.getRawDataSize(), 0))
+    if (zmqSocket == nullptr)
+    {
+        std::cout << "Failed to send message: no socket" << std::endl;
+    }
+	else if (-1 == zmqSocket->send(&type, sizeof(type), ZMQ_SNDMORE) ||
+		     -1 == zmqSocket->send(&timestampSeconds, sizeof(timestampSeconds), ZMQ_SNDMORE) ||
+		     -1 == zmqSocket->send(event.getRawData(), event.getRawDataSize(), 0))
 	{
 		std::cout << "Failed to send message: " << zmq_strerror(zmq_errno()) << std::endl;
 	}
@@ -209,7 +240,7 @@ void EventBroadcaster::handleSpike(const SpikeChannel* channelInfo, const MidiMe
 void EventBroadcaster::saveCustomParametersToXml(XmlElement* parentElement)
 {
     XmlElement* mainNode = parentElement->createNewChildElement("EVENTBROADCASTER");
-    mainNode->setAttribute("port", listeningPort);
+    mainNode->setAttribute("port", getListeningPort());
 }
 
 
@@ -221,7 +252,7 @@ void EventBroadcaster::loadCustomParametersFromXml()
         {
             if (mainNode->hasTagName("EVENTBROADCASTER"))
             {
-                setListeningPort(mainNode->getIntAttribute("port"));
+                setListeningPort(mainNode->getIntAttribute("port", getListeningPort()));
             }
         }
     }

Source/Plugins/EventBroadcaster/EventBroadcaster.h

@@ -22,8 +22,6 @@
     #endif
 #endif
 
-#include <memory>
-
 class EventBroadcaster : public GenericProcessor
 {
 public:
@@ -33,7 +31,7 @@ class EventBroadcaster : public GenericProcessor
 
     int getListeningPort() const;
     // returns 0 on success, else the errno value for the error that occurred.
-    int setListeningPort (int port, bool forceRestart = false);
+    int setListeningPort(int port, bool forceRestart = false);
 
     void process (AudioSampleBuffer& continuousBuffer) override;
     void handleEvent (const EventChannel* channelInfo, const MidiMessage& event, int samplePosition = 0) override;
@@ -54,27 +52,32 @@ class EventBroadcaster : public GenericProcessor
         void* context;
     };
 
-    static void closeZMQSocket(void* socket);
-
-    class ZMQSocketPtr : public std::unique_ptr<void, decltype(&closeZMQSocket)>
+    class ZMQSocket
     {
     public:
-        ZMQSocketPtr();
-        ~ZMQSocketPtr();
+        ZMQSocket();
+        ~ZMQSocket();
+
+        bool isValid() const;
+        int getBoundPort() const;
+
+        int send(const void* buf, size_t len, int flags);
+        int bind(int port);
+        int unbind();
     private:
+        int boundPort;
+        void* socket;
+
+        // see here for why the context can't just be static:
+        // https://github.com/zeromq/libzmq/issues/1708
         SharedResourcePointer<ZMQContext> context;
     };
-    
-    int unbindZMQSocket();
-    int rebindZMQSocket();
 
 	void sendEvent(const MidiMessage& event, float eventSampleRate) const;
-    static String getEndpoint(int port);
-    // called from getListeningPort() depending on success/failure of ZMQ operations
-    void reportActualListeningPort(int port);
 
-    ZMQSocketPtr zmqSocket;
-    int listeningPort;
+    static String getEndpoint(int port);
+    
+    ScopedPointer<ZMQSocket> zmqSocket;
 };