play with cpp_redis

CMakeLists.txt

@@ -37,6 +37,7 @@ set( IXWEBSOCKET_SOURCES
     ixwebsocket/IXHttpClient.cpp
     ixwebsocket/IXUrlParser.cpp
     ixwebsocket/IXSelectInterrupt.cpp
+    ixwebsocket/IXSelectInterruptPipe.cpp
     ixwebsocket/IXSelectInterruptFactory.cpp
     ixwebsocket/IXConnectionState.cpp
 )
@@ -64,16 +65,11 @@ set( IXWEBSOCKET_HEADERS
     ixwebsocket/IXHttpClient.h
     ixwebsocket/IXUrlParser.h
     ixwebsocket/IXSelectInterrupt.h
+    ixwebsocket/IXSelectInterruptPipe.h
     ixwebsocket/IXSelectInterruptFactory.h
     ixwebsocket/IXConnectionState.h
 )
 
-if (UNIX)
-    # Linux, Mac, iOS, Android
-    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptPipe.cpp )
-    list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSelectInterruptPipe.h )
-endif()
-
 # Platform specific code
 if (APPLE)
     list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/apple/IXSetThreadName_apple.cpp)
@@ -85,7 +81,6 @@ else()
     list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSelectInterruptEventFd.h)
 endif()
 
-set(USE_OPEN_SSL FALSE)
 if (USE_TLS)
     add_definitions(-DIXWEBSOCKET_USE_TLS)
 
@@ -96,7 +91,6 @@ if (USE_TLS)
         list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSocketSChannel.h)
         list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSocketSChannel.cpp)
     else()
-        set(USE_OPEN_SSL TRUE)
         list( APPEND IXWEBSOCKET_HEADERS ixwebsocket/IXSocketOpenSSL.h)
         list( APPEND IXWEBSOCKET_SOURCES ixwebsocket/IXSocketOpenSSL.cpp)
     endif()
@@ -107,16 +101,14 @@ add_library( ixwebsocket STATIC
     ${IXWEBSOCKET_HEADERS}
 )
 
-if (APPLE AND USE_TLS)
-  target_link_libraries(ixwebsocket "-framework foundation" "-framework security")
-endif()
+# gcc/Linux needs -pthread
+find_package(Threads)
 
-if(USE_OPEN_SSL)
+if(UNIX AND NOT APPLE)
   find_package(OpenSSL REQUIRED)
   add_definitions(${OPENSSL_DEFINITIONS})
   message(STATUS "OpenSSL: " ${OPENSSL_VERSION})
   include_directories(${OPENSSL_INCLUDE_DIR})
-  target_link_libraries(ixwebsocket ${OPENSSL_LIBRARIES})
 endif()
 
 if (WIN32)
@@ -131,21 +123,16 @@ if (WIN32)
 
   target_link_libraries(ixwebsocket libz wsock32 ws2_32)
   add_definitions(-D_CRT_SECURE_NO_WARNINGS)
-
+  
 else()
-  # gcc/Linux needs -pthread
-  find_package(Threads)
-
   target_link_libraries(ixwebsocket 
-    z ${CMAKE_THREAD_LIBS_INIT})
+    z ${OPENSSL_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
 endif()
 
 set( IXWEBSOCKET_INCLUDE_DIRS
     .
-)
-
+    ../../shared/OpenSSL/include)
 target_include_directories( ixwebsocket PUBLIC ${IXWEBSOCKET_INCLUDE_DIRS} )
 
-if (NOT WIN32)
-    add_subdirectory(ws)
-endif()
+add_subdirectory(ws)
+add_subdirectory(third_party/cpp_redis)

DOCKER_VERSION

@@ -1 +1 @@
-1.4.0
+1.3.2

Dockerfile

@@ -1,47 +0,0 @@
-# Build time
-FROM debian:buster as build
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-RUN apt-get -y install wget 
-RUN mkdir -p /tmp/cmake
-WORKDIR /tmp/cmake
-RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
-RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
-
-RUN apt-get -y install g++ 
-RUN apt-get -y install libssl-dev
-RUN apt-get -y install libz-dev
-RUN apt-get -y install make
-
-COPY . .
-
-ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
-ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
-
-RUN ["make"]
-
-# Runtime
-FROM debian:buster as runtime
-
-ENV DEBIAN_FRONTEND noninteractive
-RUN apt-get update 
-# Runtime 
-RUN apt-get install -y libssl1.1 
-
-# Debugging
-RUN apt-get install -y strace
-RUN apt-get install -y gdb
-RUN apt-get install -y procps
-RUN apt-get install -y htop
-
-RUN adduser --disabled-password --gecos '' app
-COPY --chown=app:app --from=build /usr/local/bin/ws /usr/local/bin/ws
-RUN chmod +x /usr/local/bin/ws
-RUN ldd /usr/local/bin/ws
-
-# Now run in usermode
-USER app
-WORKDIR /home/app
-
-CMD ["ws"]

Dockerfile

@@ -0,0 +1 @@
+Dockerfile.dev

Dockerfile.dev

@@ -0,0 +1,31 @@
+FROM debian:stretch
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update 
+RUN apt-get -y install g++ 
+RUN apt-get -y install libssl-dev
+RUN apt-get -y install gdb
+RUN apt-get -y install screen
+RUN apt-get -y install procps
+RUN apt-get -y install lsof
+RUN apt-get -y install libz-dev
+RUN apt-get -y install vim
+RUN apt-get -y install make
+RUN apt-get -y install cmake
+RUN apt-get -y install curl
+RUN apt-get -y install python
+RUN apt-get -y install netcat
+
+# debian strech cmake is too old for building with Docker
+COPY makefile .
+RUN ["make", "install_cmake_for_linux"]
+
+COPY . .
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-rc4-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+# RUN ["make"]
+
+EXPOSE 8765
+CMD ["/ws/ws", "transfer", "--port", "8765", "--host", "0.0.0.0"]

Dockerfile.prod

@@ -0,0 +1,30 @@
+FROM debian:buster
+
+ENV DEBIAN_FRONTEND noninteractive
+RUN apt-get update 
+
+RUN apt-get -y install g++ 
+RUN apt-get -y install libssl-dev
+RUN apt-get -y install libz-dev
+RUN apt-get -y install make
+
+RUN apt-get -y install wget 
+RUN mkdir -p /tmp/cmake
+WORKDIR /tmp/cmake
+RUN wget https://github.com/Kitware/CMake/releases/download/v3.14.0/cmake-3.14.0-Linux-x86_64.tar.gz 
+RUN tar zxf cmake-3.14.0-Linux-x86_64.tar.gz
+
+RUN adduser app 
+
+COPY . .
+
+ARG CMAKE_BIN_PATH=/tmp/cmake/cmake-3.14.0-Linux-x86_64/bin
+ENV PATH="${CMAKE_BIN_PATH}:${PATH}"
+
+RUN ["make"]
+
+# Now run in usermode
+USER app
+
+EXPOSE 8765
+CMD ["bash"]

doc/redis_conf_2019/makefile

@@ -0,0 +1,2 @@
+all:
+	(cd .. ; make docker && make docker_push)

ixwebsocket/IXConnectionState.cpp

@@ -10,7 +10,7 @@ namespace ix
 {
     std::atomic<uint64_t> ConnectionState::_globalId(0);
 
-    ConnectionState::ConnectionState() : _terminated(false)
+    ConnectionState::ConnectionState()
     {
         computeId();
     }
@@ -29,15 +29,5 @@ namespace ix
     {
         return std::make_shared<ConnectionState>();
     }
-
-    bool ConnectionState::isTerminated() const
-    {
-        return _terminated;
-    }
-
-    bool ConnectionState::setTerminated()
-    {
-        _terminated = true;
-    }
 }
 

ixwebsocket/IXConnectionState.h

@@ -21,13 +21,9 @@ namespace ix
         virtual void computeId();
         virtual const std::string& getId() const;
 
-        bool setTerminated();
-        bool isTerminated() const;
-
         static std::shared_ptr<ConnectionState> createConnectionState();
 
     protected:
-        std::atomic<bool> _terminated;
         std::string _id;
 
         static std::atomic<uint64_t> _globalId;

ixwebsocket/IXSocket.cpp

@@ -21,10 +21,6 @@
 #include <algorithm>
 #include <iostream>
 
-#ifdef min
-#undef min
-#endif
-
 namespace ix
 {
     const int Socket::kDefaultPollNoTimeout = -1; // No poll timeout by default

ixwebsocket/IXSocketFactory.cpp

@@ -6,20 +6,12 @@
 
 #include "IXSocketFactory.h"
 
-#ifdef IXWEBSOCKET_USE_TLS
-
+#if defined(__APPLE__) or defined(__linux__)
 # ifdef __APPLE__
 #  include <ixwebsocket/IXSocketAppleSSL.h>
-# elif defined(_WIN32)
-#  include <ixwebsocket/IXSocketSChannel.h>
 # else
 #  include <ixwebsocket/IXSocketOpenSSL.h>
 # endif
-
-#else
-
-#include <ixwebsocket/IXSocket.h>
-
 #endif
 
 namespace ix
@@ -39,8 +31,6 @@ namespace ix
 #ifdef IXWEBSOCKET_USE_TLS
 # ifdef __APPLE__
             socket = std::make_shared<SocketAppleSSL>();
-# elif defined(_WIN32)
-            socket = std::make_shared<SocketSChannel>();
 # else
             socket = std::make_shared<SocketOpenSSL>();
 # endif

ixwebsocket/IXSocketFactory.h

@@ -8,7 +8,6 @@
 #pragma once
 
 #include <memory>
-#include <string>
 
 namespace ix
 {

ixwebsocket/IXSocketSChannel.cpp

@@ -18,7 +18,7 @@
 # include <ws2def.h>
 # include <WS2tcpip.h>
 # include <schannel.h>
-//# include <sslsock.h>
+# include <sslsock.h>
 # include <io.h>
 
 #define WIN32_LEAN_AND_MEAN
@@ -75,7 +75,7 @@ namespace ix
                                  int port,
                                  std::string& errMsg)
     {
-        return Socket::connect(host, port, errMsg, nullptr);
+        return Socket::connect(host, port, errMsg);
     }
 
 
@@ -89,17 +89,17 @@ namespace ix
         Socket::close();
     }
 
-    ssize_t SocketSChannel::send(char* buf, size_t nbyte)
+    int SocketSChannel::send(char* buf, size_t nbyte)
     {
         return Socket::send(buf, nbyte);
     }
 
-    ssize_t SocketSChannel::send(const std::string& buffer)
+    int SocketSChannel::send(const std::string& buffer)
     {
         return Socket::send(buffer);
     }
 
-    ssize_t SocketSChannel::recv(void* buf, size_t nbyte)
+    int SocketSChannel::recv(void* buf, size_t nbyte)
     {
         return Socket::recv(buf, nbyte);
     }

ixwebsocket/IXSocketSChannel.h

@@ -24,9 +24,9 @@ namespace ix
         // The important override
         virtual void secureSocket() final;
 
-        virtual ssize_t send(char* buffer, size_t length) final;
-        virtual ssize_t send(const std::string& buffer) final;
-        virtual ssize_t recv(void* buffer, size_t length) final;
+        virtual int send(char* buffer, size_t length) final;
+        virtual int send(const std::string& buffer) final;
+        virtual int recv(void* buffer, size_t length) final;
 
     private:
     };

ixwebsocket/IXSocketServer.cpp

@@ -136,9 +136,6 @@ namespace ix
 
     void SocketServer::stop()
     {
-        closeTerminatedThreads();
-        assert(_connectionsThreads.empty());
-
         if (!_thread.joinable()) return; // nothing to do
 
         _stop = true;
@@ -155,44 +152,18 @@ namespace ix
         _connectionStateFactory = connectionStateFactory;
     }
 
-    // join the threads for connections that have been closed
-    void SocketServer::closeTerminatedThreads()
-    {
-        auto it = _connectionsThreads.begin();
-        auto itEnd  = _connectionsThreads.end();
-
-        while (it != itEnd)
-        {
-            auto& connectionState = it->first;
-            auto& thread = it->second;
-
-            if (!connectionState->isTerminated() ||
-                !thread.joinable())
-            {
-                ++it;
-                continue;
-            }
-
-            thread.join();
-            it = _connectionsThreads.erase(it);
-        }
-    }
-
     void SocketServer::run()
     {
         // Set the socket to non blocking mode, so that accept calls are not blocking
         SocketConnect::configure(_serverFd);
 
+        // Return value of std::async, ignored
+        std::future<void> f;
+
         for (;;)
         {
             if (_stop) return;
 
-            // Garbage collection to shutdown/join threads for closed connections.
-            // We could run this in its own thread, so that we dont need to accept 
-            // a new connection to close a thread.
-            // We could also use a condition variable to be notify when we need to do this
-            closeTerminatedThreads();
-
             // Use select to check whether a new connection is in progress
             fd_set rfds;
             struct timeval timeout;
@@ -257,12 +228,14 @@ namespace ix
             }
 
             // Launch the handleConnection work asynchronously in its own thread.
-            _connectionsThreads.push_back(std::make_pair(
-                    connectionState,
-                    std::thread(&SocketServer::handleConnection,
-                                this,
-                                clientFd,
-                                connectionState)));
+            //
+            // the destructor of a future returned by std::async blocks,
+            // so we need to declare it outside of this loop
+            f = std::async(std::launch::async,
+                           &SocketServer::handleConnection,
+                           this,
+                           clientFd,
+                           connectionState);
         }
     }
 }

ixwebsocket/IXSocketServer.h

@@ -12,7 +12,6 @@
 #include <string>
 #include <set>
 #include <thread>
-#include <list>
 #include <mutex>
 #include <functional>
 #include <memory>
@@ -25,10 +24,6 @@ namespace ix
     public:
         using ConnectionStateFactory = std::function<std::shared_ptr<ConnectionState>()>;
 
-        // We use a list as we only care about remove and append operations.
-        using ConnectionThreads = std::list<std::pair<std::shared_ptr<ConnectionState>,
-                                                      std::thread>>;
-
         SocketServer(int port = SocketServer::kDefaultPort,
                      const std::string& host = SocketServer::kDefaultHost,
                      int backlog = SocketServer::kDefaultTcpBacklog,
@@ -68,8 +63,6 @@ namespace ix
         std::atomic<bool> _stop;
         std::thread _thread;
 
-        ConnectionThreads _connectionsThreads;
-
         std::condition_variable _conditionVariable;
         std::mutex _conditionVariableMutex;
 
@@ -81,7 +74,5 @@ namespace ix
         virtual void handleConnection(int fd,
                                       std::shared_ptr<ConnectionState> connectionState) = 0;
         virtual size_t getConnectedClientsCount() = 0;
-
-        void closeTerminatedThreads();
     };
 }

ixwebsocket/IXWebSocketHttpHeaders.cpp

@@ -6,28 +6,12 @@
 
 #include "IXWebSocketHttpHeaders.h"
 #include "IXSocket.h"
