Reduce errors binding socket and improve feedback for Event Broadcaster

Author: ethanbb

Source/Plugins/EventBroadcaster/EventBroadcaster.cpp

@@ -22,6 +22,29 @@ std::shared_ptr<void> EventBroadcaster::getZMQContext() {
     return ctx;
 }
 
+int EventBroadcaster::unbindZMQSocket()
+{
+#ifdef ZEROMQ
+    void* socket = zmqSocket.get();
+    if (socket != nullptr && listeningPort != 0)
+    {
+        return zmq_unbind(socket, getEndpoint(listeningPort).toRawUTF8());
+    }
+#endif
+    return 0;
+}
+
+int EventBroadcaster::rebindZMQSocket()
+{
+#ifdef ZEROMQ
+    void* socket = zmqSocket.get();
+    if (socket != nullptr && listeningPort != 0)
+    {
+        return zmq_bind(socket, getEndpoint(listeningPort).toRawUTF8());
+    }
+#endif
+    return 0;
+}
 
 void EventBroadcaster::closeZMQSocket(void* socket)
 {
@@ -30,16 +53,35 @@ void EventBroadcaster::closeZMQSocket(void* socket)
 #endif
 }
 
+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")
     , zmqContext        (getZMQContext())
-    , zmqSocket         (nullptr, &closeZMQSocket)
+    , zmqSocket         (nullptr)
     , listeningPort     (0)
 {
     setProcessorType (PROCESSOR_TYPE_SINK);
 
-    setListeningPort(5557);
+    int portToTry = 5557;
+    while (setListeningPort(portToTry) == EADDRINUSE)
+    {
+        // try the next port, looking for one not in use
+        portToTry++;
+    }
 }
 
 
@@ -56,27 +98,53 @@ int EventBroadcaster::getListeningPort() const
 }
 
 
-void EventBroadcaster::setListeningPort(int port, bool forceRestart)
+int EventBroadcaster::setListeningPort(int port, bool forceRestart)
 {
     if ((listeningPort != port) || forceRestart)
     {
 #ifdef ZEROMQ
-        zmqSocket.reset(zmq_socket(zmqContext.get(), ZMQ_PUB));
-        if (!zmqSocket)
+        // unbind current socket (if any) to free up port
+        unbindZMQSocket();
+        zmqSocketPtr newSocket(zmq_socket(zmqContext.get(), ZMQ_PUB));
+        auto editor = static_cast<EventBroadcasterEditor*>(getEditor());
+        int status = 0;
+
+        if (!newSocket.get())
         {
-            std::cout << "Failed to create socket: " << zmq_strerror(zmq_errno()) << std::endl;
-            return;
+            status = zmq_errno();
+            std::cout << "Failed to create socket: " << zmq_strerror(status) << std::endl;
+        }
+        else
+        {
+            if (0 != zmq_bind(newSocket.get(), getEndpoint(port).toRawUTF8()))
+            {
+                status = zmq_errno();
+                std::cout << "Failed to open socket: " << zmq_strerror(status) << std::endl;
+            }
+            else
+            {
+                // success
+                zmqSocket.swap(newSocket);
+                reportActualListeningPort(port);
+                return status;
+            }
         }
 
-        String url = String("tcp://*:") + String(port);
-        if (0 != zmq_bind(zmqSocket.get(), url.toRawUTF8()))
+        // failure, try to rebind current socket to previous port
+        if (0 == rebindZMQSocket())
         {
-            std::cout << "Failed to open socket: " << zmq_strerror(zmq_errno()) << std::endl;
-            return;
+            reportActualListeningPort(listeningPort);
         }
-#endif
+        else
+        {
+            reportActualListeningPort(0);
+        }
+        return status;
 
-        listeningPort = port;
+#else
+        reportActualListeningPort(port);
+        return 0;
+#endif
     }
 }
 

Source/Plugins/EventBroadcaster/EventBroadcaster.h

@@ -33,7 +33,8 @@ class EventBroadcaster : public GenericProcessor
     AudioProcessorEditor* createEditor() override;
 
     int getListeningPort() const;
-    void setListeningPort (int port, bool forceRestart = false);
+    // returns 0 on success, else the errno value for the error that occurred.
+    int setListeningPort (int port, bool forceRestart = false);
 
     void process (AudioSampleBuffer& continuousBuffer) override;
     void handleEvent (const EventChannel* channelInfo, const MidiMessage& event, int samplePosition = 0) override;
@@ -46,10 +47,23 @@ class EventBroadcaster : public GenericProcessor
 private:
 	void sendEvent(const MidiMessage& event, float eventSampleRate) const;
     static std::shared_ptr<void> getZMQContext();
+    int unbindZMQSocket();
+    int rebindZMQSocket();
     static void closeZMQSocket (void* socket);
+    static String getEndpoint(int port);
+    // called from getListeningPort() depending on success/failure of ZMQ operations
+    void reportActualListeningPort(int port);
+
+    // encapuslate closing sockets when their pointers go out of scope
+    struct zmqSocketPtr : public std::unique_ptr<void, decltype(&closeZMQSocket)>
+    {
+        zmqSocketPtr(void* ptr) 
+            : std::unique_ptr<void, decltype(&closeZMQSocket)>(ptr, &closeZMQSocket)
+        {}
+    };
 
     const std::shared_ptr<void> zmqContext;
-    std::unique_ptr<void, decltype (&closeZMQSocket)> zmqSocket;
+    zmqSocketPtr zmqSocket;
     int listeningPort;
 
 };

Source/Plugins/EventBroadcaster/EventBroadcasterEditor.cpp

@@ -46,7 +46,14 @@ void EventBroadcasterEditor::buttonEvent(Button* button)
     if (button == restartConnection)
     {
         EventBroadcaster* p = (EventBroadcaster*)getProcessor();
-        p->setListeningPort(p->getListeningPort(), true);
+        int status = p->setListeningPort(p->getListeningPort(), true);
+
+#ifdef ZEROMQ
+        if (status != 0)
+        {
+            CoreServices::sendStatusMessage(String("Restart failed: ") + zmq_strerror(status));
+        }
+#endif
     }
 }
 
@@ -58,6 +65,18 @@ void EventBroadcasterEditor::labelTextChanged(juce::Label* label)
         Value val = label->getTextValue();
 
         EventBroadcaster* p = (EventBroadcaster*)getProcessor();
-        p->setListeningPort(val.getValue());
+        int status = p->setListeningPort(val.getValue());
+
+#ifdef ZEROMQ
+        if (status != 0)
+        {
+            CoreServices::sendStatusMessage(String("Port change failed: ") + zmq_strerror(status));
+        }
+#endif
     }
 }
+
+void EventBroadcasterEditor::setDisplayedPort(int port)
+{
+    portLabel->setText(String(port), dontSendNotification);
+}

Source/Plugins/EventBroadcaster/EventBroadcasterEditor.h

@@ -30,6 +30,8 @@ class EventBroadcasterEditor : public GenericEditor, public Label::Listener
     void buttonEvent(Button* button) override;
     void labelTextChanged(juce::Label* label) override;
 
+    void setDisplayedPort(int port);
+
 private:
     ScopedPointer<UtilityButton> restartConnection;
     ScopedPointer<Label> urlLabel;