bump version

CobraConnection: sets a counter as a field for each event published.

.pre-commit-config.yaml

@@ -0,0 +1,12 @@
+repos:
+-   repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v2.3.0
+    hooks:
+    -   id: check-yaml
+    -   id: end-of-file-fixer
+    -   id: trailing-whitespace
+
+-   repo: https://github.com/pocc/pre-commit-hooks
+    rev: ''
+    hooks:
+    -   id: clang-format

.travis.yml

@@ -15,15 +15,17 @@ matrix:
         - python test/run.py
         - make ws
 
-    # # Linux
+    # Linux
-    # - os: linux
+    - os: linux
-    #   dist: xenial
+      dist: bionic
-    #   script: 
+      before_install:
-    #     - python test/run.py
+        - sudo apt-get install -y libmbedtls-dev
-    #     - make ws
+      script: 
-    #   env:
+        - python test/run.py
-    #     - CC=gcc
+        - make ws
-    #     - CXX=g++
+      env:
+        - CC=gcc
+        - CXX=g++
 
     # Clang + Linux disabled for now
     # - os: linux

CHANGELOG.md

@@ -1,8 +1,36 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [5.0.7] - 2019-08-23
+- WebSocket: add new option to pass in extra HTTP headers when connecting.
+- `ws connect` add new option (-H, works like [curl](https://stackoverflow.com/questions/356705/how-to-send-a-header-using-a-http-request-through-a-curl-call)) to pass in extra HTTP headers when connecting 
+
+If you run against `ws echo_server` you will see the headers being received printed in the terminal.
+```
+ws connect -H "foo: bar" -H "baz: buz" ws://127.0.0.1:8008
+```
+
+- CobraConnection: sets a unique id field for all messages sent to [cobra](https://github.com/machinezone/cobra).
+- CobraConnection: sets a counter as a field for each event published.
+
+## [5.0.6] - 2019-08-22
+- Windows: silly compile error (poll should be in the global namespace)
+
+## [5.0.5] - 2019-08-22
+- Windows: use select instead of WSAPoll, through a poll wrapper
+
+## [5.0.4] - 2019-08-20
+- Windows build fixes (there was a problem with the use of ::poll that has a different name on Windows (WSAPoll))
+
+## [5.0.3] - 2019-08-14
+- CobraMetricThreadedPublisher _enable flag is an atomic, and CobraMetricsPublisher is enabled by default
+
+## [5.0.2] - 2019-08-01
+- ws cobra_subscribe has a new -q (quiet) option
+- ws cobra_subscribe knows to and display msg stats (count and # of messages received per second)
+- ws cobra_subscribe, cobra_to_statsd and cobra_to_sentry commands have a new option, --filter to restrict the events they want to receive
+
 ## [5.0.1] - 2019-07-25
-### Unreleased
 - ws connect command has a new option to send in binary mode (still default to text)
 - ws connect command has readline history thanks to libnoise-cpp. Now ws connect one can use using arrows to lookup previous sent messages and edit them
 

DOCKER_VERSION

@@ -1 +1 @@
-5.0.1
+5.0.7

README.md

@@ -16,9 +16,32 @@
 
 The [*ws*](https://github.com/machinezone/IXWebSocket/tree/master/ws) folder countains many interactive programs for chat, [file transfers](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_send.cpp), [curl like](https://github.com/machinezone/IXWebSocket/blob/master/ws/ws_http_client.cpp) http clients, demonstrating client and server usage.
 
-Here is what the client API looks like.
+### Windows note
+
+To use the network system on Windows, you need to initialize it once with *WSAStartup()* and clean it up with *WSACleanup()*. We have helpers for that which you can use, see below. This init would typically take place in your main function.
 
 ```
+#include <ixwebsocket/IXNetSystem.h>
+
+int main()
+{
+    ix::initNetSystem();
+
+    ...
+
+    ix::uninitNetSystem();
+    return 0;
+}
+```
+
+### WebSocket client API
+
+```
+#include <ixwebsocket/IXWebSocket.h>
+
+...
+
+# Our websocket object
 ix::WebSocket webSocket;
 
 std::string url("ws://localhost:8080/");
@@ -56,9 +79,13 @@ webSocket.sendBinary("some serialized binary data");
 webSocket.stop()
 ```
 
-Here is what the server API looks like. Note that server support is very recent and subject to changes.
+### WebSocket server API
 
 ```
+#include <ixwebsocket/IXWebSocketServer.h>
+
+...
+
 // Run a server on localhost at a given port.
 // Bound host name, max connections and listen backlog can also be passed in as parameters.
 ix::WebSocketServer server(port);
@@ -117,9 +144,13 @@ server.wait();
 
 ```
 
-Here is what the HTTP client API looks like.
+### HTTP client API
 
 ```
+#include <ixwebsocket/IXHttpClient.h>
+
+...
+
 //
 // Preparation
 //
@@ -170,7 +201,7 @@ out = httpClient.post(url, std::string("foo=bar"), args);
 //
 // Result
 //
-auto errorCode = response->errorCode; // Can be HttpErrorCode::Ok, HttpErrorCode::UrlMalformed, etc...
+auto statusCode = response->statusCode; // Can be HttpErrorCode::Ok, HttpErrorCode::UrlMalformed, etc...
 auto errorCode = response->errorCode; // 200, 404, etc...
 auto responseHeaders = response->headers; // All the headers in a special case-insensitive unordered_map of (string, string)
 auto payload = response->payload; // All the bytes from the response as an std::string
@@ -196,9 +227,11 @@ bool ok = httpClient.performRequest(args, [](const HttpResponsePtr& response)
 // ok will be false if your httpClient is not async
 ```
 
-Here is what the HTTP server API looks like. Note that HTTP server support is very, very recent and subject to changes.
+### HTTP server API
 
 ```
+#include <ixwebsocket/IXHttpServer.h>
+
 ix::HttpServer server(port, hostname);
 
 auto res = server.listen();
@@ -243,7 +276,7 @@ CMakefiles for the library and the examples are available. This library has few
 ```
 mkdir build # make a build dir so that you can build out of tree.
 cd build
-cmake ..
+cmake -DUSE_TLS=1 ..
 make -j
 make install # will install to /usr/local on Unix, on macOS it is a good idea to sudo chown -R `whoami`:staff /usr/local
 ```
@@ -251,6 +284,12 @@ make install # will install to /usr/local on Unix, on macOS it is a good idea to
 Headers and a static library will be installed to the target dir.
 There is a unittest which can be executed by typing `make test`.
 
+Options for building:
+
+* `-DUSE_TLS=1` will enable TLS support
+* `-DUSE_MBED_TLS=1` will use [mbedlts](https://tls.mbed.org/) for the TLS support (default on Windows)
+* `-DUSE_WS=1` will build the ws interactive command line tool
+
 ### vcpkg
 
 It is possible to get IXWebSocket through Microsoft [vcpkg](https://github.com/microsoft/vcpkg).
@@ -471,3 +510,13 @@ idle connection.
 ```
 webSocket.setHeartBeatPeriod(45);
 ```
+
+### Supply extra HTTP headers.
+
+You can set extra HTTP headers to be sent during the WebSocket handshake.
+
+```
+WebSocketHttpHeaders headers;
+headers["foo"] = "bar";
+webSocket.setExtraHeaders(headers);
+```

ixwebsocket/IXDNSLookup.h

@@ -12,10 +12,10 @@
 
 #include "IXCancellationRequest.h"
 #include <atomic>
+#include <memory>
 #include <mutex>
 #include <set>
 #include <string>
-#include <memory>
 
 struct addrinfo;
 

ixwebsocket/IXHttp.h

@@ -6,8 +6,8 @@
 
 #pragma once
 
-#include "IXWebSocketHttpHeaders.h"
 #include "IXProgressCallback.h"
+#include "IXWebSocketHttpHeaders.h"
 #include <tuple>
 
 namespace ix
@@ -111,10 +111,12 @@ namespace ix
     class Http
     {
     public:
-        static std::tuple<bool, std::string, HttpRequestPtr> parseRequest(std::shared_ptr<Socket> socket);
+        static std::tuple<bool, std::string, HttpRequestPtr> parseRequest(
+            std::shared_ptr<Socket> socket);
         static bool sendResponse(HttpResponsePtr response, std::shared_ptr<Socket> socket);
 
-        static std::tuple<std::string, std::string, std::string> parseRequestLine(const std::string& line);
+        static std::tuple<std::string, std::string, std::string> parseRequestLine(
+            const std::string& line);
         static std::string trim(const std::string& str);
     };
-}
+} // namespace ix

ixwebsocket/IXHttpClient.cpp

@@ -14,6 +14,7 @@
 #include <vector>
 #include <cstring>
 
+#include <assert.h>
 #include <zlib.h>
 
 namespace ix
@@ -52,6 +53,8 @@ namespace ix
     bool HttpClient::performRequest(HttpRequestArgsPtr args,
                                     const OnResponseCallback& onResponseCallback)
     {
+        assert(_async && "HttpClient needs its async parameter set to true "
+                         "in order to call performRequest");
         if (!_async) return false;
 
         // Enqueue the task

ixwebsocket/IXHttpClient.h

@@ -6,9 +6,9 @@
 
 #pragma once
 
+#include "IXHttp.h"
 #include "IXSocket.h"
 #include "IXWebSocketHttpHeaders.h"
-#include "IXHttp.h"
 #include <algorithm>
 #include <atomic>
 #include <condition_variable>

ixwebsocket/IXHttpServer.h

@@ -6,9 +6,9 @@
 
 #pragma once
 
+#include "IXHttp.h"
 #include "IXSocketServer.h"
 #include "IXWebSocket.h"
-#include "IXHttp.h"
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -47,4 +47,3 @@ namespace ix
         void setDefaultConnectionCallback();
     };
 } // namespace ix
-

ixwebsocket/IXNetSystem.cpp

@@ -15,9 +15,8 @@ namespace ix
         WSADATA wsaData;
         int err;
 
-        /* Use the MAKEWORD(lowbyte, highbyte) macro declared in Windef.h */
+        // Use the MAKEWORD(lowbyte, highbyte) macro declared in Windef.h
         wVersionRequested = MAKEWORD(2, 2);
-
         err = WSAStartup(wVersionRequested, &wsaData);
 
         return err == 0;
@@ -30,10 +29,83 @@ namespace ix
     {
 #ifdef _WIN32
         int err = WSACleanup();
-
         return err == 0;
 #else
         return true;
 #endif
     }
 }
+
+// This function should be in the global namespace
+#ifdef _WIN32
+    //
+    // That function could 'return WSAPoll(pfd, nfds, timeout);'
+    // but WSAPoll is said to have weird behaviors on the internet
+    // (the curl folks have had problems with it).
+    //
+    // So we make it a select wrapper
+    //
+    int poll(struct pollfd *fds, nfds_t nfds, int timeout)
+    {
+        int maxfd = 0;
+        fd_set readfds, writefds, errorfds;
+        FD_ZERO(&readfds);
+        FD_ZERO(&writefds);
+        FD_ZERO(&errorfds);
+
+        for (nfds_t i = 0; i < nfds; ++i)
+        {
+            struct pollfd *fd = &fds[i];
+
+            if (fd->fd > maxfd)
+            {
+                maxfd = fd->fd;
+            }
+            if ((fd->events & POLLIN))
+            {
+                FD_SET(fd->fd, &readfds);
+            }
+            if ((fd->events & POLLOUT))
+            {
+                FD_SET(fd->fd, &writefds);
+            }
+            if ((fd->events & POLLERR))
+            {
+                FD_SET(fd->fd, &errorfds);
+            }
+        }
+
+        struct timeval tv;
+        tv.tv_sec = timeout / 1000;
+        tv.tv_usec = (timeout % 1000) * 1000;
+
+        int ret = select(maxfd + 1, &readfds, &writefds, &errorfds,
+                         timeout != -1 ? &tv : NULL);
+
+        if (ret < 0) 
+        {
+            return ret;
+        }
+
+        for (nfds_t i = 0; i < nfds; ++i)
+        {
+            struct pollfd *fd = &fds[i];
+            fd->revents = 0;
+
+            if (FD_ISSET(fd->fd, &readfds))
+            {
+                fd->revents |= POLLIN;
+            }
+            if (FD_ISSET(fd->fd, &writefds))
+            {
+                fd->revents |= POLLOUT;
+            }
+            if (FD_ISSET(fd->fd, &errorfds))
+            {
+                fd->revents |= POLLERR;
+            }
+        }
+
+        return ret;
+    }
+#endif

ixwebsocket/IXNetSystem.h

@@ -12,11 +12,18 @@
 #include <basetsd.h>
 #include <io.h>
 #include <ws2def.h>
+
+// Define our own poll on Windows
+typedef unsigned long int nfds_t;
+
+int poll(struct pollfd* fds, nfds_t nfds, int timeout);
+
 #else
 #include <arpa/inet.h>
 #include <errno.h>
 #include <netdb.h>
 #include <netinet/tcp.h>
+#include <poll.h>
 #include <sys/select.h>
 #include <sys/socket.h>
 #include <sys/stat.h>

ixwebsocket/IXSocket.cpp

@@ -17,7 +17,6 @@
 #include <stdint.h>
 #include <fcntl.h>
 #include <sys/types.h>
-#include <poll.h>
 
 #include <algorithm>
 
@@ -59,7 +58,7 @@ namespace ix
         // shim to fallback to select on those platforms.
         // See https://github.com/mpv-player/mpv/pull/5203/files for such a select wrapper.
         //
-        int nfds = 1;
+        nfds_t nfds = 1;
         struct pollfd fds[2];
 
         fds[0].fd = sockfd;
@@ -128,6 +127,10 @@ namespace ix
                 optval != 0)
             {
                 pollResult = PollResultType::Error;
+
+                // set errno to optval so that external callers can have an
+                // appropriate error description when calling strerror
+                errno = optval;
             }
 #endif
         }

ixwebsocket/IXWebSocket.cpp

@@ -70,6 +70,11 @@ namespace ix
         std::lock_guard<std::mutex> lock(_configMutex);
         _url = url;
     }
+    void WebSocket::setExtraHeaders(const WebSocketHttpHeaders& headers)
+    {
+        std::lock_guard<std::mutex> lock(_configMutex);
+        _extraHeaders = headers;
+    }
 
     const std::string& WebSocket::getUrl() const
     {
@@ -176,7 +181,7 @@ namespace ix
                           _pingTimeoutSecs);
         }
 
-        WebSocketInitResult status = _ws.connectToUrl(_url, timeoutSecs);
+        WebSocketInitResult status = _ws.connectToUrl(_url, _extraHeaders, timeoutSecs);
         if (!status.success)
         {
             return status;

ixwebsocket/IXWebSocket.h

@@ -44,6 +44,9 @@ namespace ix
         ~WebSocket();
 
         void setUrl(const std::string& url);
+
+        // send extra headers in client handshake request
+        void setExtraHeaders(const WebSocketHttpHeaders& headers);
         void setPerMessageDeflateOptions(
             const WebSocketPerMessageDeflateOptions& perMessageDeflateOptions);
         void setHeartBeatPeriod(int heartBeatPeriodSecs);
@@ -111,6 +114,8 @@ namespace ix
         WebSocketTransport _ws;
 
         std::string _url;
+        WebSocketHttpHeaders _extraHeaders;
+
         WebSocketPerMessageDeflateOptions _perMessageDeflateOptions;
         mutable std::mutex _configMutex; // protect all config variables access
 

ixwebsocket/IXWebSocketHandshake.cpp

@@ -88,6 +88,7 @@ namespace ix
     }
 
     WebSocketInitResult WebSocketHandshake::clientHandshake(const std::string& url,
+                                                            const WebSocketHttpHeaders& extraHeaders,
                                                             const std::string& host,
                                                             const std::string& path,
                                                             int port,
@@ -127,6 +128,9 @@ namespace ix
         ss << "Sec-WebSocket-Version: 13\r\n";
         ss << "Sec-WebSocket-Key: " << secWebSocketKey << "\r\n";
 
+        for (auto& it : extraHeaders) {
+            ss << it.first << ":" << it.second << "\r\n";
+        }
         if (_enablePerMessageDeflate)
         {
             ss << _perMessageDeflateOptions.generateHeader();

ixwebsocket/IXWebSocketHandshake.h

@@ -16,6 +16,7 @@
 #include <memory>
 #include <string>
 #include <tuple>
+#include <unordered_map>
 
 namespace ix
 {
@@ -50,11 +51,13 @@ namespace ix
                            WebSocketPerMessageDeflateOptions& perMessageDeflateOptions,
                            std::atomic<bool>& enablePerMessageDeflate);
 
-        WebSocketInitResult clientHandshake(const std::string& url,
+        WebSocketInitResult clientHandshake(
-                                            const std::string& host,
+            const std::string& url,
-                                            const std::string& path,
+            const WebSocketHttpHeaders& extraHeaders,
-                                            int port,
+            const std::string& host,
-                                            int timeoutSecs);
+            const std::string& path,
+            int port,
+            int timeoutSecs);
 
         WebSocketInitResult serverHandshake(int fd, int timeoutSecs);
 

ixwebsocket/IXWebSocketTransport.cpp

@@ -127,8 +127,10 @@ namespace ix
     }
 
     // Client
-    WebSocketInitResult WebSocketTransport::connectToUrl(const std::string& url,
+    WebSocketInitResult WebSocketTransport::connectToUrl(
-                                                         int timeoutSecs)
+        const std::string& url,
+        const WebSocketHttpHeaders& headers,
+        int timeoutSecs)
     {
         std::lock_guard<std::mutex> lock(_socketMutex);
 
@@ -156,8 +158,8 @@ namespace ix
                                               _perMessageDeflateOptions,
                                               _enablePerMessageDeflate);
 
-        auto result = webSocketHandshake.clientHandshake(url, host, path, port,
+        auto result = webSocketHandshake.clientHandshake(url, headers, host, path,
-                                                         timeoutSecs);
+                                                         port, timeoutSecs);
         if (result.success)
         {
             setReadyState(ReadyState::OPEN);

ixwebsocket/IXWebSocketTransport.h

@@ -24,6 +24,7 @@
 #include <memory>
 #include <mutex>
 #include <string>
+
 #include <vector>
 
 namespace ix
@@ -75,8 +76,10 @@ namespace ix
                        int pingIntervalSecs,
                        int pingTimeoutSecs);
 
-        WebSocketInitResult connectToUrl(const std::string& url, // Client
+        WebSocketInitResult connectToUrl( // Client
-                                         int timeoutSecs);
+            const std::string& url,
+            const WebSocketHttpHeaders& headers,
+            int timeoutSecs);
         WebSocketInitResult connectToSocket(int fd, // Server
                                             int timeoutSecs);
 

makefile

@@ -17,6 +17,9 @@ ws:
 uninstall:
 	xargs rm -fv < build/install_manifest.txt
 
+tag:
+	git tag v"`cat DOCKER_VERSION`"
+
 .PHONY: docker
 
 NAME   := bsergean/ws

test/IXTest.h

@@ -6,9 +6,9 @@
 
 #pragma once
 
+#include "IXGetFreePort.h"
 #include <iostream>
 #include <ixwebsocket/IXWebSocketServer.h>
-#include "IXGetFreePort.h"
 #include <mutex>
 #include <spdlog/spdlog.h>
 #include <sstream>

ws/ixcobra/IXCobraConnection.cpp

@@ -24,7 +24,8 @@ namespace ix
         _webSocket(new WebSocket()),
         _publishMode(CobraConnection_PublishMode_Immediate),
         _authenticated(false),
-        _eventCallback(nullptr)
+        _eventCallback(nullptr),
+        _id(0)
     {
         _pdu["action"] = "rtm/publish";
 
@@ -244,6 +245,7 @@ namespace ix
         Json::Value pdu;
         pdu["action"] = "auth/handshake";
         pdu["body"] = body;
+        pdu["id"] = _id++;
 
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
@@ -306,6 +308,7 @@ namespace ix
         Json::Value pdu;
         pdu["action"] = "auth/authenticate";
         pdu["body"] = body;
+        pdu["id"] = _id++;
 
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
@@ -402,6 +405,7 @@ namespace ix
         _body["channels"] = channels;
         _body["message"] = msg;
         _pdu["body"] = _body;
+        _pdu["id"] = _id++;
 
         std::string serializedJson = serializeJson(_pdu);
 
@@ -429,15 +433,22 @@ namespace ix
     }
 
     void CobraConnection::subscribe(const std::string& channel,
-                                     SubscriptionCallback cb)
+                                    const std::string& filter,
+                                    SubscriptionCallback cb)
     {
         // Create and send a subscribe pdu
         Json::Value body;
         body["channel"] = channel;
 
+        if (!filter.empty())
+        {
+            body["filter"] = filter;
+        }
+
         Json::Value pdu;
         pdu["action"] = "rtm/subscribe";
         pdu["body"] = body;
+        pdu["id"] = _id++;
 
         _webSocket->send(pdu.toStyledString());
 
@@ -463,6 +474,7 @@ namespace ix
         Json::Value pdu;
         pdu["action"] = "rtm/unsubscribe";
         pdu["body"] = body;
+        pdu["id"] = _id++;
 
         _webSocket->send(pdu.toStyledString());
     }

ws/ixcobra/IXCobraConnection.h

@@ -75,7 +75,9 @@ namespace ix
 
         // Subscribe to a channel, and execute a callback when an incoming
         // message arrives.
-        void subscribe(const std::string& channel, SubscriptionCallback cb);
+        void subscribe(const std::string& channel,
+                       const std::string& filter = std::string(),
+                       SubscriptionCallback cb = nullptr);
 
         /// Unsubscribe from a channel
         void unsubscribe(const std::string& channel);
@@ -166,6 +168,9 @@ namespace ix
 
         // Cap the queue size (100 elems so far -> ~100k)
         static constexpr size_t kQueueMaxSize = 256;
+
+        // Each pdu sent should have an incremental unique id
+        std::atomic<uint64_t> _id;
     };
 
 } // namespace ix

ws/ixcobra/IXCobraMetricsPublisher.cpp

@@ -17,7 +17,7 @@ namespace ix
     const std::string CobraMetricsPublisher::kSetBlacklistId = "sms_set_blacklist_id";
 
     CobraMetricsPublisher::CobraMetricsPublisher() :
-        _enabled(false)
+        _enabled(true)
     {
     }
 
@@ -191,6 +191,19 @@ namespace ix
             msg["device"] = _device;
         }
 
+        {
+            //
+            // Bump a counter for each id
+            // This is used to make sure that we are not
+            // dropping messages, by checking that all the ids is the list of
+            // all natural numbers until the last value sent (0, 1, 2, ..., N)
+            //
+            std::lock_guard<std::mutex> lock(_device_mutex);
+            auto it = _counters.emplace(id, 0);
+            msg["per_id_counter"] = it.first->second;
+            it.first->second += 1;
+        }
+
         // Now actually enqueue the task
         _cobra_metrics_theaded_publisher.push(msg);
     }

ws/ixcobra/IXCobraMetricsPublisher.h

@@ -7,6 +7,7 @@
 #pragma once
 
 #include "IXCobraMetricsThreadedPublisher.h"
+#include <atomic>
 #include <chrono>
 #include <jsoncpp/json/json.h>
 #include <string>
@@ -132,8 +133,8 @@ namespace ix
         CobraMetricsThreadedPublisher _cobra_metrics_theaded_publisher;
 
         /// A boolean to enable or disable this system
-        /// push becomes a no-op when _enabled is true
+        /// push becomes a no-op when _enabled is false
-        bool _enabled;
+        std::atomic<bool> _enabled;
 
         /// A uuid used to uniquely identify a session
         std::string _session;
@@ -150,6 +151,10 @@ namespace ix
             _last_update;
         mutable std::mutex _last_update_mutex; // protect access to _last_update
 
+        /// Bump a counter for each metric type
+        std::unordered_map<std::string, int> _counters;
+        mutable std::mutex _counters_mutex; // protect access to _counters
+
         // const strings for internal ids
         static const std::string kSetRateControlId;
         static const std::string kSetBlacklistId;

ws/ws.cpp

@@ -9,15 +9,10 @@
 //
 #include "ws.h"
 
-//
-// Main drive for websocket utilities
-//
-
 #include <string>
 #include <sstream>
 #include <iostream>
 #include <fstream>
-// #include <unistd.h>
 
 #include <cli11/CLI11.hpp>
 #include <spdlog/spdlog.h>
@@ -60,6 +55,7 @@ int main(int argc, char** argv)
     std::string hostname("127.0.0.1");
     std::string pidfile;
     std::string channel;
+    std::string filter;
     std::string message;
     std::string password;
     std::string appkey;
@@ -76,6 +72,7 @@ int main(int argc, char** argv)
     bool followRedirects = false;
     bool verbose = false;
     bool save = false;
+    bool quiet = false;
     bool compress = false;
     bool strict = false;
     bool stress = false;
@@ -112,6 +109,7 @@ int main(int argc, char** argv)
 
     CLI::App* connectApp = app.add_subcommand("connect", "Connect to a remote server");
     connectApp->add_option("url", url, "Connection url")->required();
+    connectApp->add_option("-H", headers, "Header")->join();
     connectApp->add_flag("-d", disableAutomaticReconnection, "Disable Automatic Reconnection");
     connectApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
     connectApp->add_flag("-b", binaryMode, "Send in binary mode");
@@ -170,6 +168,8 @@ int main(int argc, char** argv)
     cobraSubscribeApp->add_option("--rolesecret", rolesecret, "Role secret");
     cobraSubscribeApp->add_option("channel", channel, "Channel")->required();
     cobraSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
+    cobraSubscribeApp->add_option("--filter", filter, "Stream SQL Filter");
+    cobraSubscribeApp->add_flag("-q", quiet, "Quiet / only display stats");
 
     CLI::App* cobraPublish = app.add_subcommand("cobra_publish", "Cobra publisher");
     cobraPublish->add_option("--appkey", appkey, "Appkey");
@@ -194,6 +194,7 @@ int main(int argc, char** argv)
     cobra2statsd->add_option("channel", channel, "Channel")->required();
     cobra2statsd->add_flag("-v", verbose, "Verbose");
     cobra2statsd->add_option("--pidfile", pidfile, "Pid file");
+    cobra2statsd->add_option("--filter", filter, "Stream SQL Filter");
 
     CLI::App* cobra2sentry = app.add_subcommand("cobra_to_sentry", "Cobra to sentry");
     cobra2sentry->add_option("--appkey", appkey, "Appkey");
@@ -206,6 +207,7 @@ int main(int argc, char** argv)
     cobra2sentry->add_flag("-v", verbose, "Verbose");
     cobra2sentry->add_flag("-s", strict, "Strict mode. Error out when sending to sentry fails");
     cobra2sentry->add_option("--pidfile", pidfile, "Pid file");
+    cobra2sentry->add_option("--filter", filter, "Stream SQL Filter");
 
     CLI::App* runApp = app.add_subcommand("snake", "Snake server");
     runApp->add_option("--port", port, "Connection url");
@@ -251,7 +253,7 @@ int main(int argc, char** argv)
     }
     else if (app.got_subcommand("connect"))
     {
-        ret = ix::ws_connect_main(url, disableAutomaticReconnection,
+        ret = ix::ws_connect_main(url, headers, disableAutomaticReconnection,
                                   disablePerMessageDeflate, binaryMode);
     }
     else if (app.got_subcommand("chat"))
@@ -290,7 +292,7 @@ int main(int argc, char** argv)
     {
         ret = ix::ws_cobra_subscribe_main(appkey, endpoint,
                                           rolename, rolesecret,
-                                          channel);
+                                          channel, filter, quiet);
     }
     else if (app.got_subcommand("cobra_publish"))
     {
@@ -302,14 +304,14 @@ int main(int argc, char** argv)
     {
         ret = ix::ws_cobra_to_statsd_main(appkey, endpoint,
                                           rolename, rolesecret,
-                                          channel, hostname, statsdPort,
+                                          channel, filter, hostname, statsdPort,
                                           prefix, fields, verbose);
     }
     else if (app.got_subcommand("cobra_to_sentry"))
     {
         ret = ix::ws_cobra_to_sentry_main(appkey, endpoint,
                                           rolename, rolesecret,
-                                          channel, dsn,
+                                          channel, filter, dsn,
                                           verbose, strict, jobs);
     }
     else if (app.got_subcommand("snake"))

ws/ws.h

@@ -31,6 +31,7 @@ namespace ix
     int ws_chat_main(const std::string& url, const std::string& user);
 
     int ws_connect_main(const std::string& url,
+                        const std::string& headers,
                         bool disableAutomaticReconnection,
                         bool disablePerMessageDeflate,
                         bool binaryMode);
@@ -56,7 +57,9 @@ namespace ix
                                 const std::string& endpoint,
                                 const std::string& rolename,
                                 const std::string& rolesecret,
-                                const std::string& channel);
+                                const std::string& channel,
+                                const std::string& filter,
+                                bool quiet);
 
     int ws_cobra_publish_main(const std::string& appkey,
                               const std::string& endpoint,
@@ -71,6 +74,7 @@ namespace ix
                                 const std::string& rolename,
                                 const std::string& rolesecret,
                                 const std::string& channel,
+                                const std::string& filter,
                                 const std::string& host,
                                 int port,
                                 const std::string& prefix,
@@ -82,6 +86,7 @@ namespace ix
                                 const std::string& rolename,
                                 const std::string& rolesecret,
                                 const std::string& channel,
+                                const std::string& filter,
                                 const std::string& dsn,
                                 bool verbose,
                                 bool strict,

ws/ws_cobra_subscribe.cpp

@@ -11,13 +11,17 @@
 #include <atomic>
 #include <ixcobra/IXCobraConnection.h>
 
+#include <spdlog/spdlog.h>
+
 namespace ix
 {
     int ws_cobra_subscribe_main(const std::string& appkey,
                                 const std::string& endpoint,
                                 const std::string& rolename,
                                 const std::string& rolesecret,
-                                const std::string& channel)
+                                const std::string& channel,
+                                const std::string& filter,
+                                bool quiet)
     {
 
         ix::CobraConnection conn;
@@ -28,8 +32,28 @@ namespace ix
 
         Json::FastWriter jsonWriter;
 
+        // Display incoming messages
+        std::atomic<int> msgPerSeconds(0);
+        std::atomic<int> msgCount(0);
+        
+        auto timer = [&msgPerSeconds, &msgCount]
+        {
+            while (true)
+            {
+                std::cout << "#messages " << msgCount << " "
+                          << "msg/s " << msgPerSeconds
+                          << std::endl;
+
+                msgPerSeconds = 0;
+                auto duration = std::chrono::seconds(1);
+                std::this_thread::sleep_for(duration);
+            }
+        };
+
+        std::thread t(timer);
+
         conn.setEventCallback(
-            [&conn, &channel, &jsonWriter]
+            [&conn, &channel, &jsonWriter, &filter, &msgCount, &msgPerSeconds, &quiet]
             (ix::CobraConnectionEventType eventType,
              const std::string& errMsg,
              const ix::WebSocketHttpHeaders& headers,
@@ -37,33 +61,40 @@ namespace ix
             {
                 if (eventType == ix::CobraConnection_EventType_Open)
                 {
-                    std::cout << "Subscriber: connected" << std::endl;
+                    spdlog::info("Subscriber connected");
 
                     for (auto it : headers)
                     {
-                        std::cerr << it.first << ": " << it.second << std::endl;
+                        spdlog::info("{}: {}", it.first, it.second);
                     }
                 }
                 else if (eventType == ix::CobraConnection_EventType_Authenticated)
                 {
-                    std::cout << "Subscriber authenticated" << std::endl;
+                    spdlog::info("Subscriber authenticated");
-                    conn.subscribe(channel,
+                    conn.subscribe(channel, filter,
-                                   [&jsonWriter](const Json::Value& msg)
+                                   [&jsonWriter, &quiet,
+                                    &msgPerSeconds, &msgCount](const Json::Value& msg)
                                    {
-                                       std::cout << jsonWriter.write(msg) << std::endl;
+                                       if (!quiet)
+                                       {
+                                           std::cout << jsonWriter.write(msg) << std::endl;
+                                       }
+
+                                       msgPerSeconds++;
+                                       msgCount++;
                                    });
                 }
                 else if (eventType == ix::CobraConnection_EventType_Subscribed)
                 {
-                    std::cout << "Subscriber: subscribed to channel " << subscriptionId << std::endl;
+                    spdlog::info("Subscriber: subscribed to channel {}", subscriptionId);
                 }
                 else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
                 {
-                    std::cout << "Subscriber: unsubscribed from channel " << subscriptionId << std::endl;
+                    spdlog::info("Subscriber: unsubscribed from channel {}", subscriptionId);
                 }
                 else if (eventType == ix::CobraConnection_EventType_Error)
                 {
-                    std::cout << "Subscriber: error" << errMsg << std::endl;
+                    spdlog::error("Subscriber: error {}", errMsg);
                 }
             }
         );

ws/ws_cobra_to_sentry.cpp

@@ -25,6 +25,7 @@ namespace ix
                                 const std::string& rolename,
                                 const std::string& rolesecret,
                                 const std::string& channel,
+                                const std::string& filter,
                                 const std::string& dsn,
                                 bool verbose,
                                 bool strict,
@@ -94,7 +95,7 @@ namespace ix
         }
 
         conn.setEventCallback(
-            [&conn, &channel, &jsonWriter,
+            [&conn, &channel, &filter, &jsonWriter,
              verbose, &receivedCount, &sentCount,
              &condition, &conditionVariableMutex,
              &progressCondition, &queue]
@@ -119,7 +120,7 @@ namespace ix
                 else if (eventType == ix::CobraConnection_EventType_Authenticated)
                 {
                     std::cerr << "Subscriber authenticated" << std::endl;
-                    conn.subscribe(channel,
+                    conn.subscribe(channel, filter,
                                    [&jsonWriter, verbose,
                                     &sentCount, &receivedCount,
                                     &condition, &conditionVariableMutex,

ws/ws_cobra_to_statsd.cpp

@@ -63,6 +63,7 @@ namespace ix
                                 const std::string& rolename,
                                 const std::string& rolesecret,
                                 const std::string& channel,
+                                const std::string& filter,
                                 const std::string& host,
                                 int port,
                                 const std::string& prefix,
@@ -90,7 +91,7 @@ namespace ix
         uint64_t msgCount = 0;
 
         conn.setEventCallback(
-            [&conn, &channel, &jsonWriter, &statsdClient, verbose, &tokens, &prefix, &msgCount]
+            [&conn, &channel, &filter, &jsonWriter, &statsdClient, verbose, &tokens, &prefix, &msgCount]
             (ix::CobraConnectionEventType eventType,
              const std::string& errMsg,
              const ix::WebSocketHttpHeaders& headers,
@@ -112,7 +113,7 @@ namespace ix
                 else if (eventType == ix::CobraConnection_EventType_Authenticated)
                 {
                     spdlog::info("Subscriber authenticated");
-                    conn.subscribe(channel,
+                    conn.subscribe(channel, filter,
                                    [&jsonWriter, &statsdClient,
                                     verbose, &tokens, &prefix, &msgCount]
                                    (const Json::Value& msg)

ws/ws_connect.cpp

@@ -18,6 +18,7 @@ namespace ix
     {
         public:
             WebSocketConnect(const std::string& _url,
+                             const std::string& headers,
                              bool disableAutomaticReconnection,
                              bool disablePerMessageDeflate,
                              bool binaryMode);
@@ -30,14 +31,17 @@ namespace ix
 
         private:
             std::string _url;
+            WebSocketHttpHeaders _headers;
             ix::WebSocket _webSocket;
             bool _disablePerMessageDeflate;
             bool _binaryMode;
 
             void log(const std::string& msg);
+            WebSocketHttpHeaders parseHeaders(const std::string& data);
     };
 
     WebSocketConnect::WebSocketConnect(const std::string& url,
+                                       const std::string& headers,
                                        bool disableAutomaticReconnection,
                                        bool disablePerMessageDeflate,
                                        bool binaryMode) :
@@ -49,6 +53,8 @@ namespace ix
         {
             _webSocket.disableAutomaticReconnection();
         }
+
+        _headers = parseHeaders(headers);
     }
 
     void WebSocketConnect::log(const std::string& msg)
@@ -56,6 +62,31 @@ namespace ix
         std::cout << msg << std::endl;
     }
 
+    WebSocketHttpHeaders WebSocketConnect::parseHeaders(const std::string& data)
+    {
+        WebSocketHttpHeaders headers;
+
+        // Split by \n
+        std::string token;
+        std::stringstream tokenStream(data);
+
+        while (std::getline(tokenStream, token))
+        {
+            std::size_t pos = token.rfind(':');
+
+            // Bail out if last '.' is found
+            if (pos == std::string::npos) continue;
+
+            auto key = token.substr(0, pos);
+            auto val = token.substr(pos+1);
+
+            std::cerr << key << ": " << val << std::endl;
+            headers[key] = val;
+        }
+
+        return headers;
+    }
+
     void WebSocketConnect::stop()
     {
         _webSocket.stop();
@@ -64,6 +95,7 @@ namespace ix
     void WebSocketConnect::start()
     {
         _webSocket.setUrl(_url);
+        _webSocket.setExtraHeaders(_headers);
 
         if (_disablePerMessageDeflate)
         {
@@ -151,12 +183,14 @@ namespace ix
     }
 
     int ws_connect_main(const std::string& url,
+                        const std::string& headers,
                         bool disableAutomaticReconnection,
                         bool disablePerMessageDeflate,
                         bool binaryMode)
     {
         std::cout << "Type Ctrl-D to exit prompt..." << std::endl;
         WebSocketConnect webSocketChat(url,
+                                       headers,
                                        disableAutomaticReconnection,
                                        disablePerMessageDeflate,
                                        binaryMode);

ws/ws_http_client.cpp

@@ -44,7 +44,7 @@ namespace ix
             if (pos == std::string::npos) continue;
 
             auto key = token.substr(0, pos);
-            auto val = token.substr(pos+2);
+            auto val = token.substr(pos+1);
 
             std::cerr << key << ": " << val << std::endl;
             headers[key] = val;