-#include <algorithm>
-#include <locale>
+
+#include <string>
+#include <unordered_map>
 
 namespace ix
 {
-    bool CaseInsensitiveLess::NocaseCompare::operator()(const unsigned char & c1, const unsigned char & c2) const
-    {
-#ifdef _WIN32
-        return std::tolower(c1, std::locale()) < std::tolower(c2, std::locale());
-#else
-        return std::tolower(c1) < std::tolower(c2);
-#endif
-    }
-
-    bool CaseInsensitiveLess::operator()(const std::string & s1, const std::string & s2) const
-    {
-        return std::lexicographical_compare
-                (s1.begin(), s1.end(),   // source range
-                 s2.begin(), s2.end(),   // dest range
-                 NocaseCompare());       // comparison
-    }
-
     std::pair<bool, WebSocketHttpHeaders> parseHttpHeaders(
         std::shared_ptr<Socket> socket,
         const CancellationRequest& isCancellationRequested)

ixwebsocket/IXWebSocketHttpHeaders.h

@@ -11,6 +11,7 @@
 #include <string>
 #include <map>
 #include <memory>
+#include <algorithm>
 
 namespace ix
 {
@@ -21,10 +22,19 @@ namespace ix
         // Case Insensitive compare_less binary function
         struct NocaseCompare
         {
-            bool operator() (const unsigned char& c1, const unsigned char& c2) const;
+            bool operator() (const unsigned char& c1, const unsigned char& c2) const
+            {
+                return std::tolower(c1) < std::tolower(c2);
+            }
         };
 
-        bool operator() (const std::string & s1, const std::string & s2) const;
+        bool operator() (const std::string & s1, const std::string & s2) const
+        {
+            return std::lexicographical_compare
+                (s1.begin(), s1.end(),   // source range
+                 s2.begin(), s2.end(),   // dest range
+                 NocaseCompare());  // comparison
+        }
     };
 
     using WebSocketHttpHeaders = std::map<std::string, std::string, CaseInsensitiveLess>;

ixwebsocket/IXWebSocketServer.cpp

@@ -91,7 +91,6 @@ namespace ix
         }
 
         logInfo("WebSocketServer::handleConnection() done");
-        connectionState->setTerminated();
     }
 
     std::set<std::shared_ptr<WebSocket>> WebSocketServer::getClients()

ixwebsocket/IXWebSocketTransport.cpp

@@ -474,8 +474,14 @@ namespace ix
                 std::string reason(_rxbuf.begin()+ws.header_size + 2,
                                    _rxbuf.begin()+ws.header_size + 2 + (size_t) ws.N);
 
+                {
+                    std::lock_guard<std::mutex> lock(_closeDataMutex);
+                    _closeCode = code;
+                    _closeReason = reason;
+                    _closeWireSize = _rxbuf.size();
+                }
 
-                close(code, reason, _rxbuf.size());
+                close();
             }
             else
             {
@@ -638,7 +644,7 @@ namespace ix
                                           std::string::const_iterator message_end,
                                           bool compress)
     {
-        uint64_t message_size = static_cast<uint64_t>(message_end - message_begin);
+        auto message_size = message_end - message_begin;
 
         unsigned x = getRandomUnsigned();
         uint8_t masking_key[4] = {};
@@ -775,7 +781,7 @@ namespace ix
         _lastSendTimePoint = std::chrono::steady_clock::now();
     }
 
-    void WebSocketTransport::close(uint16_t code, const std::string& reason, size_t closeWireSize)
+    void WebSocketTransport::close()
     {
         _requestInitCancellation = true;
 
@@ -783,22 +789,21 @@ namespace ix
 
         // See list of close events here:
         // https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
-        const std::string closure{(char)(code >> 8), (char)(code & 0xff)};
-
+        // We use 1000: normal closure.
+        //
+        // >>> struct.pack('!H', 1000)
+        // b'\x03\xe8'
+        //
+        const std::string normalClosure = std::string("\x03\xe8");
         bool compress = false;
-        sendData(wsheader_type::CLOSE, closure, compress);
+        sendData(wsheader_type::CLOSE, normalClosure, compress);
         setReadyState(CLOSING);
 
         _socket->wakeUpFromPoll(Socket::kCloseRequest);
         _socket->close();
 
-        {
-            std::lock_guard<std::mutex> lock(_closeDataMutex);
-            _closeCode = code;
-            _closeReason = reason;
-            _closeWireSize = closeWireSize;
-        }
-        
+        _closeCode = 1000;
+        _closeReason = "Normal Closure";
         setReadyState(CLOSED);
     }
 

ixwebsocket/IXWebSocketTransport.h

@@ -81,11 +81,7 @@ namespace ix
         WebSocketSendInfo sendText(const std::string& message,
                                    const OnProgressCallback& onProgressCallback);
         WebSocketSendInfo sendPing(const std::string& message);
-
-        void close(uint16_t code = 1000,
-                   const std::string& reason = "Normal closure",
-                   size_t closeWireSize = 0);
-
+        void close();
         ReadyStateValues getReadyState() const;
         void setReadyState(ReadyStateValues readyStateValue);
         void setOnCloseCallback(const OnCloseCallback& onCloseCallback);

makefile

@@ -48,7 +48,7 @@ test_server:
 # env TEST=Websocket_chat make test
 # env TEST=heartbeat make test
 test:
-	(cd test ; python2.7 run.py)
+	python test/run.py
 
 ws_test: all
 	(cd ws ; bash test_ws.sh)

test/IXSocketTest.cpp

@@ -73,7 +73,7 @@ TEST_CASE("socket", "[socket]")
         testSocket(host, port, request, socket, expectedStatus, timeoutSecs);
     }
 
-#if defined(__APPLE__) || defined(__linux__)
+#if defined(__APPLE__) or defined(__linux__)
     SECTION("Connect to google HTTPS server. Send GET request without header. Should return 200")
     {
         std::string errMsg;

test/IXTest.cpp

@@ -109,7 +109,7 @@ namespace ix
         }
 
         struct sockaddr_in sa; // server address information
