(cobra client) send a websocket ping every 30s to keep the connection opened

docs/CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
-All notable changes to this project will be documented in this file.
+All changes to this project will be documented in this file.
+
+## [7.6.5] - 2019-12-24
+
+(cobra client) send a websocket ping every 30s to keep the connection opened
+
+## [7.6.4] - 2019-12-22
+
+(client) error handling, quote url in error case when failing to parse one
+(ws) ws_cobra_publish: register callbacks before connecting
+(doc) mention mbedtls in supported ssl server backend
 
 ## [7.6.3] - 2019-12-20
 

docs/ws.md

@@ -247,7 +247,7 @@ Options:
 
 [cobra](https://github.com/machinezone/cobra) is a real time messenging server. ws has several sub-command to interact with cobra. There is also a minimal cobra compatible server named snake available.
 
-Below are examples on running a snake server and clients with TLS enabled (the server only works with the OpenSSL backend for now).
+Below are examples on running a snake server and clients with TLS enabled (the server only works with the OpenSSL and the Mbed TLS backend for now).
 
 First, generate certificates.
 
@@ -366,4 +366,4 @@ $ ws cobra_publish --endpoint wss://127.0.0.1:8765 --appkey FC2F10139A2BAc53BB72
 [2019-12-19 20:46:42.659] [info] Published message id 3 acked
 ```
 
-To use OpenSSL on macOS, compile with `make ws_openssl`. First you will have to install OpenSSL libraries, which can be done with Homebrew.
+To use OpenSSL on macOS, compile with `make ws_openssl`. First you will have to install OpenSSL libraries, which can be done with Homebrew. Use `make ws_mbedtls` accordingly to use MbedTLS.

ixcobra/ixcobra/IXCobraConnection.cpp

@@ -24,6 +24,7 @@ namespace ix
     PublishTrackerCallback CobraConnection::_publishTrackerCallback = nullptr;
     constexpr size_t CobraConnection::kQueueMaxSize;
     constexpr CobraConnection::MsgId CobraConnection::kInvalidMsgId;
+    constexpr int CobraConnection::kPingIntervalSecs;
 
     CobraConnection::CobraConnection() :
         _webSocket(new WebSocket()),
@@ -228,6 +229,10 @@ namespace ix
                     ss << "HTTP Status: "      << msg->errorInfo.http_status << std::endl;
                     invokeErrorCallback(ss.str(), std::string());
                 }
+                else if (msg->type == ix::WebSocketMessageType::Pong)
+                {
+                    invokeEventCallback(ix::CobraConnection_EventType_Pong);
+                }
         });
     }
 
@@ -260,6 +265,7 @@ namespace ix
         _webSocket->setUrl(url);
         _webSocket->setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
         _webSocket->setTLSOptions(socketTLSOptions);
+        _webSocket->setPingInterval(kPingIntervalSecs);
     }
 
     //

ixcobra/ixcobra/IXCobraConnection.h

@@ -30,7 +30,8 @@ namespace ix
         CobraConnection_EventType_Closed = 3,
         CobraConnection_EventType_Subscribed = 4,
         CobraConnection_EventType_UnSubscribed = 5,
-        CobraConnection_EventType_Published = 6
+        CobraConnection_EventType_Published = 6,
+        CobraConnection_EventType_Pong = 7
     };
 
     enum CobraConnectionPublishMode
@@ -215,6 +216,9 @@ namespace ix
 
         // Each pdu sent should have an incremental unique id
         std::atomic<uint64_t> _id;
+
+        // Frequency at which we send a websocket ping to the backing cobra connection
+        static constexpr int kPingIntervalSecs = 30;
     };
 
 } // namespace ix

ixcobra/ixcobra/IXCobraMetricsThreadedPublisher.cpp

@@ -65,6 +65,10 @@ namespace ix
                 {
                     ss << "Published message " << msgId << " acked";
                 }
+                else if (eventType == ix::CobraConnection_EventType_Pong)
+                {
+                    ss << "Received websocket pong";
+                }
 
                 ix::IXCoreLogger::Log(ss.str().c_str());
         });

ixwebsocket/IXSocketAppleSSL.cpp

@@ -24,9 +24,47 @@
 
 #include <Security/SecureTransport.h>
 
-namespace
+namespace ix
 {
-    OSStatus read_from_socket(SSLConnectionRef connection, void* data, size_t* len)
+    SocketAppleSSL::SocketAppleSSL(const SocketTLSOptions& tlsOptions, int fd)
+        : Socket(fd)
+        , _sslContext(nullptr)
+        , _tlsOptions(tlsOptions)
+    {
+        ;
+    }
+
+    SocketAppleSSL::~SocketAppleSSL()
+    {
+        SocketAppleSSL::close();
+    }
+
+    std::string SocketAppleSSL::getSSLErrorDescription(OSStatus status)
+    {
+        std::string errMsg("Unknown SSL error.");
+
+        CFErrorRef error = CFErrorCreate(kCFAllocatorDefault, kCFErrorDomainOSStatus, status, NULL);
+        if (error)
+        {
+            CFStringRef message = CFErrorCopyDescription(error);
+            if (message)
+            {
+                char localBuffer[128];
+                Boolean success;
+                success = CFStringGetCString(message, localBuffer, 128, kCFStringEncodingUTF8);
+                if (success)
+                {
+                    errMsg = localBuffer;
+                }
+                CFRelease(message);
+            }
+            CFRelease(error);
+        }
+
+        return errMsg;
+    }
+
+    OSStatus SocketAppleSSL::readFromSocket(SSLConnectionRef connection, void* data, size_t* len)
     {
         int fd = (int) (long) connection;
         if (fd < 0) return errSSLInternal;
@@ -67,7 +105,7 @@ namespace
         }
     }
 
-    OSStatus write_to_socket(SSLConnectionRef connection, const void* data, size_t* len)
+    OSStatus SocketAppleSSL::writeToSocket(SSLConnectionRef connection, const void* data, size_t* len)
     {
         int fd = (int) (long) connection;
         if (fd < 0) return errSSLInternal;
@@ -105,47 +143,6 @@ namespace
         }
     }
 
-    std::string getSSLErrorDescription(OSStatus status)
-    {
-        std::string errMsg("Unknown SSL error.");
-
-        CFErrorRef error = CFErrorCreate(kCFAllocatorDefault, kCFErrorDomainOSStatus, status, NULL);
-        if (error)
-        {
-            CFStringRef message = CFErrorCopyDescription(error);
-            if (message)
-            {
-                char localBuffer[128];
-                Boolean success;
-                success = CFStringGetCString(message, localBuffer, 128, kCFStringEncodingUTF8);
-                if (success)
-                {
-                    errMsg = localBuffer;
-                }
-                CFRelease(message);
-            }
-            CFRelease(error);
-        }
-
-        return errMsg;
-    }
-
-} // anonymous namespace
-
-namespace ix
-{
-    SocketAppleSSL::SocketAppleSSL(const SocketTLSOptions& tlsOptions, int fd)
-        : Socket(fd)
-        , _sslContext(nullptr)
-        , _tlsOptions(tlsOptions)
-    {
-        ;
-    }
-
-    SocketAppleSSL::~SocketAppleSSL()
-    {
-        SocketAppleSSL::close();
-    }
 
     bool SocketAppleSSL::accept(std::string& errMsg)
     {
@@ -168,7 +165,7 @@ namespace ix
 
             _sslContext = SSLCreateContext(kCFAllocatorDefault, kSSLClientSide, kSSLStreamType);
 
-            SSLSetIOFuncs(_sslContext, read_from_socket, write_to_socket);
+            SSLSetIOFuncs(_sslContext, SocketAppleSSL::readFromSocket, SocketAppleSSL::writeToSocket);
             SSLSetConnection(_sslContext, (SSLConnectionRef)(long) _sockfd);
             SSLSetProtocolVersionMin(_sslContext, kTLSProtocol12);
             SSLSetPeerDomainName(_sslContext, host.c_str(), host.size());

ixwebsocket/IXSocketAppleSSL.h

@@ -34,6 +34,10 @@ namespace ix
         virtual ssize_t recv(void* buffer, size_t length) final;
 
     private:
+        static std::string getSSLErrorDescription(OSStatus status);
+        static OSStatus writeToSocket(SSLConnectionRef connection, const void* data, size_t* len);
+        static OSStatus readFromSocket(SSLConnectionRef connection, void* data, size_t* len);
+
         SSLContextRef _sslContext;
         mutable std::mutex _mutex; // AppleSSL routines are not thread-safe
 

ixwebsocket/IXSocketTLSOptions.cpp

@@ -81,6 +81,7 @@ namespace ix
         ss << "  keyFile  = " << keyFile << std::endl;
         ss << "  caFile   = " << caFile << std::endl;
         ss << "  ciphers  = " << ciphers << std::endl;
+        ss << "  ciphers  = " << ciphers << std::endl;
         return ss.str();
     }
 } // namespace ix

ixwebsocket/IXWebSocketTransport.cpp

@@ -144,7 +144,9 @@ namespace ix
 
         if (!UrlParser::parse(url, protocol, host, path, query, port))
         {
-            return WebSocketInitResult(false, 0, std::string("Could not parse URL ") + url);
+            std::stringstream ss;
+            ss << "Could not parse url: '" << url << "'";
+            return WebSocketInitResult(false, 0, ss.str());
         }
 
         std::string errorMsg;

ixwebsocket/IXWebSocketVersion.h

@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "7.6.3"
+#define IX_WEBSOCKET_VERSION "7.6.5"

ws/ws_cobra_publish.cpp

@@ -43,7 +43,6 @@ namespace ix
                        rolesecret,
                        ix::WebSocketPerMessageDeflateOptions(true),
                        tlsOptions);
-        conn.connect();
 
         // Display incoming messages
         std::atomic<bool> authenticated(false);
@@ -92,8 +91,14 @@ namespace ix
                 spdlog::info("Published message id {} acked", msgId);
                 messageAcked = true;
             }
+            else if (eventType == ix::CobraConnection_EventType_Pong)
+            {
+                spdlog::info("Received websocket pong");
+            }
         });
 
+        conn.connect();
+
         while (!authenticated)
             ;
         while (!messageAcked)

ws/ws_cobra_subscribe.cpp

@@ -100,6 +100,10 @@ namespace ix
                 {
                     spdlog::error("Published message hacked: {}", msgId);
                 }
+                else if (eventType == ix::CobraConnection_EventType_Pong)
+                {
+                    spdlog::info("Received websocket pong");
+                }
             });
 
         while (true)

ws/ws_cobra_to_sentry.cpp

@@ -245,6 +245,10 @@ namespace ix
             {
                 spdlog::error("Published message hacked: {}", msgId);
             }
+            else if (eventType == ix::CobraConnection_EventType_Pong)
+            {
+                spdlog::info("Received websocket pong");
+            }
         });
 
         while (true)

ws/ws_cobra_to_statsd.cpp

@@ -160,6 +160,10 @@ namespace ix
             {
                 spdlog::error("Published message hacked: {}", msgId);
             }
+            else if (eventType == ix::CobraConnection_EventType_Pong)
+            {
+                spdlog::info("Received websocket pong");
+            }
         });
 
         while (true)