threading race condition fixes, detected by TSAN

examples/satori_publisher/IXSatoriConnection.h

@@ -26,6 +26,12 @@ namespace ix
         SatoriConnection_EventType_Closed = 3
     };
 
+    enum SatoriConnectionPublishMode
+    {
+        SatoriConnection_PublishMode_Immediate = 0,
+        SatoriConnection_PublishMode_Batch = 1
+    };
+
     using SubscriptionCallback = std::function<void(const Json::Value&)>;
     using EventCallback = std::function<void(SatoriConnectionEventType,
                                              const std::string&,
@@ -46,7 +52,6 @@ namespace ix
                        const std::string& rolesecret,
                        WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions);
 
-        /// Set the traffic tracker callback
         static void setTrafficTrackerCallback(const TrafficTrackerCallback& callback);
 
         /// Reset the traffic tracker callback to an no-op one.
@@ -61,7 +66,7 @@ namespace ix
         /// Publish a message to a channel
         ///
         /// No-op if the connection is not established
-        bool publish(const std::string& channel,
+        bool publish(const Json::Value& channels,
                      const Json::Value& msg);
 
         // Subscribe to a channel, and execute a callback when an incoming
@@ -71,7 +76,7 @@ namespace ix
         /// Unsubscribe from a channel
         void unsubscribe(const std::string& channel);
 
-        /// Close the RTM connection and free the RTM handle memory
+        /// Close the connection
         void disconnect();
 
         /// Connect to Satori and authenticate the connection
@@ -80,17 +85,27 @@ namespace ix
         /// Returns true only if we're connected
         bool isConnected() const;
         
+        /// Flush the publish queue
+        bool flushQueue();
+
+        /// Set the publish mode
+        void setPublishMode(SatoriConnectionPublishMode publishMode);
+
+        /// Lifecycle management. Free resources when backgrounding
+        void suspend();
+        void resume();
+
     private:
         bool sendHandshakeMessage();
         bool handleHandshakeResponse(const Json::Value& data);
         bool sendAuthMessage(const std::string& nonce);
         bool handleSubscriptionData(const Json::Value& pdu);
 
-        void resetWebSocketOnMessageCallback();
+        void initWebSocketOnMessageCallback();
 
         bool publishMessage(const std::string& serializedJson);
-        bool flushQueue();
         void enqueue(const std::string& msg);
+        std::string serializeJson(const Json::Value& pdu);
 
         /// Invoke the traffic tracker callback
         static void invokeTrafficTrackerCallback(size_t size, bool incoming);
@@ -111,6 +126,7 @@ namespace ix
         std::string _endpoint;
         std::string _role_name;
         std::string _role_secret;
+        std::atomic<SatoriConnectionPublishMode> _publishMode;
 
         // Can be set on control+background thread, protecting with an atomic
         std::atomic<bool> _authenticated;
@@ -119,12 +135,14 @@ namespace ix
         Json::Value _body;
         Json::Value _pdu;
         Json::FastWriter _jsonWriter;
+        mutable std::mutex _jsonWriterMutex;
 
         /// Traffic tracker callback
         static TrafficTrackerCallback _trafficTrackerCallback;
 
         /// Satori events callbacks
         EventCallback _eventCallback;
+        mutable std::mutex _eventCallbackMutex;
 
         /// Subscription callbacks, only one per channel
         std::unordered_map<std::string, SubscriptionCallback> _cbs;
@@ -139,7 +157,7 @@ namespace ix
         mutable std::mutex _queueMutex;
 
         // Cap the queue size (100 elems so far -> ~100k)
-        static constexpr size_t kQueueMaxSize = 100;
+        static constexpr size_t kQueueMaxSize = 256;
     };
     
 } // namespace ix