-        socklen_t len;
+        unsigned int len;
         if (getsockname(sockfd, (struct sockaddr *) &sa, &len) < 0)
         {
             log("Cannot compute a free port. getsockname error.");

test/run.py

@@ -1,31 +1,9 @@
-#!/usr/bin/env python2.7
-'''
-'''
-
-from __future__ import print_function
-
 import os
-import sys
 import platform
-import argparse
-import multiprocessing
-import tempfile
-import time
-import datetime
-import threading
+import shutil
+
 import subprocess
-import re
-import xml.etree.ElementTree as ET
-from xml.dom import minidom
-
-hasClick = True
-try:
-    import click
-except ImportError:
-    hasClick = False
-
-
-DEFAULT_EXE = 'ixwebsocket_unittest'
+import threading
 
 
 class Command(object):
@@ -38,16 +16,12 @@ class Command(object):
         self.cmd = cmd
         self.process = None
 
-    def run_command(self):
+    def run_command(self, capture = False):
         self.process = subprocess.Popen(self.cmd, shell=True)
         self.process.communicate()
 
-    def run(self, timeout=None):
+    def run(self, timeout = 5 * 60):
         '''5 minutes default timeout'''
-        
-        if timeout is None:
-            timeout = 5 * 60
-
         thread = threading.Thread(target=self.run_command, args=())
         thread.start()
         thread.join(timeout)
@@ -61,424 +35,85 @@ class Command(object):
             return True, self.process.returncode
 
 
-def runCommand(cmd, assertOnFailure=True, timeout=None):
-    '''Small wrapper to run a command and make sure it succeed'''
-
-    if timeout is None:
-        timeout = 30 * 60 # 30 minute default timeout
-
-    print('\nRunning', cmd)
-    command = Command(cmd)
-    timedout, ret = command.run(timeout)
-
-    if timedout:
-        print('Unittest timed out')
-
-    msg = 'cmd {} failed with error code {}'.format(cmd, ret)
-    if ret != 0:
-        print(msg)
-        if assertOnFailure:
-            assert False
-
-
-def runCMake(sanitizer, buildDir):
-    '''Generate a makefile from CMake.
-    We do an out of dir build, so that cleaning up is easy
-    (remove build sub-folder).
-    '''
-
-    # CMake installed via Self Service ends up here.
-    cmake_executable = '/Applications/CMake.app/Contents/bin/cmake'
-
-    if not os.path.exists(cmake_executable):
-        cmake_executable = 'cmake'
-
-    sanitizersFlags = {
-        'asan': '-DSANITIZE_ADDRESS=On',
-        'ubsan': '-DSANITIZE_UNDEFINED=On',
-        'tsan': '-DSANITIZE_THREAD=On',
-        'none': ''
-    }
-    sanitizerFlag = sanitizersFlags[sanitizer]
-
-    # CMake installed via Self Service ends up here.
-    cmakeExecutable = '/Applications/CMake.app/Contents/bin/cmake'
-    if not os.path.exists(cmakeExecutable):
-        cmakeExecutable = 'cmake'
-
-    fmt = '''
-{cmakeExecutable} -H. \
-    {sanitizerFlag} \
-    -B{buildDir} \
-    -DCMAKE_BUILD_TYPE=Debug \
-    -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
-'''
-    cmakeCmd = fmt.format(**locals())
-    runCommand(cmakeCmd)
-
-
-def runTest(args, buildDir, xmlOutput, testRunName):
-    '''Execute the unittest.
-    '''
-    if args is None:
-        args = ''
-
-    fmt = '{buildDir}/{DEFAULT_EXE} -o {xmlOutput} -n "{testRunName}" -r junit "{args}"'
-    testCommand = fmt.format(**locals())
-    runCommand(testCommand,
-               assertOnFailure=False)
-
-
-def validateTestSuite(xmlOutput):
-    '''
-    Parse the output XML file to validate that all tests passed.
-
-    Assume that the XML file contains only one testsuite.
-    (which is true when generate by catch2)
-    '''
-    tree = ET.parse(xmlOutput)
-    root = tree.getroot()
-    testSuite = root[0]
-    testSuiteAttributes = testSuite.attrib
-
-    tests = testSuiteAttributes['tests']
-
-    success = True
-
-    for testcase in testSuite:
-        if testcase.tag != 'testcase':
-            continue
-
-        testName = testcase.attrib['name']
-        systemOutput = None
-
-        for child in testcase:
-            if child.tag == 'system-out':
-                systemOutput = child.text
-
-            if child.tag == 'failure':
-                success = False
-
-                print("Testcase '{}' failed".format(testName))
-                print(' ', systemOutput)
-
-    return success, tests
-
-
-def log(msg, color):
-    if hasClick:
-        click.secho(msg, fg=color)
-    else:
-        print(msg)
-
-
-def isSuccessFullRun(output):
-    '''When being run from lldb, we cannot capture the exit code 
-    so we have to parse the output which is produced in a 
-    consistent way. Whenever we'll be on a recent enough version of lldb we 
-    won't have to do this.
-    '''
-    pid = None
-    matchingPids = False
-    exitCode = -1
-
-    # 'Process 279 exited with status = 1 (0x00000001) ',
-    exitPattern = re.compile('^Process (?P<pid>[0-9]+) exited with status = (?P<exitCode>[0-9]+)')
-
-    # "Process 99232 launched: '/Users/bse...
-    launchedPattern = re.compile('^Process (?P<pid>[0-9]+) launched: ')
-
-    for line in output:
-        match = exitPattern.match(line)
-        if match:
-            exitCode = int(match.group('exitCode'))
-            pid = match.group('pid')
-
-        match = launchedPattern.match(line)
-        if match:
-            matchingPids = (pid == match.group('pid'))
-
-    return exitCode == 0 and matchingPids
-
-
-def testLLDBOutput():
-    failedOutputWithCrashLines = [
-        '    frame #15: 0x00007fff73f4d305 libsystem_pthread.dylib`_pthread_body + 126',
-        '    frame #16: 0x00007fff73f5026f libsystem_pthread.dylib`_pthread_start + 70',
-        '    frame #17: 0x00007fff73f4c415 libsystem_pthread.dylib`thread_start + 13',
-        '(lldb) quit 1'
-    ]
-
-    failedOutputWithFailedUnittest = [
-        '===============================================================================', 
-        'test cases:  1 |  0 passed | 1 failed', 'assertions: 15 | 14 passed | 1 failed', 
-        '', 
-        'Process 279 exited with status = 1 (0x00000001) ',
-        '',
-        "Process 279 launched: '/Users/bsergeant/src/foss/ixwebsocket/test/build/Darwin/ixwebsocket_unittest' (x86_64)"
-    ]
-
-    successLines = [
-        '...',
-        '...',
-        'All tests passed (16 assertions in 1 test case)',
-        '',
-        'Process 99232 exited with status = 0 (0x00000000) ',
-        '',
-        "Process 99232 launched: '/Users/bsergeant/src/foss/ixwebsocket/test/build/Darwin/ixwebsocket_unittest' (x86_64)"
-    ]
-
-    assert not isSuccessFullRun(failedOutputWithCrashLines)
-    assert not isSuccessFullRun(failedOutputWithFailedUnittest)
-    assert isSuccessFullRun(successLines)
-
-
-def executeJob(job):
-    '''Execute a unittest and capture info about it (runtime, success, etc...)'''
-
-    start = time.time()
-
-    sys.stderr.write('.')
-
-    # 2 minutes of timeout for a single test
-    timeout = 2 * 60
-    command = Command(job['cmd'])
-    timedout, ret = command.run(timeout)
-
-    job['exit_code'] = ret
-    job['success'] = ret == 0
-    job['runtime'] = time.time() - start
-
-    # Record unittest console output
-    job['output'] = ''
-    path = job['output_path']
-
-    if os.path.exists(path):
-        with open(path) as f:
-            output = f.read()
-            job['output'] = output
-
-        outputLines = output.splitlines()
-
-        if job['use_lldb']:
-            job['success'] = isSuccessFullRun(outputLines)
-
-        # Cleanup tmp file now that its content was read
-        os.unlink(path)
-
-    return job
-
-
-def executeJobs(jobs):
-    '''Execute a list of job concurrently on multiple CPU/cores'''
-
-    poolSize = multiprocessing.cpu_count()
-
-    pool = multiprocessing.Pool(poolSize)
-    results = pool.map(executeJob, jobs)
-    pool.close()
-    pool.join()
-
-    return results
-
-
-def computeAllTestNames(buildDir):
-    '''Compute all test case names, by executing the unittest in a custom mode'''
-
-    executable = os.path.join(buildDir, DEFAULT_EXE)
-    cmd = '"{}" --list-test-names-only'.format(executable)
-    names = os.popen(cmd).read().splitlines()
-    names.sort()  # Sort test names for execution determinism
-    return names
-
-
-def prettyPrintXML(root):
-    '''Pretty print an XML file. Default writer write it on a single line
-    which makes it hard for human to inspect.'''
-
-    serializedXml = ET.tostring(root, encoding='utf-8')
-    reparsed = minidom.parseString(serializedXml)
-    prettyPrinted = reparsed.toprettyxml(indent="  ")
-    return prettyPrinted
-
-
-def generateXmlOutput(results, xmlOutput, testRunName, runTime):
-    '''Generate a junit compatible XML file
-    
-    We prefer doing this ourself instead of letting Catch2 do it.
-    When the test is crashing (as has happened on Jenkins), an invalid file 
-    with no trailer can be created which trigger an XML reading error in validateTestSuite.
-
-    Something like that:
-    ```
-    <testsuite>
-      <foo>
-    ```
-    '''
-
-    root = ET.Element('testsuites')
-    testSuite = ET.Element('testsuite', {
-        'name': testRunName,
-        'tests': str(len(results)),
-        'failures': str(sum(1 for result in results if not result['success'])),
-        'time': str(runTime),
-        'timestamp': datetime.datetime.utcnow().isoformat(),
-    })
-    root.append(testSuite)
-
-    for result in results:
-        testCase = ET.Element('testcase', {
-            'name': result['name'],
-            'time': str(result['runtime'])
-        })
-
-        systemOut = ET.Element('system-out')
-        systemOut.text = result['output'].decode('utf-8')
-        testCase.append(systemOut)
-
-        if not result['success']:
-            failure = ET.Element('failure')
-            testCase.append(failure)
-
-        testSuite.append(testCase)
-
-    with open(xmlOutput, 'w') as f:
-        content = prettyPrintXML(root)
-        f.write(content.encode('utf-8'))
-
-
-def run(testName, buildDir, sanitizer, xmlOutput, testRunName, buildOnly, useLLDB):
-    '''Main driver. Run cmake, compiles, execute and validate the testsuite.'''
-
-    runCMake(sanitizer, buildDir)
-    runCommand('make -C {} -j8'.format(buildDir))
-
-    if buildOnly:
-        return
-
-    # A specific test case can be provided on the command line
-    if testName:
-        testNames = [testName]
-    else:
-        # Default case
-        testNames = computeAllTestNames(buildDir)
-
-    # This should be empty. It is useful to have a blacklist during transitions
-    # We could add something for asan as well.
-    blackLists = {
-        'ubsan': []
-    }
-    blackList = blackLists.get(sanitizer, [])
-
-    # Run through LLDB to capture crashes
-    lldb = ''
-    if useLLDB:
-        lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
-
-    # Jobs is a list of python dicts
-    jobs = []
-
-    for testName in testNames:
-        outputPath = tempfile.mktemp(suffix=testName + '.log')
-
-        if testName in blackList:
-            log('Skipping blacklisted test {}'.format(testName), 'yellow')
-            continue
-
-        # testName can contains spaces, so we enclose them in double quotes
-        executable = os.path.join(buildDir, DEFAULT_EXE)
-
-        cmd = '{} "{}" "{}" >& "{}"'.format(lldb, executable, testName, outputPath)
-
-        jobs.append({
-            'name': testName,
-            'cmd': cmd,
-            'output_path': outputPath,
-            'use_lldb': useLLDB
-        })
-
-    start = time.time()
-    results = executeJobs(jobs)
-    runTime = time.time() - start
-    generateXmlOutput(results, xmlOutput, testRunName, runTime)
-
-    # Validate and report results
-    print('\nParsing junit test result file: {}'.format(xmlOutput))
-    log('## Results', 'blue')
-    success, tests = validateTestSuite(xmlOutput)
-
-    if success:
-        label = 'tests' if int(tests) > 1 else 'test'
-        msg = 'All test passed (#{} {})'.format(tests, label)
-        color = 'green'
-    else:
-        msg = 'unittest failed'
-        color = 'red'
-
-    log(msg, color)
-    log('Execution time: %.2fs' % (runTime), 'blue')
-    sys.exit(0 if success else 1)
-
-
-def main():
-    buildDir = 'build/' + platform.system()
-    if not os.path.exists(buildDir):
-        os.makedirs(buildDir)
-
-    defaultOutput = DEFAULT_EXE + '.xml'
-
-    parser = argparse.ArgumentParser(description='Build and Run the engine unittest')
-
-    sanitizers = ['tsan', 'asan', 'ubsan', 'none']
-
-    parser.add_argument('--sanitizer', choices=sanitizers,
-                        help='Run a clang sanitizer.')
-    parser.add_argument('--test', '-t', help='Test name.')
-    parser.add_argument('--list', '-l', action='store_true',
-                        help='Print test names and exit.')
-    parser.add_argument('--no_sanitizer', action='store_true',
-                        help='Do not execute a clang sanitizer.')
-    parser.add_argument('--validate', action='store_true',
-                        help='Validate XML output.')
-    parser.add_argument('--build_only', '-b', action='store_true',
-                        help='Stop after building. Do not run the unittest.')
-    parser.add_argument('--output', '-o', help='Output XML file.')
-    parser.add_argument('--lldb', action='store_true',
-                        help='Run the test through lldb.')
-    parser.add_argument('--run_name', '-n',
-                        help='Name of the test run.')
-
-    args = parser.parse_args()
-
-    # Default sanitizer is tsan
-    sanitizer = args.sanitizer
-    if args.sanitizer is None:
-        sanitizer = 'tsan'
-
-    defaultRunName = 'ixengine_{}_{}'.format(platform.system(), sanitizer)
-
-    xmlOutput = args.output or defaultOutput
-    testRunName = args.run_name or os.getenv('IXENGINE_TEST_RUN_NAME') or defaultRunName
-
-    if args.list:
-        # catch2 exit with a different error code when requesting the list of files
-        try:
-            runTest('--list-test-names-only', buildDir, xmlOutput, testRunName)
-        except AssertionError:
-            pass
-        return
-
-    if args.validate:
-        validateTestSuite(xmlOutput)
-        return
-
-    if platform.system() != 'Darwin' and args.lldb:
-        print('LLDB is only supported on Apple at this point')
-        args.lldb = False
-
-    return run(args.test, buildDir, sanitizer, xmlOutput, 
-               testRunName, args.build_only, args.lldb)
-
-
-if __name__ == '__main__':
-    main()
+osName = platform.system()
+print('os name = {}'.format(osName))
+
+root = os.path.dirname(os.path.realpath(__file__))
+buildDir = os.path.join(root, 'build', osName)
+
+if not os.path.exists(buildDir):
+    os.makedirs(buildDir)
+
+os.chdir(buildDir)
+
+if osName == 'Windows':
+    generator = '-G"NMake Makefiles"'
+    make = 'nmake'
+    testBinary ='ixwebsocket_unittest.exe'
+else:
+    generator = ''
+    make = 'make -j6'
+    testBinary ='./ixwebsocket_unittest'
+
+sanitizersFlags = {
+    'asan': '-DSANITIZE_ADDRESS=On',
+    'ubsan': '-DSANITIZE_UNDEFINED=On',
+    'tsan': '-DSANITIZE_THREAD=On',
+    'none': ''
+}
+sanitizer = 'tsan'
+if osName == 'Linux':
+    sanitizer = 'none'
+
+sanitizerFlags = sanitizersFlags[sanitizer]
+
+# if osName == 'Windows':
+#     os.environ['CC'] = 'clang-cl'
+#     os.environ['CXX'] = 'clang-cl'
+
+cmakeCmd = 'cmake -DCMAKE_BUILD_TYPE=Debug {} {} ../..'.format(generator, sanitizerFlags)
+print(cmakeCmd)
+ret = os.system(cmakeCmd)
+assert ret == 0, 'CMake failed, exiting'
+
+ret = os.system(make)
+assert ret == 0, 'Make failed, exiting'
+
+def findFiles(prefix):
+    '''Find all files under a given directory'''
+
+    paths = []
+
+    for root, _, files in os.walk(prefix):
+        for path in files:
+            fullPath = os.path.join(root, path)
+
+            if os.path.islink(fullPath):
+                continue
+
+            paths.append(fullPath)
+
+    return paths
+
+#for path in findFiles('.'):
+#    print(path)
+
+# We need to copy the zlib DLL in the current work directory
+shutil.copy(os.path.join(
+    '..',
+    '..',
+    '..',
+    'third_party',
+    'ZLIB-Windows',
+    'zlib-1.2.11_deploy_v140',
+    'release_dynamic',
+    'x64',
+    'bin',
+    'zlib.dll'), '.')
+
+# lldb = "lldb --batch -o 'run' -k 'thread backtrace all' -k 'quit 1'"
+lldb = ""  # Disabled for now
+testCommand = '{} {} {}'.format(lldb, testBinary, os.getenv('TEST', ''))
+command = Command(testCommand)
+timedout, ret = command.run()
+assert ret == 0, 'Test command failed'

test/test_runner.cpp

@@ -7,6 +7,8 @@
 #define CATCH_CONFIG_RUNNER
 #include "catch.hpp"
 
+#include <ixwebsocket/IXSocket.h>
+
 int main(int argc, char* argv[])
 {
     int result = Catch::Session().run(argc, argv);

third_party/statsd-client-cpp/.gitignore

@@ -1,13 +0,0 @@
-# Compiled Object files
-*.slo
-*.lo
-*.o
-
-# Compiled Dynamic libraries
-*.so
-*.dylib
-
-# Compiled Static libraries
-*.lai
-*.la
-*.a

third_party/statsd-client-cpp/CMakeLists.txt

@@ -1,18 +0,0 @@
-cmake_minimum_required(VERSION 3.1)
-project(helloCLion)
-
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
-
-include_directories(
-    src
-)
-
-add_library(statsdcppclient STATIC src/statsd_client.cpp)
-add_definitions("-fPIC")
-target_link_libraries(statsdcppclient pthread)
-
-add_executable(system_monitor demo/system_monitor.cpp)
-target_link_libraries(system_monitor statsdcppclient)
-
-add_executable(test_client demo/test_client.cpp)
-target_link_libraries(test_client statsdcppclient)

third_party/statsd-client-cpp/LICENSE

@@ -1,27 +0,0 @@
-Copyright (c) 2014, Rex
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
-* Redistributions of source code must retain the above copyright notice, this
-  list of conditions and the following disclaimer.
-
-* Redistributions in binary form must reproduce the above copyright notice,
-  this list of conditions and the following disclaimer in the documentation
-  and/or other materials provided with the distribution.
-
-* Neither the name of the {organization} nor the names of its
-  contributors may be used to endorse or promote products derived from
-  this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
-AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
-IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
-FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
-DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
-SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
-CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
-OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

third_party/statsd-client-cpp/README.md

@@ -1,34 +0,0 @@
-# a client sdk for StatsD, written in C++
-
-## API
-See [header file](src/statsd_client.h) for more api detail.
-
-** Notice: this client is not thread-safe **
-
-## Demo
-### test\_client
-This simple demo shows how the use this client.
-
-### system\_monitor
-This is a daemon for monitoring a Linux system.
-It'll wake up every minute and monitor the following:
-
-* load
-* cpu
-* free memory
-* free swap (disabled)
-* received bytes
-* transmitted bytes
-* procs
-* uptime
-
-The stats sent to statsd will be in "host.MACAddress" namespace.
-
-Usage:
-
-    system_monitor statsd-host interface-to-monitor
-
-e.g.
-
-    `system_monitor 172.16.42.1 eth0`
-

third_party/statsd-client-cpp/demo/system_monitor.cpp

@@ -1,164 +0,0 @@
-
-#include <sys/types.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <signal.h>
-#include <unistd.h>
-#include <string.h>
-#include <netdb.h>
-#include <sys/sysinfo.h>
-
-#include <sys/ioctl.h>
-#include <netinet/in.h>
-#include <net/if.h>
-#include <string>
-#include <vector>
-
-#include "statsd_client.h"
-
-using namespace std;
-
-static int running = 1;
-
-void sigterm(int sig)
-{
-    running = 0;
-}
-
-string localhost() {
-    struct addrinfo hints, *info, *p;
-    string hostname(1024, '\0');
-    gethostname((char*)hostname.data(), hostname.capacity());
-
-    memset(&hints, 0, sizeof hints);
-    hints.ai_family = AF_UNSPEC; /*either IPV4 or IPV6*/
-    hints.ai_socktype = SOCK_STREAM;
-    hints.ai_flags = AI_CANONNAME;
-
-    if ( getaddrinfo(hostname.c_str(), "http", &hints, &info) == 0) {
-        for(p = info; p != NULL; p = p->ai_next) {
-            hostname = p->ai_canonname;
-        }
-        freeaddrinfo(info);
-    }
-
-    string::size_type pos = hostname.find(".");
-    while ( pos != string::npos )
-    {
-        hostname[pos] = '_';
-        pos = hostname.find(".", pos);
-    }
-    return hostname;
-}
-
-vector<string>& StringSplitTrim(const string& sData,
-        const string& sDelim, vector<string>& vItems)
-{
-    vItems.clear();
-
-    string::size_type bpos = 0;
-    string::size_type epos = 0;
-    string::size_type nlen = sDelim.size();
-
-    while(sData.substr(epos,nlen) == sDelim)
-    {
-        epos += nlen;
-    }
-    bpos = epos;
-
-    while ((epos=sData.find(sDelim, epos)) != string::npos)
-    {
-        vItems.push_back(sData.substr(bpos, epos-bpos));
-        epos += nlen;
-        while(sData.substr(epos,nlen) == sDelim)
-        {
-            epos += nlen;
-        }
-        bpos = epos;
-    }
-
-    if(bpos != sData.size())
-    {
-        vItems.push_back(sData.substr(bpos, sData.size()-bpos));
-    }
-    return vItems;
-}
-
-int main(int argc, char *argv[])
-{
-    FILE *net, *stat;
-    struct sysinfo si;
-    char line[256];
-    unsigned int user, nice, sys, idle, total, busy, old_total=0, old_busy=0;
-
-    if (argc != 3) {
-        printf( "Usage: %s host port\n"
-                "Eg:    %s 127.0.0.1 8125\n",
-                argv[0], argv[0]);
-        exit(1);
-    }
-
-    signal(SIGHUP, SIG_IGN);
-    signal(SIGPIPE, SIG_IGN);
-    signal(SIGCHLD, SIG_IGN); /* will save one syscall per sleep */
-    signal(SIGTERM, sigterm);
-
-    if ( (net = fopen("/proc/net/dev", "r")) == NULL) {
-        perror("fopen");
-        exit(-1);
-    }
-
-    if ( (stat = fopen("/proc/stat", "r")) == NULL) {
-        perror("fopen");
-        exit(-1);
-    }
-
-    string ns = string("host.") + localhost().c_str() + ".";
-    statsd::StatsdClient client(argv[1], atoi(argv[2]), ns);
-
-    daemon(0,0);
-    printf("running in background.\n");
-
-    while(running) {
-        rewind(net);
-        vector<string> items;
-        while(!feof(net)) {
-            fgets(line, sizeof(line), net);
-            StringSplitTrim(line, " ", items);
-
-            if ( items.size() < 17 ) continue;
-            if ( items[0].find(":") == string::npos ) continue;
-            if ( items[1] == "0" and items[9] == "0" ) continue;
-
-            string netface = "network."+items[0].erase( items[0].find(":") );
-            client.count( netface+".receive.bytes", atoll(items[1].c_str()) );
-            client.count( netface+".receive.packets", atoll(items[2].c_str()) );
-            client.count( netface+".transmit.bytes", atoll(items[9].c_str()) );
-            client.count( netface+".transmit.packets", atoll(items[10].c_str()) );
-        }
-
-        sysinfo(&si);
-        client.gauge("system.load", 100*si.loads[0]/0x10000);
-        client.gauge("system.freemem", si.freeram/1024);
-        client.gauge("system.procs", si.procs);
-        client.count("system.uptime", si.uptime);
-
-        /* rewind doesn't do the trick for /proc/stat */
-        freopen("/proc/stat", "r", stat);
-        fgets(line, sizeof(line), stat);
-        sscanf(line, "cpu  %u %u %u %u", &user, &nice, &sys, &idle);
-        total = user + sys + idle;
-        busy = user + sys;
-
-        client.send("system.cpu", 100 * (busy - old_busy)/(total - old_total), "g", 1.0);
-
-        old_total = total;
-        old_busy = busy;
-        sleep(6);
-    }
-
-    fclose(net);
-    fclose(stat);
-
-    exit(0);
-}

third_party/statsd-client-cpp/demo/test_client.cpp

@@ -1,28 +0,0 @@
-
-#include <iostream>
-#include <unistd.h>
-#include "statsd_client.h"
-
-int main(void)
-{
-    std::cout << "running..." << std::endl;
-
-    statsd::StatsdClient client;
-    statsd::StatsdClient client2("127.0.0.1", 8125, "myproject.abx.", true);
-
-    client.count("count1", 123, 1.0);
-    client.count("count2", 125, 1.0);
-    client.gauge("speed", 10);
-    int i;
-    for (i=0; i<1000; i++)
-        client2.timing("request", i);
-    sleep(1);
-    client.inc("count1", 1.0);
-    client2.dec("count2", 1.0);
-//    for(i=0; i<1000; i++) {
-//        client2.count("count3", i, 0.8);
-//    }
-
-    std::cout << "done" << std::endl;
-    return 0;
-}

third_party/statsd-client-cpp/src/statsd_client.cpp

@@ -1,246 +0,0 @@
-#include <math.h>
-#include <netdb.h>
-#include <time.h>
-#include <unistd.h>
-#include <stdlib.h>
-#include <string.h>
-#include <stdio.h>
-#include <netinet/in.h>
-#include "statsd_client.h"
-
-using namespace std;
-namespace statsd {
-
-inline bool fequal(float a, float b)
-{
-    const float epsilon = 0.0001;
-    return ( fabs(a - b) < epsilon );
-}
-
-inline bool should_send(float sample_rate)
-{
-    if ( fequal(sample_rate, 1.0) )
-    {
-        return true;
-    }
-
-    float p = ((float)random() / RAND_MAX);
-    return sample_rate > p;
-}
-
-struct _StatsdClientData {
-    int     sock;
-    struct  sockaddr_in server;
-
-    string  ns;
-    string  host;
-    short   port;
-    bool    init;
-
-    char    errmsg[1024];
-};
-
-StatsdClient::StatsdClient(const string& host,
-                           int port,
-                           const string& ns,
-                           const bool batching)
-: batching_(batching), exit_(false)
-{
-    d = new _StatsdClientData;
-    d->sock = -1;
-    config(host, port, ns);
-    srandom(time(NULL));
-
-    if (batching_) {
-        pthread_mutex_init(&batching_mutex_lock_, nullptr);
-        batching_thread_ = std::thread([this] {
-          while (!exit_) {
-              std::deque<std::string> staged_message_queue;
-
-              pthread_mutex_lock(&batching_mutex_lock_);
-              batching_message_queue_.swap(staged_message_queue);
-              pthread_mutex_unlock(&batching_mutex_lock_);
-
-              while(!staged_message_queue.empty()) {
-                  send_to_daemon(staged_message_queue.front());
-                  staged_message_queue.pop_front();
-              }
-
-              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
-          }
-        });
-    }
-}
-
-StatsdClient::~StatsdClient()
-{
-    if (batching_) {
-        exit_ = true;
-        batching_thread_.join();
-        pthread_mutex_destroy(&batching_mutex_lock_);
-    }
-
-
-    // close socket
-    if (d->sock >= 0) {
-        close(d->sock);
-        d->sock = -1;
-        delete d;
-        d = NULL;
-    }
-}
-
-void StatsdClient::config(const string& host, int port, const string& ns)
-{
-    d->ns = ns;
-    d->host = host;
-    d->port = port;
-    d->init = false;
-    if ( d->sock >= 0 ) {
-        close(d->sock);
-    }
-    d->sock = -1;
-}
-
-int StatsdClient::init()
-{
-    if ( d->init ) return 0;
-
-    d->sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
-    if ( d->sock == -1 ) {
-        snprintf(d->errmsg, sizeof(d->errmsg), "could not create socket, err=%m");
-        return -1;
-    }
-
-    memset(&d->server, 0, sizeof(d->server));
-    d->server.sin_family = AF_INET;
-    d->server.sin_port = htons(d->port);
-
-    int ret = inet_aton(d->host.c_str(), &d->server.sin_addr);
-    if ( ret == 0 )
-    {
-        // host must be a domain, get it from internet
-        struct addrinfo hints, *result = NULL;
-        memset(&hints, 0, sizeof(hints));
-        hints.ai_family = AF_INET;
-        hints.ai_socktype = SOCK_DGRAM;
-
-        ret = getaddrinfo(d->host.c_str(), NULL, &hints, &result);
-        if ( ret ) {
-            close(d->sock);
-            d->sock = -1;
-            snprintf(d->errmsg, sizeof(d->errmsg),
-                     "getaddrinfo fail, error=%d, msg=%s", ret, gai_strerror(ret) );
-            return -2;
-        }
-        struct sockaddr_in* host_addr = (struct sockaddr_in*)result->ai_addr;
-        memcpy(&d->server.sin_addr, &host_addr->sin_addr, sizeof(struct in_addr));
-        freeaddrinfo(result);
-    }
-
-    d->init = true;
-    return 0;
-}
-
-/* will change the original string */
-void StatsdClient::cleanup(string& key)
-{
-    size_t pos = key.find_first_of(":|@");
-    while ( pos != string::npos )
-    {
-        key[pos] = '_';
-        pos = key.find_first_of(":|@");
-    }
-}
-
-int StatsdClient::dec(const string& key, float sample_rate)
-{
-    return count(key, -1, sample_rate);
-}
-
-int StatsdClient::inc(const string& key, float sample_rate)
-{
-    return count(key, 1, sample_rate);
-}
-
-int StatsdClient::count(const string& key, size_t value, float sample_rate)
-{
-    return send(key, value, "c", sample_rate);
-}
-
-int StatsdClient::gauge(const string& key, size_t value, float sample_rate)
-{
-    return send(key, value, "g", sample_rate);
-}
-
-int StatsdClient::timing(const string& key, size_t ms, float sample_rate)
-{
-    return send(key, ms, "ms", sample_rate);
-}
-
-int StatsdClient::send(string key, size_t value, const string &type, float sample_rate)
-{
-    if (!should_send(sample_rate)) {
-        return 0;
-    }
-
-    cleanup(key);
-
-    char buf[256];
-    if ( fequal( sample_rate, 1.0 ) )
-    {
-        snprintf(buf, sizeof(buf), "%s%s:%zd|%s",
-                 d->ns.c_str(), key.c_str(), value, type.c_str());
-    }
-    else
-    {
-        snprintf(buf, sizeof(buf), "%s%s:%zd|%s|@%.2f",
-                 d->ns.c_str(), key.c_str(), value, type.c_str(), sample_rate);
-    }
-
-    return send(buf);
-}
-
-int StatsdClient::send(const string &message)
-{
-    if (batching_) {
-        pthread_mutex_lock(&batching_mutex_lock_);
-        if (batching_message_queue_.empty() ||
-            batching_message_queue_.back().length() > max_batching_size) {
-            batching_message_queue_.push_back(message);
-        } else {
-            (*batching_message_queue_.rbegin()).append("\n").append(message);
-        }
-        pthread_mutex_unlock(&batching_mutex_lock_);
-
-        return 0;
-    } else {
-        return send_to_daemon(message);
-    }
-}
-
-
-int StatsdClient::send_to_daemon(const string &message) {
-    std::cout << "send_to_daemon: " << message.length() << " B" << std::endl;
-    int ret = init();
-    if ( ret )
-    {
-        return ret;
-    }
-    ret = sendto(d->sock, message.data(), message.size(), 0, (struct sockaddr *) &d->server, sizeof(d->server));
-    if ( ret == -1) {
-        snprintf(d->errmsg, sizeof(d->errmsg),
-                 "sendto server fail, host=%s:%d, err=%m", d->host.c_str(), d->port);
-        return -1;
-    }
-
-    return 0;
-}
-
-const char* StatsdClient::errmsg()
-{
-    return d->errmsg;
-}
-
-}
-

third_party/statsd-client-cpp/src/statsd_client.h

@@ -1,66 +0,0 @@
-
-#ifndef STATSD_CLIENT_H
-#define STATSD_CLIENT_H
-
-#include <sys/types.h>
-#include <arpa/inet.h>
-#include <sys/socket.h>
-#include <pthread.h>
-#include <string>
-#include <thread>
-#include <deque>
-#include <iostream>
-
-namespace statsd {
-
-struct _StatsdClientData;
-
-class StatsdClient {
-public:
-    StatsdClient(const std::string& host="127.0.0.1", int port=8125, const std::string& ns = "", const bool batching = false);
-    ~StatsdClient();
-
-public:
-    // you can config at anytime; client will use new address (useful for Singleton)
-    void config(const std::string& host, int port, const std::string& ns = "");
-    const char* errmsg();
-    int send_to_daemon(const std::string &);
-
-public:
-    int inc(const std::string& key, float sample_rate = 1.0);
-    int dec(const std::string& key, float sample_rate = 1.0);
-    int count(const std::string& key, size_t value, float sample_rate = 1.0);
-    int gauge(const std::string& key, size_t value, float sample_rate = 1.0);
-    int timing(const std::string& key, size_t ms, float sample_rate = 1.0);
-
-public:
-    /**
-     * (Low Level Api) manually send a message
-     * which might be composed of several lines.
-     */
-    int send(const std::string& message);
-
-    /* (Low Level Api) manually send a message
-     * type = "c", "g" or "ms"
-     */
-    int send(std::string key, size_t value,
-             const std::string& type, float sample_rate);
-
-protected:
-    int init();
-    void cleanup(std::string& key);
-
-protected:
-    struct _StatsdClientData* d;
-
-    bool batching_;
-    bool exit_;
-    pthread_mutex_t batching_mutex_lock_;
-    std::thread batching_thread_;
-    std::deque<std::string> batching_message_queue_;
-    const uint64_t max_batching_size = 32768;
-};
-
-}; // end namespace
-
-#endif

ws/CMakeLists.txt

@@ -7,12 +7,7 @@ cmake_minimum_required (VERSION 3.4.1)
 project (ws)
 
 # There's -Weverything too for clang
-if (NOT WIN32)
-  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
-endif()
-
-#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
-#set(CMAKE_LD_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=thread")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic")
 
 set (CMAKE_CXX_STANDARD 14)
 
@@ -21,20 +16,14 @@ option(USE_TLS "Add TLS support" ON)
 include_directories(ws .)
 include_directories(ws ..)
 include_directories(ws ../third_party)
-include_directories(ws ../third_party/statsd-client-cpp/src)
 
 add_executable(ws 
   ../third_party/msgpack11/msgpack11.cpp
-  ../third_party/jsoncpp/jsoncpp.cpp
-  ../third_party/statsd-client-cpp/src/statsd_client.cpp
   ixcrypto/IXBase64.cpp
   ixcrypto/IXHash.cpp
   ixcrypto/IXUuid.cpp
-  ixcrypto/IXHMac.cpp
 
   IXRedisClient.cpp
-  IXSentryClient.cpp
-  IXCobraConnection.cpp
 
   ws_http_client.cpp
   ws_ping_pong.cpp
@@ -47,19 +36,11 @@ add_executable(ws
   ws_receive.cpp
   ws_redis_publish.cpp
   ws_redis_subscribe.cpp
-  ws_cobra_subscribe.cpp
-  ws_cobra_to_statsd.cpp
-  ws_cobra_to_sentry.cpp
   ws.cpp)
 
-target_link_libraries(ws ixwebsocket)
-
-if(NOT APPLE)
-  find_package(OpenSSL REQUIRED)
-  add_definitions(${OPENSSL_DEFINITIONS})
-  message(STATUS "OpenSSL: " ${OPENSSL_VERSION})
-  include_directories(${OPENSSL_INCLUDE_DIR})
-  target_link_libraries(ws ${OPENSSL_LIBRARIES})
+if (APPLE AND USE_TLS)
+    target_link_libraries(ws "-framework foundation" "-framework security")
 endif()
 
+target_link_libraries(ws ixwebsocket cpp_redis tacopie)
 install(TARGETS ws RUNTIME DESTINATION bin)

ws/IXRedisClient.cpp

@@ -7,6 +7,7 @@
 #include "IXRedisClient.h"
 #include <ixwebsocket/IXSocketFactory.h>
 #include <ixwebsocket/IXSocket.h>
+#include <cpp_redis/cpp_redis>
 
 #include <sstream>
 #include <iomanip>
@@ -17,6 +18,14 @@ namespace ix
 {
     bool RedisClient::connect(const std::string& hostname, int port)
     {
+         _sub.connect(hostname, port, []
+                 (const std::string& host, std::size_t port, cpp_redis::connect_state status) {
+            if (status == cpp_redis::connect_state::dropped) {
+              std::cout << "client disconnected from " << host << ":" << port << std::endl;
+            }
+          });
+
+         // also subscribe the old way
         bool tls = false;
         std::string errorMsg;
         _socket = createSocket(tls, errorMsg);
@@ -28,11 +37,22 @@ namespace ix
 
         std::string errMsg;
         return _socket->connect(hostname, port, errMsg, nullptr);
+        
     }
 
     bool RedisClient::auth(const std::string& password,
                            std::string& response)
     {
+        // authentication if server-server requires it
+        //  _sub.auth(password, [&response](const cpp_redis::reply& reply) {
+        //    if (reply.is_error()) { std::cerr << "Authentication failed: " << reply.as_string() << std::endl; }
+        //    else {
+        //      std::cout << "successful authentication" << std::endl;
+        //    }
+        //  });
+
+        return true;
+#if 0
         response.clear();
 
         if (!_socket) return false;
@@ -60,6 +80,7 @@ namespace ix
 
         response = line;
         return lineValid;
+#endif
     }
 
 
@@ -101,6 +122,19 @@ namespace ix
                                 const OnRedisSubscribeResponseCallback& responseCallback,
                                 const OnRedisSubscribeCallback& callback)
     {
+        _sub.subscribe(channel, [&callback](const std::string& chan, const std::string& msg) {
+            callback(msg);
+        });
+        _sub.commit();
+
+        while (true)
+        {
+            auto duration = std::chrono::seconds(1);
+            std::this_thread::sleep_for(duration);
+        }
+
+        return true;
+#if 0
         if (!_socket) return false;
 
         std::stringstream ss;
@@ -203,5 +237,6 @@ namespace ix
         }
 
         return true;
+#endif
     }
 }

ws/IXRedisClient.h

@@ -8,6 +8,7 @@
 
 #include <memory>
 #include <functional>
+#include <cpp_redis/cpp_redis>
 
 namespace ix
 {
@@ -35,6 +36,8 @@ namespace ix
                        const OnRedisSubscribeCallback& callback);
 
     private:
+        cpp_redis::subscriber _sub;
+
         std::shared_ptr<Socket> _socket;
     };
 }

