kapitan/IXWebSocket
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add new example folder for publishing events to satori, with a minimal satori sdk

Browse Source
This commit is contained in:
Benjamin Sergeant 2018-11-07 12:26:32 -08:00
parent 2b203c4616
commit 6dbfe28427
18 changed files with 8886 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CMakeLists.txt
View File

@ -16,6 +16,7 @@ set( IXWEBSOCKET_SOURCES
ixwebsocket/IXSocket.cpp
ixwebsocket/IXWebSocket.cpp
ixwebsocket/IXWebSocketTransport.cpp
ixwebsocket/IXWebSocketPerMessageDeflate.cpp
)
set( IXWEBSOCKET_HEADERS
@ -23,6 +24,7 @@ set( IXWEBSOCKET_HEADERS
ixwebsocket/IXSocket.h
ixwebsocket/IXWebSocket.h
ixwebsocket/IXWebSocketTransport.h
ixwebsocket/IXWebSocketPerMessageDeflate.h
)
if (USE_TLS)
@ -42,6 +44,8 @@ add_library( ixwebsocket STATIC
${IXWEBSOCKET_HEADERS}
)
target_link_libraries(ixwebsocket "z")
set( IXWEBSOCKET_INCLUDE_DIRS
.
../../shared/OpenSSL/include)

2
examples/chat/package.json
View File

@ -1,6 +1,6 @@
{
"dependencies": {
"msgpack-js": "^0.3.0",
"ws": "^3.1.0"
"ws": "^3.3.3"
}
}

36
examples/satori_publisher/CMakeLists.txt Normal file
View File

@ -0,0 +1,36 @@
#
# Author: Benjamin Sergeant
# Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
#
cmake_minimum_required (VERSION 3.4.1)
project (satori_publisher)
set (OPENSSL_PREFIX /usr/local/opt/openssl) # Homebrew openssl
set (CMAKE_CXX_STANDARD 11)
option(USE_TLS "Add TLS support" ON)
add_subdirectory(${PROJECT_SOURCE_DIR}/../.. ixwebsocket)
include_directories(satori_publisher ${OPENSSL_PREFIX}/include)
add_executable(satori_publisher
base64.cpp
jsoncpp/jsoncpp.cpp
IXHMac.cpp
IXSatoriConnection.cpp
satori_publisher.cpp)
if (APPLE AND USE_TLS)
target_link_libraries(satori_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(satori_publisher ixwebsocket lib_crypto)
install(TARGETS satori_publisher DESTINATION bin)

27
examples/satori_publisher/IXHMac.cpp Normal file
View File

@ -0,0 +1,27 @@
/*
* IXHMac.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
#include "IXHMac.h"
#include "base64.h"
#include <openssl/hmac.h>
namespace ix
{
std::string hmac(const std::string& data, const std::string& key)
{
constexpr size_t hashSize = 16;
unsigned char hash[hashSize];
HMAC(EVP_md5(),
key.c_str(), (int) key.size(),
(unsigned char *) data.c_str(), (int) data.size(),
(unsigned char *) hash, nullptr);
std::string hashString(reinterpret_cast<char*>(hash), hashSize);
return base64_encode(hashString, (uint32_t) hashString.size());
}
}

14
examples/satori_publisher/IXHMac.h Normal file
View File

@ -0,0 +1,14 @@
/*
* IXHMac.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
std::string hmac(const std::string& data, const std::string& key);
}

453
examples/satori_publisher/IXSatoriConnection.cpp Normal file
View File

@ -0,0 +1,453 @@
/*
* IXSatoriConnection.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2017-2018 Machine Zone. All rights reserved.
*/
#include "IXSatoriConnection.h"
#include "IXHMac.h"
#include <algorithm>
#include <stdexcept>
#include <cmath>
#include <cassert>
#include <cstring>
#include <iostream>
namespace ix
{
bool parseJson(const std::string& str, Json::Value& value)
{
Json::Reader reader;
return reader.parse(str, value);
}
std::string writeJsonCompact(const Json::Value& value)
{
Json::FastWriter writer;
return writer.write(value);
}
}
namespace ix
{
OnTrafficTrackerCallback SatoriConnection::_onTrafficTrackerCallback = nullptr;
constexpr size_t SatoriConnection::kQueueMaxSize;
SatoriConnection::SatoriConnection() :
_authenticated(false),
_authenticatedCallback(nullptr)
{
_pdu["action"] = "rtm/publish";
_webSocket.setOnMessageCallback(
[](ix::WebSocketMessageType messageType,
const std::string& str,
const ix::WebSocketErrorInfo& error,
const ix::CloseInfo& closeInfo)
{
;
}
);
}
SatoriConnection::~SatoriConnection()
{
disconnect();
}
void SatoriConnection::setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback)
{
_onTrafficTrackerCallback = callback;
}
void SatoriConnection::resetTrafficTrackerCallback()
{
setTrafficTrackerCallback(nullptr);
}
void SatoriConnection::invokeTrafficTrackerCallback(size_t size, bool incoming)
{
if (_onTrafficTrackerCallback)
{
_onTrafficTrackerCallback(size, incoming);
}
}
void SatoriConnection::setAuthenticatedCallback(const AuthenticatedCallback& authenticatedCallback)
{
_authenticatedCallback = authenticatedCallback;
}
void SatoriConnection::invokeAuthenticatedCallback()
{
if (_authenticatedCallback)
{
_authenticatedCallback();
}
}
void SatoriConnection::logError(const std::string& error)
{
std::cerr << "SatoriConnection: " << error;
}
void SatoriConnection::disconnect()
{
_webSocket.stop();
_webSocket.setOnMessageCallback(
[](ix::WebSocketMessageType messageType,
const std::string& str,
const ix::WebSocketErrorInfo& error,
const ix::CloseInfo& closeInfo)
{
;
}
);
}
void SatoriConnection::configure(const std::string& appkey,
const std::string& endpoint,
const std::string& rolename,
const std::string& rolesecret)
{
_appkey = appkey;
_endpoint = endpoint;
_role_name = rolename;
_role_secret = rolesecret;
std::stringstream ss;
ss << endpoint;
ss << "/v2?appkey=";
ss << appkey;
_webSocket.configure(ss.str());
_webSocket.setOnMessageCallback(
[this](ix::WebSocketMessageType messageType,
const std::string& str,
const ix::WebSocketErrorInfo& error,
const ix::CloseInfo& closeInfo)
{
std::stringstream ss;
if (messageType == ix::WebSocket_MessageType_Open)
{
sendHandshakeMessage();
}
else if (messageType == ix::WebSocket_MessageType_Close)
{
_authenticated = false;
}
else if (messageType == ix::WebSocket_MessageType_Message)
{
Json::Value data;
if (!parseJson(str, data))
{
logError(std::string("Invalid json: ") + str);
return;
}
if (!data.isMember("action"))
{
logError("Missing action");
return;
}
auto action = data["action"].asString();
if (action == "auth/handshake/ok")
{
if (!handleHandshakeResponse(data))
{
logError("Error extracting nonce from handshake response");
}
}
else if (action == "auth/handshake/error")
{
logError("Handshake error."); // print full message ?
}
else if (action == "auth/authenticate/ok")
{
_authenticated = true;
invokeAuthenticatedCallback();
flushQueue();
}
else if (action == "auth/authenticate/error")
{
logError("Authentication error."); // print full message ?
}
else if (action == "rtm/subscription/data")
{
handleSubscriptionData(data);
}
else
{
logError(std::string("Un-handled message type: ") + action);
}
}
else if (messageType == ix::WebSocket_MessageType_Error)
{
std::stringstream ss;
ss << "Connection error: " << error.reason << std::endl;
ss << "#retries: " << error.retries << std::endl;
ss << "Wait time(ms): " << error.wait_time << std::endl;
ss << "HTTP Status: " << error.http_status << std::endl;
logError(ss.str());
}
});
}
//
// Handshake message schema.
//
// handshake = {
// "action": "auth/handshake",
// "body": {
// "data": {
// "role": role
// },
// "method": "role_secret"
// },
// }
//
//
bool SatoriConnection::sendHandshakeMessage()
{
Json::Value data;
data["role"] = _role_name;
Json::Value body;
body["data"] = data;
body["method"] = "role_secret";
Json::Value pdu;
pdu["action"] = "auth/handshake";
pdu["body"] = body;
std::string serializedJson = writeJsonCompact(pdu);
SatoriConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
return _webSocket.send(serializedJson);
}
//
// Extract the nonce from the handshake response
// use it to compute a hash during authentication
//
// {
// "action": "auth/handshake/ok",
// "body": {
// "data": {
// "nonce": "MTI0Njg4NTAyMjYxMzgxMzgzMg==",
// "version": "0.0.24"
// }
// }
// }
//
bool SatoriConnection::handleHandshakeResponse(const Json::Value& pdu)
{
if (!pdu.isMember("body")) return false;
Json::Value body = pdu["body"];
if (!body.isMember("data")) return false;
Json::Value data = body["data"];
if (!data.isMember("nonce")) return false;
Json::Value nonce = data["nonce"];
if (!nonce.isString()) return false;
return sendAuthMessage(nonce.asString());
}
//
// Authenticate message schema.
//
// challenge = {
// "action": "auth/authenticate",
// "body": {
// "method": "role_secret",
// "credentials": {
// "hash": computeHash(secret, nonce)
// }
// },
// }
//
bool SatoriConnection::sendAuthMessage(const std::string& nonce)
{
Json::Value credentials;
credentials["hash"] = hmac(nonce, _role_secret);
Json::Value body;
body["credentials"] = credentials;
body["method"] = "role_secret";
Json::Value pdu;
pdu["action"] = "auth/authenticate";
pdu["body"] = body;
std::string serializedJson = writeJsonCompact(pdu);
SatoriConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
return _webSocket.send(serializedJson);
}
bool SatoriConnection::handleSubscriptionData(const Json::Value& pdu)
{
if (!pdu.isMember("body")) return false;
Json::Value body = pdu["body"];
// 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"];
auto cb = _cbs.find(subscriptionId.asString());
if (cb == _cbs.end()) return false; // cannot find callback
// Extract messages now
if (!body.isMember("messages")) return false;
Json::Value messages = body["messages"];
for (auto&& msg : messages)
{
cb->second(msg);
}
return true;
}
bool SatoriConnection::connect()
{
_webSocket.start();
return true;
}
bool SatoriConnection::isConnected() const
{
return _webSocket.getReadyState() == ix::WebSocket_ReadyState_Open;
}
//
// publish is not thread safe as we are trying to reuse some Json objects.
//
bool SatoriConnection::publish(const std::string& channel,
const Json::Value& msg)
{
_body["channel"] = channel;
_body["message"] = msg;
_pdu["body"] = _body;
std::string serializedJson = writeJsonCompact(_pdu);
//
// Fast path. We are authenticated and the publishing succeed
// This should happen for 99% of the cases.
//
if (_authenticated && publishMessage(serializedJson))
{
return true;
}
else // Or else we enqueue
// Slow code path is when we haven't connected yet (startup),
// or when the connection drops for some reason.
{
enqueue(serializedJson);
return false;
}
}
void SatoriConnection::subscribe(const std::string& channel,
SubscriptionCallback cb)
{
// Create and send a subscribe pdu
Json::Value body;
body["channel"] = channel;
Json::Value pdu;
pdu["action"] = "rtm/subscribe";
pdu["body"] = body;
_webSocket.send(pdu.toStyledString());
// Set the callback
std::lock_guard<std::mutex> lock(_cbsMutex);
_cbs[channel] = cb;
}
void SatoriConnection::unsubscribe(const std::string& channel)
{
{
std::lock_guard<std::mutex> lock(_cbsMutex);
auto cb = _cbs.find(channel);
if (cb == _cbs.end()) return;
_cbs.erase(cb);
}
// Create and send an unsubscribe pdu
Json::Value body;
body["channel"] = channel;
Json::Value pdu;
pdu["action"] = "rtm/unsubscribe";
pdu["body"] = body;
_webSocket.send(pdu.toStyledString());
}
//
// Enqueue strategy drops old messages when we are at full capacity
//
// If we want to keep only 3 items max in the queue:
//
// enqueue(A) -> [A]
// enqueue(B) -> [B, A]
// enqueue(C) -> [C, B, A]
// enqueue(D) -> [D, C, B] -- now we drop A, the oldest message,
// -- and keep the 'fresh ones'
//
void SatoriConnection::enqueue(const std::string& msg)
{
std::lock_guard<std::mutex> lock(_queueMutex);
if (_messageQueue.size() == SatoriConnection::kQueueMaxSize)
{
_messageQueue.pop_back();
}
_messageQueue.push_front(msg);
}
//
// We process messages back (oldest) to front (newest) to respect ordering
// when sending them. If we fail to send something, we put it back in the queue
// at the end we picked it up originally (at the end).
//
bool SatoriConnection::flushQueue()
{
std::lock_guard<std::mutex> lock(_queueMutex);
while (!_messageQueue.empty())
{
auto&& msg = _messageQueue.back();
if (!publishMessage(msg))
{
_messageQueue.push_back(msg);
return false;
}
_messageQueue.pop_back();
}
return true;
}
bool SatoriConnection::publishMessage(const std::string& serializedJson)
{
SatoriConnection::invokeTrafficTrackerCallback(serializedJson.size(), false);
return _webSocket.send(serializedJson);
}
} // namespace ix

130
examples/satori_publisher/IXSatoriConnection.h Normal file
View File

@ -0,0 +1,130 @@
/*
* IXSatoriConnection.h
* Author: Benjamin Sergeant
* Copyright (c) 2017-2018 Machine Zone. All rights reserved.
*/
#pragma once
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <unordered_map>
#include "jsoncpp/json/json.h"
#include <ixwebsocket/IXWebSocket.h>
namespace ix
{
using SubscriptionCallback = std::function<void(const Json::Value&)>;
using AuthenticatedCallback = std::function<void()>;
using OnTrafficTrackerCallback = std::function<void(size_t size, bool incoming)>;
class SatoriConnection
{
public:
SatoriConnection();
~SatoriConnection();
/// Configuration / set keys, etc...
/// All input data but the channel name is encrypted with rc4
void configure(const std::string& appkey,
const std::string& endpoint,
const std::string& rolename,
const std::string& rolesecret);
/// Set the traffic tracker callback
static void setTrafficTrackerCallback(const OnTrafficTrackerCallback& callback);
/// Reset the traffic tracker callback to an no-op one.
static void resetTrafficTrackerCallback();
/// Reset the traffic tracker callback to an no-op one.
void setAuthenticatedCallback(const AuthenticatedCallback& authenticatedCallback);
/// Start the worker thread, used for background publishing
void start();
/// Publish a message to a channel
///
/// No-op if the connection is not established
bool publish(const std::string& channel,
const Json::Value& msg);
// Subscribe to a channel, and execute a callback when an incoming
// message arrives.
void subscribe(const std::string& channel, SubscriptionCallback cb);
/// Unsubscribe from a channel
void unsubscribe(const std::string& channel);
/// Close the RTM connection and free the RTM handle memory
void disconnect();
/// Connect to Satori and authenticate the connection
bool connect();
/// Returns true only if we're connected
bool isConnected() const;
void logError(const std::string& error);
private:
bool sendHandshakeMessage();
bool handleHandshakeResponse(const Json::Value& data);
bool sendAuthMessage(const std::string& nonce);
bool handleSubscriptionData(const Json::Value& pdu);
bool publishMessage(const std::string& serializedJson);
bool flushQueue();
void enqueue(const std::string& msg);
/// Invoke the traffic tracker callback
static void invokeTrafficTrackerCallback(size_t size, bool incoming);
/// Invoke the authenticated callback
void invokeAuthenticatedCallback();
///
/// Member variables
///
WebSocket _webSocket;
/// Configuration data
std::string _appkey;
std::string _endpoint;
std::string _role_name;
std::string _role_secret;
uint32_t _history;
// Can be set on control+background thread, protecting with an atomic
std::atomic<bool> _authenticated;
// Keep some objects around
Json::Value _body;
Json::Value _pdu;
/// Traffic tracker callback
static OnTrafficTrackerCallback _onTrafficTrackerCallback;
/// Callback invoked when we are authenticated
AuthenticatedCallback _authenticatedCallback;
/// Subscription callbacks, only one per channel
std::unordered_map<std::string, SubscriptionCallback> _cbs;
mutable std::mutex _cbsMutex;
// Message Queue can be touched on control+background thread,
// protecting with a mutex.
//
// 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;
// Cap the queue size (100 elems so far -> ~100k)
static constexpr size_t kQueueMaxSize = 100;
};
} // namespace ix

6
examples/satori_publisher/README.md Normal file
View File

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

89
examples/satori_publisher/base64.cpp Normal file
View File

@ -0,0 +1,89 @@
/*
base64.cpp and base64.h
Copyright (C) 2004-2008 René Nyffenegger
This source code is provided 'as-is', without any express or implied
warranty. In no event will the author be held liable for any damages
arising from the use of this software.
Permission is granted to anyone to use this software for any purpose,
including commercial applications, and to alter it and redistribute it
freely, subject to the following restrictions:
1. The origin of this source code must not be misrepresented; you must not
claim that you wrote the original source code. If you use this source code
in a product, an acknowledgment in the product documentation would be
appreciated but is not required.
2. Altered source versions must be plainly marked as such, and must not be
misrepresented as being the original source code.
3. This notice may not be removed or altered from any source distribution.
René Nyffenegger rene.nyffenegger@adp-gmbh.ch
*/
#include "base64.h"
namespace ix
{
static const std::string base64_chars =
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789+/";
static inline bool is_base64(unsigned char c)
{
return (isalnum(c) || (c == '+') || (c == '/'));
}
std::string base64_encode(const std::string& data, uint32_t len)
{
std::string ret;
int i = 0;
int j = 0;
unsigned char char_array_3[3];
unsigned char char_array_4[4];
const char* bytes_to_encode = data.c_str();
while(len--)
{
char_array_3[i++] = *(bytes_to_encode++);
if(i == 3)
{
char_array_4[0] = (char_array_3[0] & 0xfc) >> 2;
char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
char_array_4[3] = char_array_3[2] & 0x3f;
for(i = 0; (i <4) ; i++)
ret += base64_chars[char_array_4[i]];
i = 0;
}
}
if(i)
{
for(j = i; j < 3; j++)
char_array_3[j] = '\0';
char_array_4[0] = (char_array_3[0] & 0xfc) >> 2;
char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
char_array_4[3] = char_array_3[2] & 0x3f;
for(j = 0; (j < i + 1); j++)
ret += base64_chars[char_array_4[j]];
while((i++ < 3))
ret += '=';
}
return ret;
}
}

14
examples/satori_publisher/base64.h Normal file
View File

@ -0,0 +1,14 @@
/*
* base64.h
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone. All rights reserved.
*/
#pragma once
#include <string>
namespace ix
{
std::string base64_encode(const std::string& data, uint32_t len);
}

44
examples/satori_publisher/devnull_server.js Normal file
View File

@ -0,0 +1,44 @@
/*
* devnull_server.js
* Author: Benjamin Sergeant
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 5678, perMessageDeflate: false });
let handshake = false
let authenticated = false
wss.on('connection', function connection(ws) {
ws.on('message', function incoming(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)
}
});
});

43
examples/satori_publisher/devnull_server.py Normal file
View File

@ -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()

333
examples/satori_publisher/jsoncpp/json/json-forwards.h Normal file
View File

@ -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

2186
examples/satori_publisher/jsoncpp/json/json.h Normal file
View File

File diff suppressed because it is too large Load Diff

5386
examples/satori_publisher/jsoncpp/jsoncpp.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

19
examples/satori_publisher/package-lock.json generated Normal file
View File

@ -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"
}
}
}
}

90
examples/satori_publisher/satori_publisher.cpp Normal file
View File

@ -0,0 +1,90 @@
/*
* satori_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 "IXSatoriConnection.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[])
{
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::SatoriConnection::setTrafficTrackerCallback(
[&incomingBytes, &outgoingBytes](size_t size, bool incoming)
{
if (incoming)
{
incomingBytes += size;
}
else
{
outgoingBytes += size;
}
}
);
bool done = false;
ix::SatoriConnection satoriConnection;
satoriConnection.configure(appkey, endpoint, rolename, rolesecret);
satoriConnection.connect();
satoriConnection.setAuthenticatedCallback(
[&satoriConnection, channel, path, &done]()
{
std::cout << "Authenticated" << std::endl;;
std::string line;
std::ifstream f(path);
if (!f.is_open())
{
std::cerr << "error while opening file" << std::endl;
}
while (getline(f, line))
{
Json::Value value;
Json::Reader reader;
reader.parse(line, value);
satoriConnection.publish(channel, value);
}
if (f.bad())
{
std::cerr << "error while reading file" << std::endl;
}
done = true;
}
);
while (!done)
{
msleep(1000);
}
std::cout << incomingBytes << std::endl;
std::cout << "Incoming bytes: " << incomingBytes << std::endl;
std::cout << "Outgoing bytes: " << outgoingBytes << std::endl;
return 0;
}

11
examples/satori_publisher/satori_publisher.sh Normal file
View File

@ -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"
path=events.jsonl
build/satori_publisher $endpoint $appkey $channel $rolename $rolesecret $path