Simplify and fix for Linux and with no zeromq

Author: ethanbb

Source/Plugins/NetworkEvents/NetworkEvents.cpp

@@ -46,19 +46,20 @@ NetworkEvents::NetworkEvents()
     , state             (false)
     , urlport           (0)
     , firstTime         (true)
-    , restart           (0)
-    , portString        (getPortString())
+    , restart           (false)
+    , changeResponder   (false)
 {
     setProcessorType (PROCESSOR_TYPE_SOURCE);
 
+    portString = getPortString();
+    portString.addListener(this);
+
     if (!setNewListeningPort(5556))
     {
         // resort to choosing a port automatically
         setNewListeningPort(0);
     }
 
-    portString.addListener(this);
-
     sendSampleCount = false; // disable updating the continuous buffer sample counts,
     // since this processor only sends events
 
@@ -78,11 +79,6 @@ NetworkEvents::~NetworkEvents()
 
 bool NetworkEvents::setNewListeningPort(uint16 port)
 {
-    if (port == 0)
-    {
-        CoreServices::sendStatusMessage("NetworkEvents: Selecting port automatically");
-    }
-
     ScopedPointer<Responder> newResponder = Responder::makeResponder(port);
     if (newResponder == nullptr)
     {
@@ -91,25 +87,30 @@ bool NetworkEvents::setNewListeningPort(uint16 port)
 
     ScopedLock nrLock(nextResponderLock);
     nextResponder = newResponder;
+    changeResponder = true;
     return true;
 }
 
 
 String NetworkEvents::getPortString() const
 {
+#ifdef ZEROMQ
     uint16 port = urlport;
     if (port == 0)
     {
         return "<no cxn>";
     }
 
     return String(port);
+#else
+    return "<no zeromq>";
+#endif
 }
 
 
 void NetworkEvents::restartConnection()
 {
-    restart = 1;
+    restart = true;
 }
 
 
@@ -412,8 +413,9 @@ void NetworkEvents::run()
     while (!threadShouldExit())
     {
         // reopen connection if necessary
-        if (restart.compareAndSetBool(0, 1))
+        if (restart)
         {
+            restart = false; // the other thread only sets restart to true, so it's not a race condition
             responder = nullptr; // destroy old one, which frees the port
             responder = Responder::makeResponder(urlport);
 
@@ -424,16 +426,21 @@ void NetworkEvents::run()
             }
         }
 
-        // switch to responder with new port if necessary
-        // unfortunately, atomic operations aren't available for smart pointers
+        // switch to new responder if necessary
+        if (changeResponder)
         {
-            const ScopedTryLock nrLock(nextResponderLock);
-            if (nrLock.isLocked() && nextResponder != nullptr)
+            const ScopedLock nrLock(nextResponderLock);
+            if (nextResponder != nullptr)
             {
+                changeResponder = false; // this is also controlled by the nextResponderLock
                 responder = nextResponder;
                 connected = true;
                 updatePort(responder->getBoundPort());
             }
+            else
+            {
+                jassertfalse; // huh? a new responder should be available
+            }
         }
 
         // if we don't have a vaild (connected) socket, keep looping until we do
@@ -650,6 +657,7 @@ NetworkEvents::ZMQContext::ZMQContext(const ScopedLock& lock)
     : context(zmq_ctx_new())
 #endif
 {
+    // sharedContextLock should already be held here
     sharedContext = this;
 }
 
@@ -669,12 +677,16 @@ void* NetworkEvents::ZMQContext::createSocket()
 #ifdef ZEROMQ
     jassert(context != nullptr);
     return zmq_socket(context, ZMQ_REP);
+#else
+    return nullptr;
 #endif
 }
 
 
 /*** Responder ***/
 
+const int NetworkEvents::Responder::RECV_TIMEOUT_MS = 100;
+
 NetworkEvents::Responder::Responder(uint16 port)
     : socket(nullptr)
     , boundPort(0)
@@ -719,6 +731,7 @@ NetworkEvents::Responder::Responder(uint16 port)
     // if requested port was 0, find out which port was actually used
     if (port == 0)
     {
+        CoreServices::sendStatusMessage("NetworkEvents: Selecting port automatically");
         const size_t BUF_LEN = 32;
         size_t len = BUF_LEN;
         char endpoint[BUF_LEN];
@@ -737,7 +750,7 @@ NetworkEvents::Responder::Responder(uint16 port)
 }
 
 
-NetworkEvents::Responder* NetworkEvents::Responder::makeResponder(uint16 port)
+ScopedPointer<NetworkEvents::Responder> NetworkEvents::Responder::makeResponder(uint16 port)
 {
     ScopedPointer<Responder> socketPtr = new Responder(port);
     uint16 boundPort = socketPtr->getBoundPort();
@@ -752,7 +765,7 @@ NetworkEvents::Responder* NetworkEvents::Responder::makeResponder(uint16 port)
         jassertfalse; // huh?
         return nullptr;
     }
-    return socketPtr.release();
+    return socketPtr;
 }
 
 

Source/Plugins/NetworkEvents/NetworkEvents.h

@@ -115,6 +115,8 @@ class NetworkEvents : public GenericProcessor
 
         String str;
         juce::int64 timestamp;
+
+        JUCE_LEAK_DETECTOR(StringTS);
     };
 
     class ZMQContext : public ReferenceCountedObject
@@ -128,15 +130,16 @@ class NetworkEvents : public GenericProcessor
 
     private:
         void* context;
+
+        JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(ZMQContext);
     };
 
     // RAII wrapper for REP socket
     class Responder
     {
     public:
         // tries to create a responder and bind to given port; returns nullptr on failure.
-        // caller must own and destroy the returned responder if it succeeds.
-        static Responder* makeResponder(uint16 port);
+        static ScopedPointer<Responder> makeResponder(uint16 port);
         ~Responder();
 
         // returns the latest errno value
@@ -166,7 +169,9 @@ class NetworkEvents : public GenericProcessor
         uint16 boundPort; // 0 indicates not bound
         int lastErrno;
 
-        static const int RECV_TIMEOUT_MS = 100;
+        static const int RECV_TIMEOUT_MS;
+        
+        JUCE_DECLARE_NON_COPYABLE_WITH_LEAK_DETECTOR(Responder);
     };
 
     void postTimestamppedStringToMidiBuffer(StringTS s);
@@ -183,8 +188,7 @@ class NetworkEvents : public GenericProcessor
     // get an endpoint url for the given port (using 0 to represent *)
     static String getEndpoint(uint16 port);
 
-    Value portString; // underlying value of the editor's port input
-     
+
     // share a "dumb" pointer that doesn't take part in reference counting.
     // want the context to be terminated by the time the static members are
     // destroyed (see: https://github.com/zeromq/libzmq/issues/1708)
@@ -196,9 +200,12 @@ class NetworkEvents : public GenericProcessor
     ScopedPointer<Responder> nextResponder;
     CriticalSection nextResponderLock;
 
-    uint16 urlport;  // 0 indicates not connected
+    bool restart;
+    bool changeResponder;
 
-    Atomic<int> restart;
+    uint16 urlport;   // 0 indicates not connected
+    // thread can modify this to update editor's port text without acquiring MessageManagerLock
+    Value portString;
 
     float threshold;
     float bufferZone;