ws/IXSentryClient.cpp

@@ -1,178 +0,0 @@
-/*
- *  IXSentryClient.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone. All rights reserved.
- */
-
-#include "IXSentryClient.h"
-
-#include <chrono>
-#include <iostream>
-
-#include <ixwebsocket/IXWebSocketHttpHeaders.h>
-
-
-namespace ix
-{
-    SentryClient::SentryClient(const std::string& dsn) :
-        _dsn(dsn),
-        _validDsn(false),
-        _luaFrameRegex("\t([^/]+):([0-9]+): in function '([^/]+)'")
-    {
-        const std::regex dsnRegex("(http[s]?)://([^:]+):([^@]+)@([^/]+)/([0-9]+)");
-        std::smatch group;
-
-        if (std::regex_match(dsn, group, dsnRegex) and group.size() == 6)
-        {
-            _validDsn = true;
-
-            const auto scheme = group.str(1);
-            const auto host = group.str(4);
-            const auto project_id = group.str(5);
-            _url = scheme + "://" + host + "/api/" + project_id + "/store/";
-
-            _publicKey = group.str(2);
-            _secretKey = group.str(3);
-        }
-    }
-
-    int64_t SentryClient::getTimestamp()
-    {
-        const auto tp = std::chrono::system_clock::now();
-        const auto dur = tp.time_since_epoch();
-        return std::chrono::duration_cast<std::chrono::seconds>(dur).count();
-    }
-
-    std::string SentryClient::getIso8601()
-    {
-        std::time_t now;
-        std::time(&now);
-        char buf[sizeof "2011-10-08T07:07:09Z"];
-        std::strftime(buf, sizeof buf, "%Y-%m-%dT%H:%M:%SZ", std::gmtime(&now));
-        return buf;
-    }
-
-    std::string SentryClient::computeAuthHeader()
-    {
-        std::string securityHeader("Sentry sentry_version=5");
-        securityHeader += ",sentry_client=ws/1.0.0";
-        securityHeader += ",sentry_timestamp=" + std::to_string(SentryClient::getTimestamp());
-        securityHeader += ",sentry_key=" + _publicKey;
-        securityHeader += ",sentry_secret=" + _secretKey;
-
-        return securityHeader;
-    }
-
-    Json::Value SentryClient::parseLuaStackTrace(const std::string& stack)
-    {
-        Json::Value frames;
-
-        // Split by lines
-        std::string line;
-        std::stringstream tokenStream(stack);
-
-        std::stringstream ss;
-        std::smatch group;
-
-        while (std::getline(tokenStream, line))
-        {
-            //	MapScene.lua:2169: in function 'singleCB'
-            if (std::regex_match(line, group, _luaFrameRegex))
-            {
-                const auto fileName = group.str(1);
-                const auto linenoStr = group.str(2);
-                const auto function = group.str(3);
-
-                ss << linenoStr;
-                uint64_t lineno;
-                ss >> lineno;
-
-                Json::Value frame;
-                frame["lineno"] = lineno;
-                frame["filename"] = fileName;
-                frame["function"] = function;
-
-                frames.append(frame);
-            }
-        }
-
-        return frames;
-    }
-
-    std::string SentryClient::computePayload(const Json::Value& msg)
-    {
-        Json::Value payload;
-        payload["platform"] = "python";
-        payload["sdk"]["name"] = "ws";
-        payload["sdk"]["version"] = "1.0.0";
-        payload["timestamp"] = SentryClient::getIso8601();
-
-        Json::Value exception;
-        exception["value"] = msg["data"]["message"];
-
-        std::string stackTraceFieldName = 
-            (msg["id"].asString() == "game_noisytypes_id") ? "traceback" : "stack";
-                             
-        exception["stacktrace"]["frames"] = 
-            parseLuaStackTrace(msg["data"][stackTraceFieldName].asString());
-
-        payload["exception"].append(exception);
-
-        Json::Value extra;
-        extra["cobra_event"] = msg;
-
-        exception["extra"] = extra;
-
-        return _jsonWriter.write(payload);
-    }
-
-    bool SentryClient::send(const Json::Value& msg,
-                            bool verbose)
-    {
-        HttpRequestArgs args;
-        args.extraHeaders["X-Sentry-Auth"] = SentryClient::computeAuthHeader();
-        args.connectTimeout = 60;
-        args.transferTimeout = 5 * 60;
-        args.followRedirects = true;
-        args.verbose = verbose;
-        args.logger = [](const std::string& msg)
-        {
-            std::cout << msg;
-        };
-
-        std::string body = computePayload(msg);
-        HttpResponse out = _httpClient.post(_url, body, args);
-
-        auto statusCode = std::get<0>(out);
-        auto errorCode = std::get<1>(out);
-        auto responseHeaders = std::get<2>(out);
-        auto payload = std::get<3>(out);
-        auto errorMsg = std::get<4>(out);
-        auto uploadSize = std::get<5>(out);
-        auto downloadSize = std::get<6>(out);
-
-        if (verbose)
-        {
-            for (auto it : responseHeaders)
-            {
-                std::cerr << it.first << ": " << it.second << std::endl;
-            }
-
-            std::cerr << "Upload size: " << uploadSize << std::endl;
-            std::cerr << "Download size: " << downloadSize << std::endl;
-
-            std::cerr << "Status: " << statusCode << std::endl;
-            if (errorCode != HttpErrorCode_Ok)
-            {
-                std::cerr << "error message: " << errorMsg << std::endl;
-            }
-
-            if (responseHeaders["Content-Type"] != "application/octet-stream")
-            {
-                std::cerr << "payload: " << payload << std::endl;
-            }
-        }
-
-        return statusCode == 200;
-    }
-} // namespace ix

ws/IXSentryClient.h

@@ -1,47 +0,0 @@
-/*
- *  IXSentryClient.h
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone. All rights reserved.
- */
-
-#pragma once
-
-#include <jsoncpp/json/json.h>
-#include <regex>
-
-#include <ixwebsocket/IXHttpClient.h>
-
-namespace ix
-{
-    class SentryClient
-    {
-    public:
-        SentryClient(const std::string& dsn);
-        ~SentryClient() = default;
-
-        bool send(const Json::Value& msg, bool verbose);
-
-    private:
-        int64_t getTimestamp();
-        std::string computeAuthHeader();
-        std::string getIso8601();
-        std::string computePayload(const Json::Value& msg);
-
-        Json::Value parseLuaStackTrace(const std::string& stack);
-
-        std::string _dsn;
-        bool _validDsn;
-        std::string _url;
-
-        // Used for authentication with a header
-        std::string _publicKey;
-        std::string _secretKey;
-
-        Json::FastWriter _jsonWriter;
-
-        std::regex _luaFrameRegex;
-
-        HttpClient _httpClient;
-    };
-    
-} // namespace ix

ws/cobra_publisher/.gitignore

@@ -0,0 +1,3 @@
+venv
+build
+node_modules

ws/cobra_publisher/CMakeLists.txt

@@ -0,0 +1,38 @@
+#
+# Author: Benjamin Sergeant
+# Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+#
+
+cmake_minimum_required (VERSION 3.4.1)
+project (cobra_publisher)
+
+# There's -Weverything too for clang
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra -pedantic -Wshorten-64-to-32")
+
+set (OPENSSL_PREFIX /usr/local/opt/openssl) # Homebrew openssl
+
+set (CMAKE_CXX_STANDARD 14)
+
+option(USE_TLS "Add TLS support" ON)
+
+include_directories(cobra_publisher ${OPENSSL_PREFIX}/include)
+include_directories(cobra_publisher .)
+
+add_executable(cobra_publisher 
+  jsoncpp/jsoncpp.cpp
+  ixcrypto/IXHMac.cpp
+  ixcrypto/IXBase64.cpp
+  IXCobraConnection.cpp
+  cobra_publisher.cpp)
+
+if (APPLE AND USE_TLS)
+    target_link_libraries(cobra_publisher "-framework foundation" "-framework security")
+endif()
+
+get_filename_component(crypto_lib_path ${OPENSSL_PREFIX}/lib/libcrypto.a ABSOLUTE)
+add_library(lib_crypto STATIC IMPORTED)
+set_target_properties(lib_crypto PROPERTIES IMPORTED_LOCATION ${crypto_lib_path})
+
+link_directories(/usr/local/opt/openssl/lib)
+target_link_libraries(cobra_publisher ixwebsocket lib_crypto)
+install(TARGETS cobra_publisher DESTINATION bin)

ws/cobra_publisher/IXCobraConnection.cpp

@@ -6,7 +6,6 @@
 
 #include "IXCobraConnection.h"
 #include <ixcrypto/IXHMac.h>
-#include <ixwebsocket/IXWebSocket.h>
 
 #include <algorithm>
 #include <stdexcept>
@@ -21,10 +20,9 @@ namespace ix
     constexpr size_t CobraConnection::kQueueMaxSize;
 
     CobraConnection::CobraConnection() :
-        _webSocket(new WebSocket()),
-        _publishMode(CobraConnection_PublishMode_Immediate),
         _authenticated(false),
-        _eventCallback(nullptr)
+        _eventCallback(nullptr),
+        _publishMode(CobraConnection_PublishMode_Immediate)
     {
         _pdu["action"] = "rtm/publish";
 
@@ -34,7 +32,6 @@ namespace ix
     CobraConnection::~CobraConnection()
     {
         disconnect();
-        setEventCallback(nullptr);
     }
 
     void CobraConnection::setTrafficTrackerCallback(const TrafficTrackerCallback& callback)
@@ -62,40 +59,36 @@ namespace ix
     }
 
     void CobraConnection::invokeEventCallback(ix::CobraConnectionEventType eventType,
-                                               const std::string& errorMsg,
-                                               const WebSocketHttpHeaders& headers,
-                                               const std::string& subscriptionId)
+                                              const std::string& errorMsg,
+                                              const WebSocketHttpHeaders& headers)
     {
         std::lock_guard<std::mutex> lock(_eventCallbackMutex);
         if (_eventCallback)
         {
-            _eventCallback(eventType, errorMsg, headers, subscriptionId);
+            _eventCallback(eventType, errorMsg, headers);
         }
     }
 
-    void CobraConnection::invokeErrorCallback(const std::string& errorMsg,
-                                               const std::string& serializedPdu)
+    void CobraConnection::invokeErrorCallback(const std::string& errorMsg)
     {
-        std::stringstream ss;
-        ss << errorMsg << " : received pdu => " << serializedPdu;
-        invokeEventCallback(ix::CobraConnection_EventType_Error, ss.str());
+        invokeEventCallback(ix::CobraConnection_EventType_Error, errorMsg);
     }
 
     void CobraConnection::disconnect()
     {
         _authenticated = false;
-        _webSocket->stop();
+        _webSocket.stop();
     }
 
     void CobraConnection::initWebSocketOnMessageCallback()
     {
-        _webSocket->setOnMessageCallback(
+        _webSocket.setOnMessageCallback(
             [this](ix::WebSocketMessageType messageType,
                    const std::string& str,
                    size_t wireSize,
                    const ix::WebSocketErrorInfo& error,
-                   const ix::WebSocketOpenInfo& openInfo,
-                   const ix::WebSocketCloseInfo& closeInfo)
+                   const ix::WebSocketCloseInfo& closeInfo,
+                   const ix::WebSocketHttpHeaders& headers)
             {
                 CobraConnection::invokeTrafficTrackerCallback(wireSize, true);
 
@@ -104,7 +97,7 @@ namespace ix
                 {
                     invokeEventCallback(ix::CobraConnection_EventType_Open,
                                         std::string(),
-                                        openInfo.headers);
+                                        headers);
                     sendHandshakeMessage();
                 }
                 else if (messageType == ix::WebSocket_MessageType_Close)
@@ -123,13 +116,13 @@ namespace ix
                     Json::Reader reader;
                     if (!reader.parse(str, data))
                     {
-                        invokeErrorCallback("Invalid json", str);
+                        invokeErrorCallback(std::string("Invalid json: ") + str);
                         return;
                     }
 
                     if (!data.isMember("action"))
                     {
-                        invokeErrorCallback("Missing action", str);
+                        invokeErrorCallback("Missing action");
                         return;
                     }
 
@@ -139,12 +132,12 @@ namespace ix
                     {
                         if (!handleHandshakeResponse(data))
                         {
-                            invokeErrorCallback("Error extracting nonce from handshake response", str);
+                            invokeErrorCallback("Error extracting nonce from handshake response");
                         }
                     }
                     else if (action == "auth/handshake/error")
                     {
-                        invokeErrorCallback("Handshake error", str);
+                        invokeErrorCallback("Handshake error."); // print full message ?
                     }
                     else if (action == "auth/authenticate/ok")
                     {
@@ -154,37 +147,15 @@ namespace ix
                     }
                     else if (action == "auth/authenticate/error")
                     {
-                        invokeErrorCallback("Authentication error", str);
+                        invokeErrorCallback("Authentication error."); // print full message ?
                     }
                     else if (action == "rtm/subscription/data")
                     {
                         handleSubscriptionData(data);
                     }
-                    else if (action == "rtm/subscribe/ok")
-                    {
-                        if (!handleSubscriptionResponse(data))
-                        {
-                            invokeErrorCallback("Error processing subscribe response", str);
-                        }
-                    }
-                    else if (action == "rtm/subscribe/error")
-                    {
-                        invokeErrorCallback("Subscription error", str);
-                    }
-                    else if (action == "rtm/unsubscribe/ok")
-                    {
-                        if (!handleUnsubscriptionResponse(data))
-                        {
-                            invokeErrorCallback("Error processing subscribe response", str);
-                        }
-                    }
-                    else if (action == "rtm/unsubscribe/error")
-                    {
-                        invokeErrorCallback("Unsubscription error", str);
-                    }
                     else
                     {
-                        invokeErrorCallback("Un-handled message type", str);
+                        invokeErrorCallback(std::string("Un-handled message type: ") + action);
                     }
                 }
                 else if (messageType == ix::WebSocket_MessageType_Error)
@@ -194,7 +165,7 @@ namespace ix
                     ss << "#retries: "         << error.retries     << std::endl;
                     ss << "Wait time(ms): "    << error.wait_time   << std::endl;
                     ss << "HTTP Status: "      << error.http_status << std::endl;
-                    invokeErrorCallback(ss.str(), std::string());
+                    invokeErrorCallback(ss.str());
                 }
         });
     }
@@ -205,10 +176,10 @@ namespace ix
     }
 
     void CobraConnection::configure(const std::string& appkey,
-                                     const std::string& endpoint,
-                                     const std::string& rolename,
-                                     const std::string& rolesecret,
-                                     WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
+                                    const std::string& endpoint,
+                                    const std::string& rolename,
+                                    const std::string& rolesecret,
+                                    WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions)
     {
         _appkey = appkey;
         _endpoint = endpoint;
@@ -221,8 +192,8 @@ namespace ix
         ss << _appkey;
 
         std::string url = ss.str();
-        _webSocket->setUrl(url);
-        _webSocket->setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
+        _webSocket.setUrl(url);
+        _webSocket.setPerMessageDeflateOptions(webSocketPerMessageDeflateOptions);
     }
 
     //
@@ -255,10 +226,10 @@ namespace ix
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
 
-        return _webSocket->send(serializedJson).success;
+        return _webSocket.send(serializedJson).success;
     }
 
-    // 
+    //
     // Extract the nonce from the handshake response
     // use it to compute a hash during authentication
     //
@@ -317,47 +288,16 @@ namespace ix
         std::string serializedJson = serializeJson(pdu);
         CobraConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
 
-        return _webSocket->send(serializedJson).success;
+        return _webSocket.send(serializedJson).success;
     }
 
-    bool CobraConnection::handleSubscriptionResponse(const Json::Value& pdu)
-    {
-        if (!pdu.isMember("body")) return false;
-        Json::Value body = pdu["body"];
-
-        if (!body.isMember("subscription_id")) return false;
-        Json::Value subscriptionId = body["subscription_id"];
-
-        if (!subscriptionId.isString()) return false;
-
-        invokeEventCallback(ix::CobraConnection_EventType_Subscribed,
-                            std::string(), WebSocketHttpHeaders(),
-                            subscriptionId.asString());
-        return true;
-    }
-
-    bool CobraConnection::handleUnsubscriptionResponse(const Json::Value& pdu)
-    {
-        if (!pdu.isMember("body")) return false;
-        Json::Value body = pdu["body"];
-
-        if (!body.isMember("subscription_id")) return false;
-        Json::Value subscriptionId = body["subscription_id"];
-
-        if (!subscriptionId.isString()) return false;
-
-        invokeEventCallback(ix::CobraConnection_EventType_UnSubscribed,
-                            std::string(), WebSocketHttpHeaders(),
-                            subscriptionId.asString());
-        return true;
-    }
 
     bool CobraConnection::handleSubscriptionData(const Json::Value& pdu)
     {
         if (!pdu.isMember("body")) return false;
         Json::Value body = pdu["body"];
 
-        // Identify subscription_id, so that we can find 
+        // Identify subscription_id, so that we can find
         // which callback to execute
         if (!body.isMember("subscription_id")) return false;
         Json::Value subscriptionId = body["subscription_id"];
@@ -380,13 +320,13 @@ namespace ix
 
     bool CobraConnection::connect()
     {
-        _webSocket->start();
+        _webSocket.start();
         return true;
     }
 
     bool CobraConnection::isConnected() const
     {
-        return _webSocket->getReadyState() == ix::WebSocket_ReadyState_Open;
+        return _webSocket.getReadyState() == ix::WebSocket_ReadyState_Open;
     }
 
     std::string CobraConnection::serializeJson(const Json::Value& value)
@@ -399,7 +339,7 @@ namespace ix
     // publish is not thread safe as we are trying to reuse some Json objects.
     //
     bool CobraConnection::publish(const Json::Value& channels,
-                                   const Json::Value& msg)
+                                  const Json::Value& msg)
     {
         _body["channels"] = channels;
         _body["message"] = msg;
@@ -431,7 +371,7 @@ namespace ix
     }
 
     void CobraConnection::subscribe(const std::string& channel,
-                                     SubscriptionCallback cb)
+                                    SubscriptionCallback cb)
     {
         // Create and send a subscribe pdu
         Json::Value body;
@@ -441,7 +381,7 @@ namespace ix
         pdu["action"] = "rtm/subscribe";
         pdu["body"] = body;
 
-        _webSocket->send(pdu.toStyledString());
+        _webSocket.send(pdu.toStyledString());
 
         // Set the callback
         std::lock_guard<std::mutex> lock(_cbsMutex);
@@ -460,13 +400,13 @@ namespace ix
 
         // Create and send an unsubscribe pdu
         Json::Value body;
-        body["subscription_id"] = channel;
+        body["channel"] = channel;
 
         Json::Value pdu;
         pdu["action"] = "rtm/unsubscribe";
         pdu["body"] = body;
 
-        _webSocket->send(pdu.toStyledString());
+        _webSocket.send(pdu.toStyledString());
     }
 
     //
@@ -516,7 +456,7 @@ namespace ix
 
     bool CobraConnection::publishMessage(const std::string& serializedJson)
     {
-        auto webSocketSendInfo = _webSocket->send(serializedJson);
+        auto webSocketSendInfo = _webSocket.send(serializedJson);
         CobraConnection::invokeTrafficTrackerCallback(webSocketSendInfo.wireSize,
                                                        false);
         return webSocketSendInfo.success;
@@ -531,5 +471,5 @@ namespace ix
     {
         connect();
     }
-    
+
 } // namespace ix

ws/cobra_publisher/IXCobraConnection.h

@@ -11,24 +11,19 @@
 #include <string>
 #include <thread>
 #include <unordered_map>
-#include <memory>
 
 #include <jsoncpp/json/json.h>
-#include <ixwebsocket/IXWebSocketHttpHeaders.h>
+#include <ixwebsocket/IXWebSocket.h>
 #include <ixwebsocket/IXWebSocketPerMessageDeflateOptions.h>
 
 namespace ix
 {
-    class WebSocket;
-
     enum CobraConnectionEventType
     {
         CobraConnection_EventType_Authenticated = 0,
         CobraConnection_EventType_Error = 1,
         CobraConnection_EventType_Open = 2,
-        CobraConnection_EventType_Closed = 3,
-        CobraConnection_EventType_Subscribed = 4,
-        CobraConnection_EventType_UnSubscribed = 5
+        CobraConnection_EventType_Closed = 3
     };
 
     enum CobraConnectionPublishMode
@@ -40,8 +35,7 @@ namespace ix
     using SubscriptionCallback = std::function<void(const Json::Value&)>;
     using EventCallback = std::function<void(CobraConnectionEventType,
                                              const std::string&,
-                                             const WebSocketHttpHeaders&,
-                                             const std::string&)>;
+                                             const WebSocketHttpHeaders&)>;
     using TrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
 
     class CobraConnection
@@ -90,7 +84,7 @@ namespace ix
 
         /// Returns true only if we're connected
         bool isConnected() const;
-        
+
         /// Flush the publish queue
         bool flushQueue();
 
@@ -106,8 +100,6 @@ namespace ix
         bool handleHandshakeResponse(const Json::Value& data);
         bool sendAuthMessage(const std::string& nonce);
         bool handleSubscriptionData(const Json::Value& pdu);
-        bool handleSubscriptionResponse(const Json::Value& pdu);
-        bool handleUnsubscriptionResponse(const Json::Value& pdu);
 
         void initWebSocketOnMessageCallback();
 
@@ -121,15 +113,13 @@ namespace ix
         /// Invoke event callbacks
         void invokeEventCallback(CobraConnectionEventType eventType,
                                  const std::string& errorMsg = std::string(),
-                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders(),
-                                 const std::string& subscriptionId = std::string());
-        void invokeErrorCallback(const std::string& errorMsg,
-                                 const std::string& serializedPdu);
+                                 const WebSocketHttpHeaders& headers = WebSocketHttpHeaders());
+        void invokeErrorCallback(const std::string& errorMsg);
 
         ///
         /// Member variables
-        /// 
-        std::unique_ptr<WebSocket> _webSocket;
+        ///
+        WebSocket _webSocket;
 
         /// Configuration data
         std::string _appkey;
@@ -158,10 +148,10 @@ namespace ix
         std::unordered_map<std::string, SubscriptionCallback> _cbs;
         mutable std::mutex _cbsMutex;
 
-        // Message Queue can be touched on control+background thread, 
+        // Message Queue can be touched on control+background thread,
         // protecting with a mutex.
         //
-        // Message queue is used when there are problems sending messages so 
+        // Message queue is used when there are problems sending messages so
         // that sending can be retried later.
         std::deque<std::string> _messageQueue;
         mutable std::mutex _queueMutex;
@@ -169,5 +159,5 @@ namespace ix
         // Cap the queue size (100 elems so far -> ~100k)
         static constexpr size_t kQueueMaxSize = 256;
     };
-    
+
 } // namespace ix

ws/cobra_publisher/README.md

@@ -0,0 +1,6 @@
+```
+mkdir build
+cd build
+cmake ..
+make && (cd .. ; sh cobra_publisher.sh)
+```

ws/cobra_publisher/cobra_publisher.cpp

@@ -0,0 +1,123 @@
+/*
+ *  cobra_publisher.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <iostream>
+#include <sstream>
+#include <fstream>
+#include <atomic>
+#include <ixwebsocket/IXWebSocket.h>
+#include "IXCobraConnection.h"
+#include "jsoncpp/json/json.h"
+
+void msleep(int ms)
+{
+    std::chrono::duration<double, std::milli> duration(ms);
+    std::this_thread::sleep_for(duration);
+}
+
+int main(int argc, char* argv[])
+{
+    if (argc != 7)
+    {
+        std::cerr << "Usage error: need 6 arguments." << std::endl;
+    }
+
+    std::string endpoint = argv[1];
+    std::string appkey = argv[2];
+    std::string channel = argv[3];
+    std::string rolename = argv[4];
+    std::string rolesecret = argv[5];
+    std::string path = argv[6];
+
+    std::atomic<size_t> incomingBytes(0);
+    std::atomic<size_t> outgoingBytes(0);
+    ix::CobraConnection::setTrafficTrackerCallback(
+        [&incomingBytes, &outgoingBytes](size_t size, bool incoming)
+        {
+            if (incoming)
+            {
+                incomingBytes += size;
+            }
+            else
+            {
+                outgoingBytes += size;
+            }
+        }
+    );
+
+    bool done = false;
+    ix::CobraConnection cobraConnection;
+    ix::WebSocketPerMessageDeflateOptions webSocketPerMessageDeflateOptions(
+        true, false, false, 15, 15);
+    cobraConnection.configure(appkey, endpoint, rolename, rolesecret,
+                               webSocketPerMessageDeflateOptions);
+    cobraConnection.connect();
+    cobraConnection.setEventCallback(
+        [&cobraConnection, channel, path, &done]
+        (ix::CobraConnectionEventType eventType,
+         const std::string& errMsg,
+         const ix::WebSocketHttpHeaders& headers)
+        {
+            if (eventType == ix::CobraConnection_EventType_Open)
+            {
+                std::cout << "Handshake Headers:" << std::endl;
+                for (auto it : headers)
+                {
+                    std::cout << it.first << ": " << it.second << std::endl;
+                }
+            }
+            else if (eventType == ix::CobraConnection_EventType_Authenticated)
+            {
+                std::cout << "Authenticated" << std::endl;
+
+                std::string line;
+                std::ifstream f(path);
+                if (!f.is_open())
+                {
+                    std::cerr << "Error while opening file: " << path << std::endl;
+                }
+
+                int n = 0;
+                while (getline(f, line))
+                {
+                    Json::Value value;
+                    Json::Reader reader;
+                    reader.parse(line, value);
+
+                    cobraConnection.publish(channel, value);
+                    n++;
+                }
+                std::cerr << "#published messages: " << n << std::endl;
+
+                if (f.bad())
+                {
+                    std::cerr << "Error while opening file: " << path << std::endl;
+                }
+
+                done = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Error)
+            {
+                std::cerr << "Cobra Error received: " << errMsg << std::endl;
+                done = true;
+            }
+            else if (eventType == ix::CobraConnection_EventType_Closed)
+            {
+                std::cerr << "Cobra connection closed" << std::endl;
+            }
+        }
+    );
+
+    while (!done)
+    {
+        msleep(1);
+    }
+
+    std::cout << "Incoming bytes: " << incomingBytes << std::endl;
+    std::cout << "Outgoing bytes: " << outgoingBytes << std::endl;
+
+    return 0;
+}

ws/cobra_publisher/cobra_publisher.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+
+endpoint="ws://127.0.0.1:8765"
+endpoint="ws://127.0.0.1:5678"
+appkey="appkey"
+channel="foo"
+rolename="a_role"
+rolesecret="a_secret"
+filename=${FILENAME:=events.jsonl}
+
+build/cobra_publisher $endpoint $appkey $channel $rolename $rolesecret $filename

ws/cobra_publisher/devnull_server.js

@@ -0,0 +1,45 @@
+/*
+ *  devnull_server.js
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
+ */
+const WebSocket = require('ws');
+
+let wss = new WebSocket.Server({ port: 5678, perMessageDeflate: true })
+
+wss.on('connection', (ws) => {
+
+  let handshake = false
+  let authenticated = false
+
+  ws.on('message', (data) => {
+
+    console.log(data.toString('utf-8'))
+
+    if (!handshake) {
+      let response = {
+          "action": "auth/handshake/ok",
+          "body": {
+              "data": {
+                  "nonce": "MTI0Njg4NTAyMjYxMzgxMzgzMg==",
+                  "version": "0.0.24"
+              }
+          },
+          "id": 1
+      }
+      ws.send(JSON.stringify(response))
+      handshake = true
+    } else if (!authenticated) {
+      let response = {
+        "action": "auth/authenticate/ok",
+        "body": {},
+        "id": 2
+      }
+      
+      ws.send(JSON.stringify(response))
+      authenticated = true
+    } else {
+      console.log(data)
+    }
+  });
+})

ws/cobra_publisher/devnull_server.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+
+import os
+import json
+import asyncio
+import websockets
+
+
+async def echo(websocket, path):
+    handshake = False
+    authenticated = False
+
+    async for message in websocket:
+        print(message)
+
+        if not handshake:
+            response = {
+                "action": "auth/handshake/ok",
+                "body": {
+                    "data": {
+                        "nonce": "MTI0Njg4NTAyMjYxMzgxMzgzMg==",
+                        "version": "0.0.24"
+                    }
+                },
+                "id": 1
+            }
+            await websocket.send(json.dumps(response))
+            handshake = True
+
+        elif not authenticated:
+            response = {
+                "action": "auth/authenticate/ok",
+                "body": {},
+                "id": 2
+            }
+          
+            await websocket.send(json.dumps(response))
+            authenticated = True
+
+
+asyncio.get_event_loop().run_until_complete(
+    websockets.serve(echo, 'localhost', 5678))
+asyncio.get_event_loop().run_forever()

ws/cobra_publisher/events.jsonl

@@ -0,0 +1,3 @@
+{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Foo"}
+{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Bar"}
+{"array":[1,2,3],"boolean":true,"color":"#82b92c","null":null,"number":123,"object":{"a":"b","c":"d","e":"f"},"string":"Baz"}

ws/cobra_publisher/jsoncpp/json/json-forwards.h

@@ -0,0 +1,333 @@
+/// Json-cpp amalgated forward header (http://jsoncpp.sourceforge.net/).
+/// It is intended to be used with #include "json/json-forwards.h"
+/// This header provides forward declaration for all JsonCpp types.
+
+// //////////////////////////////////////////////////////////////////////
+// Beginning of content of file: LICENSE
+// //////////////////////////////////////////////////////////////////////
+
+/*
+The JsonCpp library's source code, including accompanying documentation,
+tests and demonstration applications, are licensed under the following
+conditions...
+
+Baptiste Lepilleur and The JsonCpp Authors explicitly disclaim copyright in all
+jurisdictions which recognize such a disclaimer. In such jurisdictions,
+this software is released into the Public Domain.
+
+In jurisdictions which do not recognize Public Domain property (e.g. Germany as of
+2010), this software is Copyright (c) 2007-2010 by Baptiste Lepilleur and
+The JsonCpp Authors, and is released under the terms of the MIT License (see below).
+
+In jurisdictions which recognize Public Domain property, the user of this
+software may choose to accept it either as 1) Public Domain, 2) under the
+conditions of the MIT License (see below), or 3) under the terms of dual
+Public Domain/MIT License conditions described here, as they choose.
+
+The MIT License is about as close to Public Domain as a license can get, and is
+described in clear, concise terms at:
+
+   http://en.wikipedia.org/wiki/MIT_License
+
+The full text of the MIT License follows:
+
+========================================================================
+Copyright (c) 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
+
+Permission is hereby granted, free of charge, to any person
+obtaining a copy of this software and associated documentation
+files (the "Software"), to deal in the Software without
+restriction, including without limitation the rights to use, copy,
+modify, merge, publish, distribute, sublicense, and/or sell copies
+of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+========================================================================
+(END LICENSE TEXT)
+
+The MIT license is compatible with both the GPL and commercial
+software, affording one all of the rights of Public Domain with the
+minor nuisance of being required to keep the above copyright notice
+and license text in the source code. Note also that by accepting the
+Public Domain "license" you can re-license your copy using whatever
+license you like.
+
+*/
+
+// //////////////////////////////////////////////////////////////////////
+// End of content of file: LICENSE
+// //////////////////////////////////////////////////////////////////////
+
+
+
+
+
+#ifndef JSON_FORWARD_AMALGATED_H_INCLUDED
+# define JSON_FORWARD_AMALGATED_H_INCLUDED
+/// If defined, indicates that the source file is amalgated
+/// to prevent private header inclusion.
+#define JSON_IS_AMALGAMATION
+
+// //////////////////////////////////////////////////////////////////////
+// Beginning of content of file: include/json/config.h
+// //////////////////////////////////////////////////////////////////////
+
+// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
+// Distributed under MIT license, or public domain if desired and
+// recognized in your jurisdiction.
+// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
+
+#ifndef JSON_CONFIG_H_INCLUDED
+#define JSON_CONFIG_H_INCLUDED
+#include <stddef.h>
+#include <string> //typedef String
+#include <stdint.h> //typedef int64_t, uint64_t
+
+/// If defined, indicates that json library is embedded in CppTL library.
+//# define JSON_IN_CPPTL 1
+
+/// If defined, indicates that json may leverage CppTL library
+//#  define JSON_USE_CPPTL 1
+/// If defined, indicates that cpptl vector based map should be used instead of
+/// std::map
+/// as Value container.
+//#  define JSON_USE_CPPTL_SMALLMAP 1
+
+// If non-zero, the library uses exceptions to report bad input instead of C
+// assertion macros. The default is to use exceptions.
+#ifndef JSON_USE_EXCEPTION
+#define JSON_USE_EXCEPTION 1
+#endif
+
+/// If defined, indicates that the source file is amalgated
+/// to prevent private header inclusion.
+/// Remarks: it is automatically defined in the generated amalgated header.
+// #define JSON_IS_AMALGAMATION
+
+#ifdef JSON_IN_CPPTL
+#include <cpptl/config.h>
+#ifndef JSON_USE_CPPTL
+#define JSON_USE_CPPTL 1
+#endif
+#endif
+
+#ifdef JSON_IN_CPPTL
+#define JSON_API CPPTL_API
+#elif defined(JSON_DLL_BUILD)
+#if defined(_MSC_VER) || defined(__MINGW32__)
+#define JSON_API __declspec(dllexport)
+#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
+#endif // if defined(_MSC_VER)
+#elif defined(JSON_DLL)
+#if defined(_MSC_VER) || defined(__MINGW32__)
+#define JSON_API __declspec(dllimport)
+#define JSONCPP_DISABLE_DLL_INTERFACE_WARNING
+#endif // if defined(_MSC_VER)
+#endif // ifdef JSON_IN_CPPTL
+#if !defined(JSON_API)
+#define JSON_API
+#endif
+
+// If JSON_NO_INT64 is defined, then Json only support C++ "int" type for
+// integer
+// Storages, and 64 bits integer support is disabled.
+// #define JSON_NO_INT64 1
+
+#if defined(_MSC_VER) // MSVC
+#  if _MSC_VER <= 1200 // MSVC 6
+    // Microsoft Visual Studio 6 only support conversion from __int64 to double
+    // (no conversion from unsigned __int64).
+#    define JSON_USE_INT64_DOUBLE_CONVERSION 1
+    // Disable warning 4786 for VS6 caused by STL (identifier was truncated to '255'
+    // characters in the debug information)
+    // All projects I've ever seen with VS6 were using this globally (not bothering
+    // with pragma push/pop).
+#    pragma warning(disable : 4786)
+#  endif // MSVC 6
+
+#  if _MSC_VER >= 1500 // MSVC 2008
+    /// Indicates that the following function is deprecated.
+#    define JSONCPP_DEPRECATED(message) __declspec(deprecated(message))
+#  endif
+
+#endif // defined(_MSC_VER)
+
+// In c++11 the override keyword allows you to explicity define that a function
+// is intended to override the base-class version.  This makes the code more
+// managable and fixes a set of common hard-to-find bugs.
+#if __cplusplus >= 201103L
+# define JSONCPP_OVERRIDE override
+# define JSONCPP_NOEXCEPT noexcept
+#elif defined(_MSC_VER) && _MSC_VER > 1600 && _MSC_VER < 1900
+# define JSONCPP_OVERRIDE override
+# define JSONCPP_NOEXCEPT throw()
+#elif defined(_MSC_VER) && _MSC_VER >= 1900
+# define JSONCPP_OVERRIDE override
+# define JSONCPP_NOEXCEPT noexcept
+#else
+# define JSONCPP_OVERRIDE
+# define JSONCPP_NOEXCEPT throw()
+#endif
+
+#ifndef JSON_HAS_RVALUE_REFERENCES
+
+#if defined(_MSC_VER) && _MSC_VER >= 1600 // MSVC >= 2010
+#define JSON_HAS_RVALUE_REFERENCES 1
+#endif // MSVC >= 2010
+
+#ifdef __clang__
+#if __has_feature(cxx_rvalue_references)
+#define JSON_HAS_RVALUE_REFERENCES 1
+#endif  // has_feature
+
+#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
+#if defined(__GXX_EXPERIMENTAL_CXX0X__) || (__cplusplus >= 201103L)
+#define JSON_HAS_RVALUE_REFERENCES 1
+#endif  // GXX_EXPERIMENTAL
+
+#endif // __clang__ || __GNUC__
+
+#endif // not defined JSON_HAS_RVALUE_REFERENCES
+
+#ifndef JSON_HAS_RVALUE_REFERENCES
+#define JSON_HAS_RVALUE_REFERENCES 0
+#endif
+
+#ifdef __clang__
+#  if __has_extension(attribute_deprecated_with_message)
+#    define JSONCPP_DEPRECATED(message)  __attribute__ ((deprecated(message)))
+#  endif
+#elif defined __GNUC__ // not clang (gcc comes later since clang emulates gcc)
+#  if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 5))
+#    define JSONCPP_DEPRECATED(message)  __attribute__ ((deprecated(message)))
+#  elif (__GNUC__ > 3 || (__GNUC__ == 3 && __GNUC_MINOR__ >= 1))
+#    define JSONCPP_DEPRECATED(message)  __attribute__((__deprecated__))
+#  endif  // GNUC version
+#endif // __clang__ || __GNUC__
+
+#if !defined(JSONCPP_DEPRECATED)
+#define JSONCPP_DEPRECATED(message)
+#endif // if !defined(JSONCPP_DEPRECATED)
+
+#if __GNUC__ >= 6
+#  define JSON_USE_INT64_DOUBLE_CONVERSION 1
+#endif
+
+#if !defined(JSON_IS_AMALGAMATION)
+
+# include "version.h"
+
+# if JSONCPP_USING_SECURE_MEMORY
+#  include "allocator.h" //typedef Allocator
+# endif
+
+#endif // if !defined(JSON_IS_AMALGAMATION)
+
+namespace Json {
+typedef int Int;
+typedef unsigned int UInt;
+#if defined(JSON_NO_INT64)
+typedef int LargestInt;
+typedef unsigned int LargestUInt;
+#undef JSON_HAS_INT64
+#else                 // if defined(JSON_NO_INT64)
+// For Microsoft Visual use specific types as long long is not supported
+#if defined(_MSC_VER) // Microsoft Visual Studio
+typedef __int64 Int64;
+typedef unsigned __int64 UInt64;
+#else                 // if defined(_MSC_VER) // Other platforms, use long long
+typedef int64_t Int64;
+typedef uint64_t UInt64;
+#endif // if defined(_MSC_VER)
+typedef Int64 LargestInt;
+typedef UInt64 LargestUInt;
+#define JSON_HAS_INT64
+#endif // if defined(JSON_NO_INT64)
+#if JSONCPP_USING_SECURE_MEMORY
+#define JSONCPP_STRING        std::basic_string<char, std::char_traits<char>, Json::SecureAllocator<char> >
+#define JSONCPP_OSTRINGSTREAM std::basic_ostringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
+#define JSONCPP_OSTREAM       std::basic_ostream<char, std::char_traits<char>>
+#define JSONCPP_ISTRINGSTREAM std::basic_istringstream<char, std::char_traits<char>, Json::SecureAllocator<char> >
+#define JSONCPP_ISTREAM       std::istream
+#else
+#define JSONCPP_STRING        std::string
+#define JSONCPP_OSTRINGSTREAM std::ostringstream
+#define JSONCPP_OSTREAM       std::ostream
+#define JSONCPP_ISTRINGSTREAM std::istringstream
+#define JSONCPP_ISTREAM       std::istream
+#endif // if JSONCPP_USING_SECURE_MEMORY
+} // end namespace Json
+
+#endif // JSON_CONFIG_H_INCLUDED
+
+// //////////////////////////////////////////////////////////////////////
+// End of content of file: include/json/config.h
+// //////////////////////////////////////////////////////////////////////
+
+
+
+
+
+
+// //////////////////////////////////////////////////////////////////////
+// Beginning of content of file: include/json/forwards.h
+// //////////////////////////////////////////////////////////////////////
+
+// Copyright 2007-2010 Baptiste Lepilleur and The JsonCpp Authors
+// Distributed under MIT license, or public domain if desired and
+// recognized in your jurisdiction.
+// See file LICENSE for detail or copy at http://jsoncpp.sourceforge.net/LICENSE
+
+#ifndef JSON_FORWARDS_H_INCLUDED
+#define JSON_FORWARDS_H_INCLUDED
+
+#if !defined(JSON_IS_AMALGAMATION)
+#include "config.h"
+#endif // if !defined(JSON_IS_AMALGAMATION)
+
+namespace Json {
+
+// writer.h
+class FastWriter;
+class StyledWriter;
+
+// reader.h
+class Reader;
+
+// features.h
+class Features;
+
+// value.h
+typedef unsigned int ArrayIndex;
+class StaticString;
+class Path;
+class PathArgument;
+class Value;
+class ValueIteratorBase;
+class ValueIterator;
+class ValueConstIterator;
+
+} // namespace Json
+
+#endif // JSON_FORWARDS_H_INCLUDED
+
+// //////////////////////////////////////////////////////////////////////
+// End of content of file: include/json/forwards.h
+// //////////////////////////////////////////////////////////////////////
+
+
+
+
+
+#endif //ifndef JSON_FORWARD_AMALGATED_H_INCLUDED

ws/cobra_publisher/package-lock.json

@@ -0,0 +1,19 @@
+{
+  "requires": true,
+  "lockfileVersion": 1,
+  "dependencies": {
+    "async-limiter": {
+      "version": "1.0.0",
+      "resolved": "https://registry.npmjs.org/async-limiter/-/async-limiter-1.0.0.tgz",
+      "integrity": "sha512-jp/uFnooOiO+L211eZOoSyzpOITMXx1rBITauYykG3BRYPu8h0UcxsPNB04RR5vo4Tyz3+ay17tR6JVf9qzYWg=="
+    },
+    "ws": {
+      "version": "6.1.0",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-6.1.0.tgz",
+      "integrity": "sha512-H3dGVdGvW2H8bnYpIDc3u3LH8Wue3Qh+Zto6aXXFzvESkTVT6rAfKR6tR/+coaUvxs8yHtmNV0uioBF62ZGSTg==",
+      "requires": {
+        "async-limiter": "1.0.0"
+      }
+    }
+  }
+}

ws/ixcrypto/IXHMac.cpp

@@ -6,11 +6,7 @@
 #include "IXHMac.h"
 #include "IXBase64.h"
 
-#ifdef __APPLE__
-# include <CommonCrypto/CommonHMAC.h>
-#else
-# include <openssl/hmac.h>
-#endif
+#include <openssl/hmac.h>
 
 namespace ix
 {
@@ -19,17 +15,10 @@ namespace ix
         constexpr size_t hashSize = 16;
         unsigned char hash[hashSize];
 
-#ifdef __APPLE__
-        CCHmac(kCCHmacAlgMD5,
-               key.c_str(), key.size(),
-               data.c_str(), data.size(),
-               &hash);
-#else
         HMAC(EVP_md5(),
              key.c_str(), (int) key.size(),
              (unsigned char *) data.c_str(), (int) data.size(),
              (unsigned char *) hash, nullptr);
-#endif
 
         std::string hashString(reinterpret_cast<char*>(hash), hashSize);
 

ws/ws.cpp

@@ -38,28 +38,18 @@ int main(int argc, char** argv)
     std::string channel;
     std::string message;
     std::string password;
-    std::string appkey;
-    std::string endpoint;
-    std::string rolename;
-    std::string rolesecret;
-    std::string prefix("ws.test.v0");
-    std::string fields;
-    std::string dsn;
     bool headersOnly = false;
     bool followRedirects = false;
     bool verbose = false;
     bool save = false;
     bool compress = false;
-    bool strict = false;
     int port = 8080;
     int redisPort = 6379;
-    int statsdPort = 8125;
     int connectTimeOut = 60;
     int transferTimeout = 1800;
     int maxRedirects = 5;
     int delayMs = -1;
     int count = 1;
-    int jobs = 4;
 
     CLI::App* sendApp = app.add_subcommand("send", "Send a file");
     sendApp->add_option("url", url, "Connection url")->required();
@@ -127,40 +117,6 @@ int main(int argc, char** argv)
     redisSubscribeApp->add_flag("-v", verbose, "Verbose");
     redisSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
 
-    CLI::App* cobraSubscribeApp = app.add_subcommand("cobra_subscribe", "Cobra subscriber");
-    cobraSubscribeApp->add_option("--appkey", appkey, "Appkey");
-    cobraSubscribeApp->add_option("--endpoint", endpoint, "Endpoint");
-    cobraSubscribeApp->add_option("--rolename", rolename, "Role name");
-    cobraSubscribeApp->add_option("--rolesecret", rolesecret, "Role secret");
-    cobraSubscribeApp->add_option("channel", channel, "Channel")->required();
-    cobraSubscribeApp->add_flag("-v", verbose, "Verbose");
-    cobraSubscribeApp->add_option("--pidfile", pidfile, "Pid file");
-
-    CLI::App* cobra2statsd = app.add_subcommand("cobra_to_statsd", "Cobra to statsd");
-    cobra2statsd->add_option("--appkey", appkey, "Appkey");
-    cobra2statsd->add_option("--endpoint", endpoint, "Endpoint");
-    cobra2statsd->add_option("--rolename", rolename, "Role name");
-    cobra2statsd->add_option("--rolesecret", rolesecret, "Role secret");
-    cobra2statsd->add_option("--host", hostname, "Statsd host");
-    cobra2statsd->add_option("--port", statsdPort, "Statsd port");
-    cobra2statsd->add_option("--prefix", prefix, "Statsd prefix");
-    cobra2statsd->add_option("--fields", fields, "Extract fields for naming the event")->join();
-    cobra2statsd->add_option("channel", channel, "Channel")->required();
-    cobra2statsd->add_flag("-v", verbose, "Verbose");
-    cobra2statsd->add_option("--pidfile", pidfile, "Pid file");
-
-    CLI::App* cobra2sentry = app.add_subcommand("cobra_to_sentry", "Cobra to sentry");
-    cobra2sentry->add_option("--appkey", appkey, "Appkey");
-    cobra2sentry->add_option("--endpoint", endpoint, "Endpoint");
-    cobra2sentry->add_option("--rolename", rolename, "Role name");
-    cobra2sentry->add_option("--rolesecret", rolesecret, "Role secret");
-    cobra2sentry->add_option("--dsn", dsn, "Sentry DSN");
-    cobra2sentry->add_option("--jobs", jobs, "Number of thread sending events to Sentry");
-    cobra2sentry->add_option("channel", channel, "Channel")->required();
-    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");
-
     CLI11_PARSE(app, argc, argv);
 
     // pid file handling
@@ -223,26 +179,6 @@ int main(int argc, char** argv)
     {
         return ix::ws_redis_subscribe_main(hostname, redisPort, password, channel, verbose);
     }
-    else if (app.got_subcommand("cobra_subscribe"))
-    {
-        return ix::ws_cobra_subscribe_main(appkey, endpoint,
-                                           rolename, rolesecret,
-                                           channel, verbose);
-    }
-    else if (app.got_subcommand("cobra_to_statsd"))
-    {
-        return ix::ws_cobra_to_statsd_main(appkey, endpoint,
-                                           rolename, rolesecret,
-                                           channel, hostname, statsdPort,
-                                           prefix, fields, verbose);
-    }
-    else if (app.got_subcommand("cobra_to_sentry"))
-    {
-        return ix::ws_cobra_to_sentry_main(appkey, endpoint,
-                                           rolename, rolesecret,
-                                           channel, dsn,
-                                           verbose, strict, jobs);
-    }
 
     return 1;
 }

ws/ws.h

@@ -52,32 +52,4 @@ namespace ix
                                 const std::string& password,
                                 const std::string& channel,
                                 bool verbose);
-
-    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,
-                                bool verbose);
-
-    int ws_cobra_to_statsd_main(const std::string& appkey,
-                                const std::string& endpoint,
-                                const std::string& rolename,
-                                const std::string& rolesecret,
-                                const std::string& channel,
-                                const std::string& host,
-                                int port,
-                                const std::string& prefix,
-                                const std::string& fields,
-                                bool verbose);
-    
-    int ws_cobra_to_sentry_main(const std::string& appkey,
-                                const std::string& endpoint,
-                                const std::string& rolename,
-                                const std::string& rolesecret,
-                                const std::string& channel,
-                                const std::string& dsn,
-                                bool verbose,
-                                bool strict,
-                                int jobs);
 }

ws/ws_cobra_subscribe.cpp

@@ -1,76 +0,0 @@
-/*
- *  ws_cobra_subscribe.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include <chrono>
-#include <thread>
-#include <atomic>
-#include "IXCobraConnection.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,
-                                bool verbose)
-    {
-
-        ix::CobraConnection conn;
-        conn.configure(appkey, endpoint,
-                       rolename, rolesecret,
-                       ix::WebSocketPerMessageDeflateOptions(true));
-        conn.connect();
-
-        Json::FastWriter jsonWriter;
-
-        conn.setEventCallback(
-            [&conn, &channel, &jsonWriter]
-            (ix::CobraConnectionEventType eventType,
-             const std::string& errMsg,
-             const ix::WebSocketHttpHeaders& headers,
-             const std::string& subscriptionId)
-            {
-                if (eventType == ix::CobraConnection_EventType_Open)
-                {
-                    std::cout << "Subscriber: connected" << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Authenticated)
-                {
-                    std::cout << "Subscriber authenticated" << std::endl;
-                    conn.subscribe(channel,
-                                   [&jsonWriter](const Json::Value& msg)
-                                   {
-                                       // std::cout << "Received message" << std::endl;
-                                       std::cout << jsonWriter.write(msg) << std::endl;
-                                   });
-                }
-                else if (eventType == ix::CobraConnection_EventType_Subscribed)
-                {
-                    std::cout << "Subscriber: subscribed to channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
-                {
-                    std::cout << "Subscriber: unsubscribed from channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Error)
-                {
-                    std::cout << "Subscriber: error" << errMsg << std::endl;
-                }
-            }
-        );
-
-        while (true)
-        {
-            std::chrono::duration<double, std::milli> duration(10);
-            std::this_thread::sleep_for(duration);
-        }
-
-        return 0;
-    }
-}

ws/ws_cobra_to_sentry.cpp

@@ -1,189 +0,0 @@
-/*
- *  ws_cobra_to_sentry.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include <chrono>
-#include <thread>
-#include <atomic>
-#include <vector>
-#include <queue>
-#include <mutex>
-#include <condition_variable>
-#include "IXCobraConnection.h"
-
-#include "IXSentryClient.h"
-
-namespace ix
-{
-    int ws_cobra_to_sentry_main(const std::string& appkey,
-                                const std::string& endpoint,
-                                const std::string& rolename,
-                                const std::string& rolesecret,
-                                const std::string& channel,
-                                const std::string& dsn,
-                                bool verbose,
-                                bool strict,
-                                int jobs)
-    {
-        ix::CobraConnection conn;
-        conn.configure(appkey, endpoint,
-                       rolename, rolesecret,
-                       ix::WebSocketPerMessageDeflateOptions(true));
-        conn.connect();
-
-        Json::FastWriter jsonWriter;
-        std::atomic<uint64_t> sentCount(0);
-        std::atomic<uint64_t> receivedCount(0);
-        std::atomic<bool> errorSending(false);
-        std::atomic<bool> stop(false);
-
-        std::mutex conditionVariableMutex;
-        std::condition_variable condition;
-        std::condition_variable progressCondition;
-        std::queue<Json::Value> queue;
-
-        auto sentrySender = [&condition, &progressCondition, &conditionVariableMutex,
-                             &queue, verbose, &errorSending, &sentCount,
-                             &stop, &dsn]
-        {
-            while (true)
-            {
-                Json::Value msg;
-
-                {
-                    std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                    condition.wait(lock, [&queue, &stop]{ return !queue.empty() && !stop; });
-
-                    msg = queue.front();
-                    queue.pop();
-                }
-
-                SentryClient sc(dsn);
-
-                if (!sc.send(msg, verbose))
-                {
-                    errorSending = true;
-                }
-                else
-                {
-                    ++sentCount;
-                }
-
-                progressCondition.notify_one();
-
-                if (stop) return;
-            }
-        };
-
-        // Create a thread pool
-        std::cerr << "Starting " << jobs << " sentry sender jobs" << std::endl;
-        std::vector<std::thread> pool;
-        for (int i = 0; i < jobs; i++)
-        {
-            pool.push_back(std::thread(sentrySender));
-        }
-
-        conn.setEventCallback(
-            [&conn, &channel, &jsonWriter, 
-             verbose, &receivedCount, &sentCount,
-             &condition, &conditionVariableMutex, 
-             &progressCondition, &queue]
-            (ix::CobraConnectionEventType eventType,
-             const std::string& errMsg,
-             const ix::WebSocketHttpHeaders& headers,
-             const std::string& subscriptionId)
-            {
-                if (eventType == ix::CobraConnection_EventType_Open)
-                {
-                    std::cerr << "Subscriber: connected" << std::endl;
-
-                    for (auto it : headers)
-                    {
-                        std::cerr << it.first << ": " << it.second << std::endl;
-                    }
-                }
-                if (eventType == ix::CobraConnection_EventType_Closed)
-                {
-                    std::cerr << "Subscriber: closed" << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Authenticated)
-                {
-                    std::cerr << "Subscriber authenticated" << std::endl;
-                    conn.subscribe(channel,
-                                   [&jsonWriter, verbose, 
-                                    &sentCount, &receivedCount,
-                                    &condition, &conditionVariableMutex,
-                                    &progressCondition, &queue]
-                                   (const Json::Value& msg)
-                                   {
-                                       if (verbose)
-                                       {
-                                           std::cerr << jsonWriter.write(msg) << std::endl;
-                                       }
-
-                                       // If we cannot send to sentry fast enough, drop the message
-                                       const uint64_t scaleFactor = 2;
-
-                                       if (sentCount != 0 &&
-                                           receivedCount != 0 &&
-                                           (sentCount * scaleFactor < receivedCount))
-                                       {
-                                           std::cerr << "message dropped: sending is backlogged !" 
-                                                     << std::endl;
-
-                                           condition.notify_one();
-                                           progressCondition.notify_one();
-                                           return;
-                                       }
-
-                                       ++receivedCount;
-
-                                       {
-                                           std::unique_lock<std::mutex> lock(conditionVariableMutex);
-                                           queue.push(msg);
-                                       }
-
-                                       condition.notify_one();
-                                       progressCondition.notify_one();
-                                   });
-                }
-                else if (eventType == ix::CobraConnection_EventType_Subscribed)
-                {
-                    std::cerr << "Subscriber: subscribed to channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
-                {
-                    std::cerr << "Subscriber: unsubscribed from channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Error)
-                {
-                    std::cerr << "Subscriber: error" << errMsg << std::endl;
-                }
-            }
-        );
-
-        std::mutex progressConditionVariableMutex;
-        while (true)
-        {
-            std::unique_lock<std::mutex> lock(progressConditionVariableMutex);
-            progressCondition.wait(lock);
-
-            std::cout << "messages"
-                      << " received " << receivedCount
-                      << " sent " << sentCount
-                      << std::endl;
-
-            if (strict && errorSending) break;
-        }
-
-        conn.disconnect();
-
-        // FIXME: join all the bg threads and stop them.
-
-        return 0;
-    }
-}

ws/ws_cobra_to_statsd.cpp

@@ -1,148 +0,0 @@
-/*
- *  ws_cobra_to_statsd.cpp
- *  Author: Benjamin Sergeant
- *  Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
- */
-
-#include <iostream>
-#include <sstream>
-#include <chrono>
-#include <thread>
-#include <atomic>
-#include <vector>
-#include "IXCobraConnection.h"
-
-#include <statsd_client.h>
-
-namespace ix
-{
-    // fields are command line argument that can be specified multiple times
-    std::vector<std::string> parseFields(const std::string& fields)
-    {
-        std::vector<std::string> tokens;
-
-        // Split by \n
-        std::string token;
-        std::stringstream tokenStream(fields);
-
-        while (std::getline(tokenStream, token))
-        {
-            tokens.push_back(token);
-        }
-
-        return tokens;
-    }
-
-    //
-    // Extract an attribute from a Json Value.
-    // extractAttr("foo.bar", {"foo": {"bar": "baz"}}) => baz
-    //
-    std::string extractAttr(const std::string& attr,
-                            const Json::Value& jsonValue)
-    {
-        // Split by .
-        std::string token;
-        std::stringstream tokenStream(attr);
-
-        Json::Value val(jsonValue);
-
-        int i = 0;
-        while (std::getline(tokenStream, token, '.'))
-        {
-            val = val[token];
-        }
-
-        return val.asString();
-    }
-
-    int ws_cobra_to_statsd_main(const std::string& appkey,
-                                const std::string& endpoint,
-                                const std::string& rolename,
-                                const std::string& rolesecret,
-                                const std::string& channel,
-                                const std::string& host,
-                                int port,
-                                const std::string& prefix,
-                                const std::string& fields,
-                                bool verbose)
-    {
-        ix::CobraConnection conn;
-        conn.configure(appkey, endpoint,
-                       rolename, rolesecret,
-                       ix::WebSocketPerMessageDeflateOptions(true));
-        conn.connect();
-
-        auto tokens = parseFields(fields);
-
-        // statsd client
-        // test with netcat as a server: `nc -ul 8125`
-        bool statsdBatch = true;
-        statsd::StatsdClient statsdClient(host, port, prefix, statsdBatch);
-
-        Json::FastWriter jsonWriter;
-        uint64_t msgCount = 0;
-
-        conn.setEventCallback(
-            [&conn, &channel, &jsonWriter, &statsdClient, verbose, &tokens, &prefix, &msgCount]
-            (ix::CobraConnectionEventType eventType,
-             const std::string& errMsg,
-             const ix::WebSocketHttpHeaders& headers,
-             const std::string& subscriptionId)
-            {
-                if (eventType == ix::CobraConnection_EventType_Open)
-                {
-                    std::cout << "Subscriber: connected" << std::endl;
-                }
-                if (eventType == ix::CobraConnection_EventType_Closed)
-                {
-                    std::cout << "Subscriber: closed" << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Authenticated)
-                {
-                    std::cout << "Subscriber authenticated" << std::endl;
-                    conn.subscribe(channel,
-                                   [&jsonWriter, &statsdClient, &channel,
-                                    verbose, &tokens, &prefix, &msgCount]
-                                   (const Json::Value& msg)
-                                   {
-                                       if (verbose)
-                                       {
-                                           std::cout << jsonWriter.write(msg) << std::endl;
-                                       }
-
-                                       std::string id;
-                                       for (auto&& attr : tokens)
-                                       {
-                                           id += ".";
-                                           id += extractAttr(attr, msg);
-                                       }
-
-                                       std::cout << msgCount++ << " " << prefix << id << std::endl;
-
-                                       statsdClient.count(id, 1);
-                                   });
-                }
-                else if (eventType == ix::CobraConnection_EventType_Subscribed)
-                {
-                    std::cout << "Subscriber: subscribed to channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
-                {
-                    std::cout << "Subscriber: unsubscribed from channel " << subscriptionId << std::endl;
-                }
-                else if (eventType == ix::CobraConnection_EventType_Error)
-                {
-                    std::cout << "Subscriber: error" << errMsg << std::endl;
-                }
-            }
-        );
-
-        while (true)
-        {
-            std::chrono::duration<double, std::milli> duration(1000);
-            std::this_thread::sleep_for(duration);
-        }
-
-        return 0;
-    }
-}