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

reformat everything with clang-format

Browse Source
This commit is contained in:
Benjamin Sergeant
2019-09-23 10:25:23 -07:00
parent 398c4fbf99
commit cd3c9d879c
92 changed files with 3158 additions and 3348 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
ws/IXRedisClient.cpp
View File

@ -5,14 +5,14 @@
*/
#include "IXRedisClient.h"
#include <ixwebsocket/IXSocketFactory.h>
#include <ixwebsocket/IXSocket.h>
#include <iostream>
#include <sstream>
#include <iomanip>
#include <vector>
#include <cstring>
#include <iomanip>
#include <iostream>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXSocketFactory.h>
#include <sstream>
#include <vector>
namespace ix
{
@ -31,8 +31,7 @@ namespace ix
return _socket->connect(hostname, port, errMsg, nullptr);
}
bool RedisClient::auth(const std::string& password,
std::string& response)
bool RedisClient::auth(const std::string& password, std::string& response)
{
response.clear();
@ -203,7 +202,7 @@ namespace ix
int arraySize;
{
std::stringstream ss;
ss << line.substr(1, line.size()-1);
ss << line.substr(1, line.size() - 1);
ss >> arraySize;
}
@ -220,7 +219,7 @@ namespace ix
// => $7 (7 bytes)
int stringSize;
std::stringstream ss;
ss << line.substr(1, line.size()-1);
ss << line.substr(1, line.size() - 1);
ss >> stringSize;
auto readResult = _socket->readBytes(stringSize, nullptr, nullptr);
@ -246,4 +245,4 @@ namespace ix
{
_stop = true;
}
}
} // namespace ix

7
ws/IXRedisClient.h
View File

@ -6,9 +6,9 @@
#pragma once
#include <atomic>
#include <functional>
#include <memory>
#include <atomic>
namespace ix
{
@ -20,7 +20,10 @@ namespace ix
using OnRedisSubscribeResponseCallback = std::function<void(const std::string&)>;
using OnRedisSubscribeCallback = std::function<void(const std::string&)>;
RedisClient() : _stop(false) {}
RedisClient()
: _stop(false)
{
}
~RedisClient() = default;
bool connect(const std::string& hostname, int port);

19
ws/IXSentryClient.cpp
View File

@ -8,17 +8,16 @@
#include <chrono>
#include <iostream>
#include <spdlog/spdlog.h>
#include <ixwebsocket/IXWebSocketHttpHeaders.h>
#include <spdlog/spdlog.h>
namespace ix
{
SentryClient::SentryClient(const std::string& dsn) :
_dsn(dsn),
_validDsn(false),
_luaFrameRegex("\t([^/]+):([0-9]+): in function '([^/]+)'")
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;
@ -168,8 +167,7 @@ namespace ix
return _jsonWriter.write(payload);
}
std::pair<HttpResponsePtr, std::string> SentryClient::send(const Json::Value& msg,
bool verbose)
std::pair<HttpResponsePtr, std::string> SentryClient::send(const Json::Value& msg, bool verbose)
{
auto args = _httpClient.createRequest();
args->extraHeaders["X-Sentry-Auth"] = SentryClient::computeAuthHeader();
@ -177,10 +175,7 @@ namespace ix
args->transferTimeout = 5 * 60;
args->followRedirects = true;
args->verbose = verbose;
args->logger = [](const std::string& msg)
{
spdlog::info("request logger: {}", msg);
};
args->logger = [](const std::string& msg) { spdlog::info("request logger: {}", msg); };
std::string body = computePayload(msg);
HttpResponsePtr response = _httpClient.post(_url, body, args);

14
ws/snake/IXAppConfig.cpp
View File

@ -4,24 +4,20 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include "IXSnakeProtocol.h"
#include "IXAppConfig.h"
#include "IXSnakeProtocol.h"
#include <iostream>
#include <ixcrypto/IXUuid.h>
namespace snake
{
bool isAppKeyValid(
const AppConfig& appConfig,
std::string appkey)
bool isAppKeyValid(const AppConfig& appConfig, std::string appkey)
{
return appConfig.apps.count(appkey) != 0;
}
std::string getRoleSecret(
const AppConfig& appConfig,
std::string appkey,
std::string role)
std::string getRoleSecret(const AppConfig& appConfig, std::string appkey, std::string role)
{
if (!isAppKeyValid(appConfig, appkey))
{
@ -49,4 +45,4 @@ namespace snake
std::cout << "redis password: " << appConfig.redisPassword << std::endl;
std::cout << "redis port: " << appConfig.redisPort << std::endl;
}
}
} // namespace snake

35
ws/snake/IXSnakeConnectionState.h
View File

@ -16,16 +16,37 @@ namespace snake
class SnakeConnectionState : public ix::ConnectionState
{
public:
std::string getNonce() { return _nonce; }
void setNonce(const std::string& nonce) { _nonce = nonce; }
std::string getNonce()
{
return _nonce;
}
void setNonce(const std::string& nonce)
{
_nonce = nonce;
}
std::string appkey() { return _appkey; }
void setAppkey(const std::string& appkey) { _appkey = appkey; }
std::string appkey()
{
return _appkey;
}
void setAppkey(const std::string& appkey)
{
_appkey = appkey;
}
std::string role() { return _role; }
void setRole(const std::string& role) { _role = role; }
std::string role()
{
return _role;
}
void setRole(const std::string& role)
{
_role = role;
}
ix::RedisClient& redisClient() { return _redisClient; }
ix::RedisClient& redisClient()
{
return _redisClient;
}
std::future<void> fut;

166
ws/snake/IXSnakeProtocol.cpp
View File

@ -6,41 +6,32 @@
#include "IXSnakeProtocol.h"
#include <ixwebsocket/IXWebSocket.h>
#include <ixcrypto/IXHMac.h>
#include "IXSnakeConnectionState.h"
#include "IXAppConfig.h"
#include "IXSnakeConnectionState.h"
#include "nlohmann/json.hpp"
#include <sstream>
#include <iostream>
#include <ixcrypto/IXHMac.h>
#include <ixwebsocket/IXWebSocket.h>
#include <sstream>
namespace snake
{
void handleError(
const std::string& action,
std::shared_ptr<ix::WebSocket> ws,
nlohmann::json pdu,
const std::string& errMsg)
void handleError(const std::string& action,
std::shared_ptr<ix::WebSocket> ws,
nlohmann::json pdu,
const std::string& errMsg)
{
std::string actionError(action);
actionError += "/error";
nlohmann::json response = {
{"action", actionError},
{"id", pdu.value("id", 1)},
{"body", {
{"reason", errMsg}
}}
};
{"action", actionError}, {"id", pdu.value("id", 1)}, {"body", {{"reason", errMsg}}}};
ws->sendText(response.dump());
}
void handleHandshake(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
void handleHandshake(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
{
std::string role = pdu["body"]["data"]["role"];
@ -50,13 +41,10 @@ namespace snake
nlohmann::json response = {
{"action", "auth/handshake/ok"},
{"id", pdu.value("id", 1)},
{"body", {
{"data", {
{"nonce", state->getNonce()},
{"connection_id", state->getId()}
}},
}}
};
{"body",
{
{"data", {{"nonce", state->getNonce()}, {"connection_id", state->getId()}}},
}}};
auto serializedResponse = response.dump();
std::cout << "response = " << serializedResponse << std::endl;
@ -64,11 +52,10 @@ namespace snake
ws->sendText(serializedResponse);
}
void handleAuth(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
void handleAuth(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
{
auto secret = getRoleSecret(appConfig, state->appkey(), state->role());
std::cout << "secret = " << secret << std::endl;
@ -78,11 +65,7 @@ namespace snake
nlohmann::json response = {
{"action", "auth/authenticate/error"},
{"id", pdu.value("id", 1)},
{"body", {
{"error", "authentication_failed"},
{"reason", "invalid secret"}
}}
};
{"body", {{"error", "authentication_failed"}, {"reason", "invalid secret"}}}};
ws->sendText(response.dump());
return;
}
@ -102,28 +85,20 @@ namespace snake
nlohmann::json response = {
{"action", "auth/authenticate/error"},
{"id", pdu.value("id", 1)},
{"body", {
{"error", "authentication_failed"},
{"reason", "invalid hash"}
}}
};
{"body", {{"error", "authentication_failed"}, {"reason", "invalid hash"}}}};
ws->sendText(response.dump());
return;
}
nlohmann::json response = {
{"action", "auth/authenticate/ok"},
{"id", pdu.value("id", 1)},
{"body", {}}
};
{"action", "auth/authenticate/ok"}, {"id", pdu.value("id", 1)}, {"body", {}}};
ws->sendText(response.dump());
}
void handlePublish(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
void handlePublish(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
{
std::vector<std::string> channels;
@ -150,9 +125,7 @@ namespace snake
for (auto&& channel : channels)
{
std::stringstream ss;
ss << state->appkey()
<< "::"
<< channel;
ss << state->appkey() << "::" << channel;
std::string errMsg;
if (!state->redisClient().publish(ss.str(), pdu.dump(), errMsg))
@ -165,10 +138,7 @@ namespace snake
}
nlohmann::json response = {
{"action", "rtm/publish/ok"},
{"id", pdu.value("id", 1)},
{"body", {}}
};
{"action", "rtm/publish/ok"}, {"id", pdu.value("id", 1)}, {"body", {}}};
ws->sendText(response.dump());
}
@ -176,19 +146,16 @@ namespace snake
//
// FIXME: this is not cancellable. We should be able to cancel the redis subscription
//
void handleRedisSubscription(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
void handleRedisSubscription(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
{
std::string channel = pdu["body"]["channel"];
std::string subscriptionId = channel;
std::stringstream ss;
ss << state->appkey()
<< "::"
<< channel;
ss << state->appkey() << "::" << channel;
std::string appChannel(ss.str());
@ -224,8 +191,7 @@ namespace snake
}
int id = 0;
auto callback = [ws, &id, &subscriptionId](const std::string& messageStr)
{
auto callback = [ws, &id, &subscriptionId](const std::string& messageStr) {
auto msg = nlohmann::json::parse(messageStr);
msg = msg["body"]["message"];
@ -233,27 +199,18 @@ namespace snake
nlohmann::json response = {
{"action", "rtm/subscription/data"},
{"id", id++},
{"body", {
{"subscription_id", subscriptionId},
{"messages", {msg}}
}}
};
{"body", {{"subscription_id", subscriptionId}, {"messages", {msg}}}}};
ws->sendText(response.dump());
};
auto responseCallback = [ws, pdu, &subscriptionId](const std::string& redisResponse)
{
auto responseCallback = [ws, pdu, &subscriptionId](const std::string& redisResponse) {
std::cout << "Redis subscribe response: " << redisResponse << std::endl;
// Success
nlohmann::json response = {
{"action", "rtm/subscribe/ok"},
{"id", pdu.value("id", 1)},
{"body", {
{"subscription_id", subscriptionId}
}}
};
nlohmann::json response = {{"action", "rtm/subscribe/ok"},
{"id", pdu.value("id", 1)},
{"body", {{"subscription_id", subscriptionId}}}};
ws->sendText(response.dump());
};
@ -267,24 +224,18 @@ namespace snake
}
}
void handleSubscribe(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
void handleSubscribe(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const nlohmann::json& pdu)
{
state->fut = std::async(std::launch::async,
handleRedisSubscription,
state,
ws,
appConfig,
pdu);
state->fut =
std::async(std::launch::async, handleRedisSubscription, state, ws, appConfig, pdu);
}
void handleUnSubscribe(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
void handleUnSubscribe(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const nlohmann::json& pdu)
{
// extract subscription_id
auto body = pdu["body"];
@ -292,21 +243,16 @@ namespace snake
state->redisClient().stop();
nlohmann::json response = {
{"action", "rtm/unsubscribe/ok"},
{"id", pdu.value("id", 1)},
{"body", {
{"subscription_id", subscriptionId}
}}
};
nlohmann::json response = {{"action", "rtm/unsubscribe/ok"},
{"id", pdu.value("id", 1)},
{"body", {{"subscription_id", subscriptionId}}}};
ws->sendText(response.dump());
}
void processCobraMessage(
std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const std::string& str)
void processCobraMessage(std::shared_ptr<SnakeConnectionState> state,
std::shared_ptr<ix::WebSocket> ws,
const AppConfig& appConfig,
const std::string& str)
{
auto pdu = nlohmann::json::parse(str);
std::cout << "Got " << str << std::endl;
@ -339,4 +285,4 @@ namespace snake
std::cerr << "Unhandled action: " << action << std::endl;
}
}
}
} // namespace snake

43
ws/snake/IXSnakeServer.cpp
View File

@ -5,18 +5,18 @@
*/
#include "IXSnakeServer.h"
#include "IXSnakeProtocol.h"
#include "IXSnakeConnectionState.h"
#include "IXAppConfig.h"
#include "IXAppConfig.h"
#include "IXSnakeConnectionState.h"
#include "IXSnakeProtocol.h"
#include <iostream>
#include <sstream>
namespace snake
{
SnakeServer::SnakeServer(const AppConfig& appConfig) :
_appConfig(appConfig),
_server(appConfig.port, appConfig.hostname)
SnakeServer::SnakeServer(const AppConfig& appConfig)
: _appConfig(appConfig)
, _server(appConfig.port, appConfig.hostname)
{
;
}
@ -32,7 +32,7 @@ namespace snake
idx = path.rfind('=');
if (idx != std::string::npos)
{
std::string appkey = path.substr(idx+1);
std::string appkey = path.substr(idx + 1);
return appkey;
}
else
@ -45,21 +45,18 @@ namespace snake
{
std::cout << "Listening on " << _appConfig.hostname << ":" << _appConfig.port << std::endl;
auto factory = []() -> std::shared_ptr<ix::ConnectionState>
{
auto factory = []() -> std::shared_ptr<ix::ConnectionState> {
return std::make_shared<SnakeConnectionState>();
};
_server.setConnectionStateFactory(factory);
_server.setOnConnectionCallback(
[this](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ix::ConnectionState> connectionState)
{
std::shared_ptr<ix::ConnectionState> connectionState) {
auto state = std::dynamic_pointer_cast<SnakeConnectionState>(connectionState);
webSocket->setOnMessageCallback(
[this, webSocket, state](const ix::WebSocketMessagePtr& msg)
{
[this, webSocket, state](const ix::WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
std::cerr << "New connection" << std::endl;
@ -84,16 +81,16 @@ namespace snake
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason << std::endl;
<< " code " << msg->closeInfo.code << " reason "
<< msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
@ -105,10 +102,8 @@ namespace snake
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
processCobraMessage(state, webSocket, _appConfig, msg->str);
}
}
);
}
);
});
});
auto res = _server.listen();
if (!res.first)
@ -133,4 +128,4 @@ namespace snake
{
_server.stop();
}
}
} // namespace snake

112
ws/ws.cpp
View File

@ -9,27 +9,22 @@
//
#include "ws.h"
#include <string>
#include <sstream>
#include <iostream>
#include <fstream>
#include <cli11/CLI11.hpp>
#include <spdlog/spdlog.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXNetSystem.h>
#include <fstream>
#include <iostream>
#include <ixcore/utils/IXCoreLogger.h>
#include <ixwebsocket/IXNetSystem.h>
#include <ixwebsocket/IXSocket.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <string>
int main(int argc, char** argv)
{
ix::initNetSystem();
ix::IXCoreLogger::LogFunc logFunc = [](const char* msg)
{
spdlog::info(msg);
};
ix::IXCoreLogger::LogFunc logFunc = [](const char* msg) { spdlog::info(msg); };
ix::IXCoreLogger::setLogFunction(logFunc);
// Display command.
@ -43,7 +38,7 @@ int main(int argc, char** argv)
std::cout << std::endl;
}
CLI::App app{"ws is a websocket tool"};
CLI::App app {"ws is a websocket tool"};
app.require_subcommand();
std::string url("ws://127.0.0.1:8008");
@ -94,13 +89,16 @@ int main(int argc, char** argv)
CLI::App* sendApp = app.add_subcommand("send", "Send a file");
sendApp->add_option("url", url, "Connection url")->required();
sendApp->add_option("path", path, "Path to the file to send")
->required()->check(CLI::ExistingPath);
->required()
->check(CLI::ExistingPath);
sendApp->add_option("--pidfile", pidfile, "Pid file");
CLI::App* receiveApp = app.add_subcommand("receive", "Receive a file");
receiveApp->add_option("url", url, "Connection url")->required();
receiveApp->add_option("--delay", delayMs, "Delay (ms) to wait after receiving a fragment"
" to artificially slow down the receiver");
receiveApp->add_option("--delay",
delayMs,
"Delay (ms) to wait after receiving a fragment"
" to artificially slow down the receiver");
receiveApp->add_option("--pidfile", pidfile, "Pid file");
CLI::App* transferApp = app.add_subcommand("transfer", "Broadcasting server");
@ -114,7 +112,9 @@ int main(int argc, char** argv)
connectApp->add_flag("-d", disableAutomaticReconnection, "Disable Automatic Reconnection");
connectApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
connectApp->add_flag("-b", binaryMode, "Send in binary mode");
connectApp->add_option("--max_wait", maxWaitBetweenReconnectionRetries, "Max Wait Time between reconnection retries");
connectApp->add_option("--max_wait",
maxWaitBetweenReconnectionRetries,
"Max Wait Time between reconnection retries");
CLI::App* chatApp = app.add_subcommand("chat", "Group chat");
chatApp->add_option("url", url, "Connection url")->required();
@ -181,9 +181,11 @@ int main(int argc, char** argv)
cobraPublish->add_option("channel", channel, "Channel")->required();
cobraPublish->add_option("--pidfile", pidfile, "Pid file");
cobraPublish->add_option("path", path, "Path to the file to send")
->required()->check(CLI::ExistingPath);
->required()
->check(CLI::ExistingPath);
CLI::App* cobraMetricsPublish = app.add_subcommand("cobra_metrics_publish", "Cobra metrics publisher");
CLI::App* cobraMetricsPublish =
app.add_subcommand("cobra_metrics_publish", "Cobra metrics publisher");
cobraMetricsPublish->add_option("--appkey", appkey, "Appkey");
cobraMetricsPublish->add_option("--endpoint", endpoint, "Endpoint");
cobraMetricsPublish->add_option("--rolename", rolename, "Role name");
@ -191,7 +193,8 @@ int main(int argc, char** argv)
cobraMetricsPublish->add_option("channel", channel, "Channel")->required();
cobraMetricsPublish->add_option("--pidfile", pidfile, "Pid file");
cobraMetricsPublish->add_option("path", path, "Path to the file to send")
->required()->check(CLI::ExistingPath);
->required()
->check(CLI::ExistingPath);
cobraMetricsPublish->add_flag("--stress", stress, "Stress mode");
CLI::App* cobra2statsd = app.add_subcommand("cobra_to_statsd", "Cobra to statsd");
@ -269,8 +272,11 @@ int main(int argc, char** argv)
}
else if (app.got_subcommand("connect"))
{
ret = ix::ws_connect_main(url, headers, disableAutomaticReconnection,
disablePerMessageDeflate, binaryMode,
ret = ix::ws_connect_main(url,
headers,
disableAutomaticReconnection,
disablePerMessageDeflate,
binaryMode,
maxWaitBetweenReconnectionRetries);
}
else if (app.got_subcommand("chat"))
@ -291,15 +297,22 @@ int main(int argc, char** argv)
}
else if (app.got_subcommand("curl"))
{
ret = ix::ws_http_client_main(url, headers, data, headersOnly,
connectTimeOut, transferTimeout,
followRedirects, maxRedirects, verbose,
save, output, compress);
ret = ix::ws_http_client_main(url,
headers,
data,
headersOnly,
connectTimeOut,
transferTimeout,
followRedirects,
maxRedirects,
verbose,
save,
output,
compress);
}
else if (app.got_subcommand("redis_publish"))
{
ret = ix::ws_redis_publish_main(hostname, redisPort, password,
channel, message, count);
ret = ix::ws_redis_publish_main(hostname, redisPort, password, channel, message, count);
}
else if (app.got_subcommand("redis_subscribe"))
{
@ -307,42 +320,41 @@ int main(int argc, char** argv)
}
else if (app.got_subcommand("cobra_subscribe"))
{
ret = ix::ws_cobra_subscribe_main(appkey, endpoint,
rolename, rolesecret,
channel, filter, quiet);
ret = ix::ws_cobra_subscribe_main(
appkey, endpoint, rolename, rolesecret, channel, filter, quiet);
}
else if (app.got_subcommand("cobra_publish"))
{
ret = ix::ws_cobra_publish_main(appkey, endpoint,
rolename, rolesecret,
channel, path);
ret = ix::ws_cobra_publish_main(appkey, endpoint, rolename, rolesecret, channel, path);
}
else if (app.got_subcommand("cobra_metrics_publish"))
{
ret = ix::ws_cobra_metrics_publish_main(appkey, endpoint,
rolename, rolesecret,
channel, path, stress);
ret = ix::ws_cobra_metrics_publish_main(
appkey, endpoint, rolename, rolesecret, channel, path, stress);
}
else if (app.got_subcommand("cobra_to_statsd"))
{
ret = ix::ws_cobra_to_statsd_main(appkey, endpoint,
rolename, rolesecret,
channel, filter, hostname, statsdPort,
prefix, fields, verbose);
ret = ix::ws_cobra_to_statsd_main(appkey,
endpoint,
rolename,
rolesecret,
channel,
filter,
hostname,
statsdPort,
prefix,
fields,
verbose);
}
else if (app.got_subcommand("cobra_to_sentry"))
{
ret = ix::ws_cobra_to_sentry_main(appkey, endpoint,
rolename, rolesecret,
channel, filter, dsn,
verbose, strict, jobs);
ret = ix::ws_cobra_to_sentry_main(
appkey, endpoint, rolename, rolesecret, channel, filter, dsn, verbose, strict, jobs);
}
else if (app.got_subcommand("snake"))
{
ret = ix::ws_snake_main(port, hostname,
redisHosts, redisPort,
redisPassword, verbose,
appsConfigPath);
ret = ix::ws_snake_main(
port, hostname, redisHosts, redisPort, redisPassword, verbose, appsConfigPath);
}
else if (app.got_subcommand("httpd"))
{

233
ws/ws_autobahn.cpp
View File

@ -21,7 +21,8 @@
//
//
// 2 Run the test server (using docker)
// docker run -it --rm -v "${PWD}/config:/config" -v "${PWD}/reports:/reports" -p 9001:9001 --name fuzzingserver crossbario/autobahn-testsuite
// docker run -it --rm -v "${PWD}/config:/config" -v "${PWD}/reports:/reports" -p 9001:9001 --name
// fuzzingserver crossbario/autobahn-testsuite
//
// 3. Run this command
// ws autobahn -q --url ws://localhost:9001
@ -29,15 +30,14 @@
// 4. A HTML report will be generated, you can inspect it to see if you are compliant or not
//
#include <iostream>
#include <sstream>
#include <atomic>
#include <mutex>
#include <condition_variable>
#include <ixwebsocket/IXWebSocket.h>
#include <iostream>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <mutex>
#include <spdlog/spdlog.h>
#include <sstream>
namespace
@ -53,31 +53,31 @@ namespace
return str.substr(0, n) + "...";
}
}
}
} // namespace
namespace ix
{
class AutobahnTestCase
{
public:
AutobahnTestCase(const std::string& _url, bool quiet);
void run();
public:
AutobahnTestCase(const std::string& _url, bool quiet);
void run();
private:
void log(const std::string& msg);
private:
void log(const std::string& msg);
std::string _url;
ix::WebSocket _webSocket;
std::string _url;
ix::WebSocket _webSocket;
bool _quiet;
bool _quiet;
std::mutex _mutex;
std::condition_variable _condition;
std::mutex _mutex;
std::condition_variable _condition;
};
AutobahnTestCase::AutobahnTestCase(const std::string& url, bool quiet) :
_url(url),
_quiet(quiet)
AutobahnTestCase::AutobahnTestCase(const std::string& url, bool quiet)
: _url(url)
, _quiet(quiet)
{
_webSocket.disableAutomaticReconnection();
@ -102,67 +102,63 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const ix::WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) {
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
log("autobahn: connected");
ss << "Uri: " << msg->openInfo.uri << std::endl;
ss << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
log("autobahn: connected");
ss << "Uri: " << msg->openInfo.uri << std::endl;
ss << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
ss << it.first << ": " << it.second << std::endl;
}
ss << it.first << ": " << it.second << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "autobahn: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "autobahn: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "Received " << msg->wireSize << " bytes" << std::endl;
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "Received " << msg->wireSize << " bytes" << std::endl;
ss << "autobahn: received message: "
<< truncate(msg->str, 40)
<< std::endl;
ss << "autobahn: received message: " << truncate(msg->str, 40) << std::endl;
_webSocket.send(msg->str, msg->binary);
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
_webSocket.send(msg->str, msg->binary);
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
// And error can happen, in which case the test-case is marked done
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
ss << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
ss << "Received ping" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
ss << "Received pong" << std::endl;
}
else
{
ss << "Invalid ix::WebSocketMessageType" << std::endl;
}
// And error can happen, in which case the test-case is marked done
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
ss << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
ss << "Received ping" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
ss << "Received pong" << std::endl;
}
else
{
ss << "Invalid ix::WebSocketMessageType" << std::endl;
}
log(ss.str());
});
log(ss.str());
});
_webSocket.start();
@ -184,27 +180,24 @@ namespace ix
std::atomic<bool> success(true);
std::condition_variable condition;
webSocket.setOnMessageCallback(
[&condition, &success](const ix::WebSocketMessagePtr& msg)
webSocket.setOnMessageCallback([&condition, &success](const ix::WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Close)
{
if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Report generated" << std::endl;
condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str() << std::endl;
success = false;
}
std::cerr << "Report generated" << std::endl;
condition.notify_one();
}
);
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str() << std::endl;
success = false;
}
});
webSocket.start();
std::mutex mutex;
@ -231,33 +224,30 @@ namespace ix
int count = -1;
std::condition_variable condition;
webSocket.setOnMessageCallback(
[&condition, &count](const ix::WebSocketMessagePtr& msg)
webSocket.setOnMessageCallback([&condition, &count](const ix::WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Close)
{
if (msg->type == ix::WebSocketMessageType::Close)
{
condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str() << std::endl;
condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
// response is a string
std::stringstream ss;
ss << msg->str;
ss >> count;
}
condition.notify_one();
}
);
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str() << std::endl;
condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
// response is a string
std::stringstream ss;
ss << msg->str;
ss >> count;
}
});
webSocket.start();
std::mutex mutex;
@ -289,17 +279,14 @@ namespace ix
testCasesCount++;
for (int i = 1 ; i < testCasesCount; ++i)
for (int i = 1; i < testCasesCount; ++i)
{
spdlog::info("Execute test case {}", i);
int caseNumber = i;
std::stringstream ss;
ss << url
<< "/runCase?case="
<< caseNumber
<< "&agent=ixwebsocket";
ss << url << "/runCase?case=" << caseNumber << "&agent=ixwebsocket";
std::string url(ss.str());
@ -309,4 +296,4 @@ namespace ix
return generateReport(url) ? 0 : 1;
}
}
} // namespace ix

122
ws/ws_broadcast_server.cpp
View File

@ -5,8 +5,8 @@
*/
#include <iostream>
#include <sstream>
#include <ixwebsocket/IXWebSocketServer.h>
#include <sstream>
namespace ix
{
@ -16,75 +16,67 @@ namespace ix
ix::WebSocketServer server(port, hostname);
server.setOnConnectionCallback(
[&server](std::shared_ptr<WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState)
{
webSocket->setOnMessageCallback(
[webSocket, connectionState, &server](const WebSocketMessagePtr& msg)
server.setOnConnectionCallback([&server](std::shared_ptr<WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState) {
webSocket->setOnMessageCallback([webSocket, connectionState, &server](
const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
std::cerr << "New connection" << std::endl;
std::cerr << "id: " << connectionState->getId() << std::endl;
std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
std::cerr << "Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
if (msg->type == ix::WebSocketMessageType::Open)
std::cerr << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code << " reason "
<< msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
for (auto&& client : server.getClients())
{
if (client != webSocket)
{
std::cerr << "New connection" << std::endl;
std::cerr << "id: " << connectionState->getId() << std::endl;
std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
std::cerr << "Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
client->send(msg->str, msg->binary, [](int current, int total) -> bool {
std::cerr << "Step " << current << " out of " << total << std::endl;
return true;
});
do
{
std::cerr << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
size_t bufferedAmount = client->bufferedAmount();
std::cerr << bufferedAmount << " bytes left to be sent"
<< std::endl;
for (auto&& client : server.getClients())
{
if (client != webSocket)
{
client->send(msg->str,
msg->binary,
[](int current, int total) -> bool
{
std::cerr << "Step " << current
<< " out of " << total << std::endl;
return true;
});
do
{
size_t bufferedAmount = client->bufferedAmount();
std::cerr << bufferedAmount << " bytes left to be sent" << std::endl;
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
} while (client->bufferedAmount() != 0);
}
}
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
} while (client->bufferedAmount() != 0);
}
}
);
}
);
}
});
});
auto res = server.listen();
if (!res.first)
@ -98,4 +90,4 @@ namespace ix
return 0;
}
}
} // namespace ix

153
ws/ws_chat.cpp
View File

@ -9,13 +9,12 @@
// Broadcast server can be ran with `ws broadcast_server`
//
#include <iostream>
#include <sstream>
#include <queue>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include "nlohmann/json.hpp"
#include <iostream>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <queue>
#include <sstream>
// for convenience
using json = nlohmann::json;
@ -24,34 +23,32 @@ namespace ix
{
class WebSocketChat
{
public:
WebSocketChat(const std::string& url,
const std::string& user);
public:
WebSocketChat(const std::string& url, const std::string& user);
void subscribe(const std::string& channel);
void start();
void stop();
bool isReady() const;
void subscribe(const std::string& channel);
void start();
void stop();
bool isReady() const;
void sendMessage(const std::string& text);
size_t getReceivedMessagesCount() const;
void sendMessage(const std::string& text);
size_t getReceivedMessagesCount() const;
std::string encodeMessage(const std::string& text);
std::pair<std::string, std::string> decodeMessage(const std::string& str);
std::string encodeMessage(const std::string& text);
std::pair<std::string, std::string> decodeMessage(const std::string& str);
private:
std::string _url;
std::string _user;
ix::WebSocket _webSocket;
std::queue<std::string> _receivedQueue;
private:
std::string _url;
std::string _user;
ix::WebSocket _webSocket;
std::queue<std::string> _receivedQueue;
void log(const std::string& msg);
void log(const std::string& msg);
};
WebSocketChat::WebSocketChat(const std::string& url,
const std::string& user) :
_url(url),
_user(user)
WebSocketChat::WebSocketChat(const std::string& url, const std::string& user)
: _url(url)
, _user(user)
{
;
}
@ -83,64 +80,57 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const WebSocketMessagePtr& msg) {
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
log("ws chat: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
log("ws chat: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
std::cout << it.first << ": " << it.second << std::endl;
}
ss << "ws chat: user "
<< _user
<< " Connected !";
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws chat: user "
<< _user
<< " disconnected !"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
auto result = decodeMessage(msg->str);
ss << "ws chat: user " << _user << " Connected !";
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws chat: user " << _user << " disconnected !"
<< " code " << msg->closeInfo.code << " reason " << msg->closeInfo.reason;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
auto result = decodeMessage(msg->str);
// Our "chat" / "broacast" node.js server does not send us
// the messages we send, so we don't have to filter it out.
// Our "chat" / "broacast" node.js server does not send us
// the messages we send, so we don't have to filter it out.
// store text
_receivedQueue.push(result.second);
// store text
_receivedQueue.push(result.second);
ss << std::endl
<< result.first << "(" << msg->wireSize << " bytes)" << " > " << result.second
<< std::endl
<< _user << " > ";
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
ss << std::endl
<< result.first << "(" << msg->wireSize << " bytes)"
<< " > " << result.second << std::endl
<< _user << " > ";
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
@ -170,8 +160,7 @@ namespace ix
_webSocket.sendText(encodeMessage(text));
}
int ws_chat_main(const std::string& url,
const std::string& user)
int ws_chat_main(const std::string& url, const std::string& user)
{
std::cout << "Type Ctrl-D to exit prompt..." << std::endl;
WebSocketChat webSocketChat(url, user);
@ -196,4 +185,4 @@ namespace ix
return 0;
}
}
} // namespace ix

36
ws/ws_cobra_metrics_publish.cpp
View File

@ -4,15 +4,15 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <chrono>
#include <thread>
#include <atomic>
#include <jsoncpp/json/json.h>
#include <chrono>
#include <fstream>
#include <iostream>
#include <ixcobra/IXCobraMetricsPublisher.h>
#include <jsoncpp/json/json.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
namespace ix
{
@ -27,25 +27,23 @@ namespace ix
std::atomic<int> sentMessages(0);
std::atomic<int> ackedMessages(0);
CobraConnection::setPublishTrackerCallback(
[&sentMessages, &ackedMessages](bool sent, bool acked)
{
[&sentMessages, &ackedMessages](bool sent, bool acked) {
if (sent) sentMessages++;
if (acked) ackedMessages++;
}
);
});
CobraMetricsPublisher cobraMetricsPublisher;
cobraMetricsPublisher.enable(true);
bool enablePerMessageDeflate = true;
cobraMetricsPublisher.configure(appkey, endpoint, channel,
rolename, rolesecret, enablePerMessageDeflate);
cobraMetricsPublisher.configure(
appkey, endpoint, channel, rolename, rolesecret, enablePerMessageDeflate);
while (!cobraMetricsPublisher.isAuthenticated()) ;
while (!cobraMetricsPublisher.isAuthenticated())
;
std::ifstream f(path);
std::string str((std::istreambuf_iterator<char>(f)),
std::istreambuf_iterator<char>());
std::string str((std::istreambuf_iterator<char>(f)), std::istreambuf_iterator<char>());
Json::Value data;
Json::Reader reader;
@ -61,7 +59,7 @@ namespace ix
// Stress mode to try to trigger server and client bugs
while (true)
{
for (int i = 0 ; i < 1000; ++i)
for (int i = 0; i < 1000; ++i)
{
cobraMetricsPublisher.push(channel, data);
}
@ -70,7 +68,8 @@ namespace ix
cobraMetricsPublisher.resume();
// FIXME: investigate why without this check we trigger a lock
while (!cobraMetricsPublisher.isAuthenticated()) ;
while (!cobraMetricsPublisher.isAuthenticated())
;
}
}
@ -83,5 +82,4 @@ namespace ix
return 0;
}
}
} // namespace ix

115
ws/ws_cobra_publish.cpp
View File

@ -4,17 +4,17 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <chrono>
#include <thread>
#include <atomic>
#include <mutex>
#include <chrono>
#include <condition_variable>
#include <jsoncpp/json/json.h>
#include <fstream>
#include <iostream>
#include <ixcobra/IXCobraMetricsPublisher.h>
#include <jsoncpp/json/json.h>
#include <mutex>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
namespace ix
{
@ -26,8 +26,7 @@ namespace ix
const std::string& path)
{
std::ifstream f(path);
std::string str((std::istreambuf_iterator<char>(f)),
std::istreambuf_iterator<char>());
std::string str((std::istreambuf_iterator<char>(f)), std::istreambuf_iterator<char>());
Json::Value data;
Json::Reader reader;
@ -38,9 +37,8 @@ namespace ix
}
ix::CobraConnection conn;
conn.configure(appkey, endpoint,
rolename, rolesecret,
ix::WebSocketPerMessageDeflateOptions(true));
conn.configure(
appkey, endpoint, rolename, rolesecret, ix::WebSocketPerMessageDeflateOptions(true));
conn.connect();
// Display incoming messages
@ -48,59 +46,58 @@ namespace ix
std::atomic<bool> messageAcked(false);
std::condition_variable condition;
conn.setEventCallback(
[&conn, &channel, &data, &authenticated, &messageAcked, &condition]
(ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
conn.setEventCallback([&conn, &channel, &data, &authenticated, &messageAcked, &condition](
ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId) {
if (eventType == ix::CobraConnection_EventType_Open)
{
if (eventType == ix::CobraConnection_EventType_Open)
{
spdlog::info("Publisher connected");
spdlog::info("Publisher connected");
for (auto it : headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
else if (eventType == ix::CobraConnection_EventType_Authenticated)
for (auto it : headers)
{
spdlog::info("Publisher authenticated");
authenticated = true;
Json::Value channels;
channels[0] = channel;
auto msgId = conn.publish(channels, data);
spdlog::info("Published msg {}", msgId);
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
spdlog::info("Publisher: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Publisher: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Publisher: error {}", errMsg);
condition.notify_one();
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::info("Published message id {} acked", msgId);
messageAcked = true;
condition.notify_one();
spdlog::info("{}: {}", it.first, it.second);
}
}
);
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
spdlog::info("Publisher authenticated");
authenticated = true;
while (!authenticated) ;
while (!messageAcked) ;
Json::Value channels;
channels[0] = channel;
auto msgId = conn.publish(channels, data);
spdlog::info("Published msg {}", msgId);
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
spdlog::info("Publisher: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Publisher: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Publisher: error {}", errMsg);
condition.notify_one();
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::info("Published message id {} acked", msgId);
messageAcked = true;
condition.notify_one();
}
});
while (!authenticated)
;
while (!messageAcked)
;
return 0;
}
}
} // namespace ix

60
ws/ws_cobra_subscribe.cpp
View File

@ -4,14 +4,13 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <sstream>
#include <chrono>
#include <thread>
#include <atomic>
#include <chrono>
#include <iostream>
#include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
namespace ix
{
@ -24,9 +23,8 @@ namespace ix
bool quiet)
{
ix::CobraConnection conn;
conn.configure(appkey, endpoint,
rolename, rolesecret,
ix::WebSocketPerMessageDeflateOptions(true));
conn.configure(
appkey, endpoint, rolename, rolesecret, ix::WebSocketPerMessageDeflateOptions(true));
conn.connect();
Json::FastWriter jsonWriter;
@ -35,13 +33,11 @@ namespace ix
std::atomic<int> msgPerSeconds(0);
std::atomic<int> msgCount(0);
auto timer = [&msgPerSeconds, &msgCount]
{
auto timer = [&msgPerSeconds, &msgCount] {
while (true)
{
std::cout << "#messages " << msgCount << " "
<< "msg/s " << msgPerSeconds
<< std::endl;
<< "msg/s " << msgPerSeconds << std::endl;
msgPerSeconds = 0;
auto duration = std::chrono::seconds(1);
@ -52,13 +48,12 @@ namespace ix
std::thread t(timer);
conn.setEventCallback(
[&conn, &channel, &jsonWriter, &filter, &msgCount, &msgPerSeconds, &quiet]
(ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
{
[&conn, &channel, &jsonWriter, &filter, &msgCount, &msgPerSeconds, &quiet](
ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId) {
if (eventType == ix::CobraConnection_EventType_Open)
{
spdlog::info("Subscriber connected");
@ -71,18 +66,18 @@ namespace ix
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
spdlog::info("Subscriber authenticated");
conn.subscribe(channel, filter,
[&jsonWriter, &quiet,
&msgPerSeconds, &msgCount](const Json::Value& msg)
{
if (!quiet)
{
std::cout << jsonWriter.write(msg) << std::endl;
}
conn.subscribe(
channel,
filter,
[&jsonWriter, &quiet, &msgPerSeconds, &msgCount](const Json::Value& msg) {
if (!quiet)
{
std::cout << jsonWriter.write(msg) << std::endl;
}
msgPerSeconds++;
msgCount++;
});
msgPerSeconds++;
msgCount++;
});
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
@ -100,8 +95,7 @@ namespace ix
{
spdlog::error("Published message hacked: {}", msgId);
}
}
);
});
while (true)
{
@ -111,4 +105,4 @@ namespace ix
return 0;
}
}
} // namespace ix

192
ws/ws_cobra_to_sentry.cpp
View File

@ -4,19 +4,18 @@
* 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 <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include "IXSentryClient.h"
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <iostream>
#include <ixcobra/IXCobraConnection.h>
#include <mutex>
#include <queue>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
#include <vector>
namespace ix
{
@ -32,9 +31,8 @@ namespace ix
int jobs)
{
ix::CobraConnection conn;
conn.configure(appkey, endpoint,
rolename, rolesecret,
ix::WebSocketPerMessageDeflateOptions(true));
conn.configure(
appkey, endpoint, rolename, rolesecret, ix::WebSocketPerMessageDeflateOptions(true));
conn.connect();
Json::FastWriter jsonWriter;
@ -48,10 +46,15 @@ namespace ix
std::condition_variable progressCondition;
std::queue<Json::Value> queue;
auto sentrySender = [&condition, &progressCondition, &conditionVariableMutex,
&queue, verbose, &errorSending, &sentCount,
&stop, &dsn]
{
auto sentrySender = [&condition,
&progressCondition,
&conditionVariableMutex,
&queue,
verbose,
&errorSending,
&sentCount,
&stop,
&dsn] {
SentryClient sentryClient(dsn);
while (true)
@ -60,7 +63,7 @@ namespace ix
{
std::unique_lock<std::mutex> lock(conditionVariableMutex);
condition.wait(lock, [&queue, &stop]{ return !queue.empty() && !stop; });
condition.wait(lock, [&queue, &stop] { return !queue.empty() && !stop; });
msg = queue.front();
queue.pop();
@ -94,88 +97,93 @@ namespace ix
pool.push_back(std::thread(sentrySender));
}
conn.setEventCallback(
[&conn, &channel, &filter, &jsonWriter,
verbose, &receivedCount, &sentCount,
&condition, &conditionVariableMutex,
&progressCondition, &queue]
(ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
conn.setEventCallback([&conn,
&channel,
&filter,
&jsonWriter,
verbose,
&receivedCount,
&sentCount,
&condition,
&conditionVariableMutex,
&progressCondition,
&queue](ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId) {
if (eventType == ix::CobraConnection_EventType_Open)
{
if (eventType == ix::CobraConnection_EventType_Open)
{
spdlog::info("Subscriber connected");
spdlog::info("Subscriber connected");
for (auto it : headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
if (eventType == ix::CobraConnection_EventType_Closed)
for (auto it : headers)
{
spdlog::info("Subscriber closed");
spdlog::info("{}: {}", it.first, it.second);
}
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
std::cerr << "Subscriber authenticated" << std::endl;
conn.subscribe(channel, filter,
[&jsonWriter, verbose,
&sentCount, &receivedCount,
&condition, &conditionVariableMutex,
&progressCondition, &queue]
(const Json::Value& msg)
}
if (eventType == ix::CobraConnection_EventType_Closed)
{
spdlog::info("Subscriber closed");
}
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
std::cerr << "Subscriber authenticated" << std::endl;
conn.subscribe(channel,
filter,
[&jsonWriter,
verbose,
&sentCount,
&receivedCount,
&condition,
&conditionVariableMutex,
&progressCondition,
&queue](const Json::Value& msg) {
if (verbose)
{
if (verbose)
{
spdlog::info(jsonWriter.write(msg));
}
spdlog::info(jsonWriter.write(msg));
}
// If we cannot send to sentry fast enough, drop the message
const uint64_t scaleFactor = 2;
// If we cannot send to sentry fast enough, drop the message
const uint64_t scaleFactor = 2;
if (sentCount != 0 &&
receivedCount != 0 &&
(sentCount * scaleFactor < receivedCount))
{
spdlog::warn("message dropped: sending is backlogged !");
condition.notify_one();
progressCondition.notify_one();
return;
}
++receivedCount;
{
std::unique_lock<std::mutex> lock(conditionVariableMutex);
queue.push(msg);
}
if (sentCount != 0 && receivedCount != 0 &&
(sentCount * scaleFactor < receivedCount))
{
spdlog::warn("message dropped: sending is backlogged !");
condition.notify_one();
progressCondition.notify_one();
});
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
spdlog::info("Subscriber: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Subscriber: error {}", errMsg);
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::error("Published message hacked: {}", msgId);
}
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)
{
spdlog::info("Subscriber: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Subscriber: error {}", errMsg);
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::error("Published message hacked: {}", msgId);
}
});
std::mutex progressConditionVariableMutex;
while (true)
@ -200,4 +208,4 @@ namespace ix
return (strict && errorSending) ? 1 : 0;
}
}
} // namespace ix

146
ws/ws_cobra_to_statsd.cpp
View File

@ -4,15 +4,14 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <sstream>
#include <chrono>
#include <thread>
#include <atomic>
#include <vector>
#include <chrono>
#include <iostream>
#include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
#include <vector>
#ifndef _WIN32
#include <statsd_client.h>
@ -41,8 +40,7 @@ namespace ix
// 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)
std::string extractAttr(const std::string& attr, const Json::Value& jsonValue)
{
// Split by .
std::string token;
@ -71,9 +69,8 @@ namespace ix
bool verbose)
{
ix::CobraConnection conn;
conn.configure(appkey, endpoint,
rolename, rolesecret,
ix::WebSocketPerMessageDeflateOptions(true));
conn.configure(
appkey, endpoint, rolename, rolesecret, ix::WebSocketPerMessageDeflateOptions(true));
conn.connect();
auto tokens = parseFields(fields);
@ -90,72 +87,75 @@ namespace ix
Json::FastWriter jsonWriter;
uint64_t msgCount = 0;
conn.setEventCallback(
[&conn, &channel, &filter, &jsonWriter, &statsdClient, verbose, &tokens, &prefix, &msgCount]
(ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId)
conn.setEventCallback([&conn,
&channel,
&filter,
&jsonWriter,
&statsdClient,
verbose,
&tokens,
&prefix,
&msgCount](ix::CobraConnectionEventType eventType,
const std::string& errMsg,
const ix::WebSocketHttpHeaders& headers,
const std::string& subscriptionId,
CobraConnection::MsgId msgId) {
if (eventType == ix::CobraConnection_EventType_Open)
{
if (eventType == ix::CobraConnection_EventType_Open)
{
spdlog::info("Subscriber connected");
spdlog::info("Subscriber connected");
for (auto it : headers)
{
spdlog::info("{}: {}", it.first, it.second);
}
}
if (eventType == ix::CobraConnection_EventType_Closed)
for (auto it : headers)
{
spdlog::info("Subscriber closed");
}
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
spdlog::info("Subscriber authenticated");
conn.subscribe(channel, filter,
[&jsonWriter, &statsdClient,
verbose, &tokens, &prefix, &msgCount]
(const Json::Value& msg)
{
if (verbose)
{
spdlog::info(jsonWriter.write(msg));
}
std::string id;
for (auto&& attr : tokens)
{
id += ".";
id += extractAttr(attr, msg);
}
spdlog::info("{} {}{}", msgCount++, prefix, id);
#ifndef _WIN32
statsdClient.count(id, 1);
#endif
});
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
spdlog::info("Subscriber: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Subscriber: error {}", errMsg);
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::error("Published message hacked: {}", msgId);
spdlog::info("{}: {}", it.first, it.second);
}
}
);
if (eventType == ix::CobraConnection_EventType_Closed)
{
spdlog::info("Subscriber closed");
}
else if (eventType == ix::CobraConnection_EventType_Authenticated)
{
spdlog::info("Subscriber authenticated");
conn.subscribe(channel,
filter,
[&jsonWriter, &statsdClient, verbose, &tokens, &prefix, &msgCount](
const Json::Value& msg) {
if (verbose)
{
spdlog::info(jsonWriter.write(msg));
}
std::string id;
for (auto&& attr : tokens)
{
id += ".";
id += extractAttr(attr, msg);
}
spdlog::info("{} {}{}", msgCount++, prefix, id);
#ifndef _WIN32
statsdClient.count(id, 1);
#endif
});
}
else if (eventType == ix::CobraConnection_EventType_Subscribed)
{
spdlog::info("Subscriber: subscribed to channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
{
spdlog::info("Subscriber: unsubscribed from channel {}", subscriptionId);
}
else if (eventType == ix::CobraConnection_EventType_Error)
{
spdlog::error("Subscriber: error {}", errMsg);
}
else if (eventType == ix::CobraConnection_EventType_Published)
{
spdlog::error("Published message hacked: {}", msgId);
}
});
while (true)
{
@ -165,4 +165,4 @@ namespace ix
return 0;
}
}
} // namespace ix

159
ws/ws_connect.cpp
View File

@ -4,41 +4,40 @@
* Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <sstream>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include "linenoise.hpp"
#include <iostream>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <sstream>
namespace ix
{
class WebSocketConnect
{
public:
WebSocketConnect(const std::string& _url,
const std::string& headers,
bool disableAutomaticReconnection,
bool disablePerMessageDeflate,
bool binaryMode,
uint32_t maxWaitBetweenReconnectionRetries);
public:
WebSocketConnect(const std::string& _url,
const std::string& headers,
bool disableAutomaticReconnection,
bool disablePerMessageDeflate,
bool binaryMode,
uint32_t maxWaitBetweenReconnectionRetries);
void subscribe(const std::string& channel);
void start();
void stop();
void subscribe(const std::string& channel);
void start();
void stop();
void sendMessage(const std::string& text);
void sendMessage(const std::string& text);
private:
std::string _url;
WebSocketHttpHeaders _headers;
ix::WebSocket _webSocket;
bool _disablePerMessageDeflate;
bool _binaryMode;
private:
std::string _url;
WebSocketHttpHeaders _headers;
ix::WebSocket _webSocket;
bool _disablePerMessageDeflate;
bool _binaryMode;
void log(const std::string& msg);
WebSocketHttpHeaders parseHeaders(const std::string& data);
void log(const std::string& msg);
WebSocketHttpHeaders parseHeaders(const std::string& data);
};
WebSocketConnect::WebSocketConnect(const std::string& url,
@ -46,10 +45,10 @@ namespace ix
bool disableAutomaticReconnection,
bool disablePerMessageDeflate,
bool binaryMode,
uint32_t maxWaitBetweenReconnectionRetries) :
_url(url),
_disablePerMessageDeflate(disablePerMessageDeflate),
_binaryMode(binaryMode)
uint32_t maxWaitBetweenReconnectionRetries)
: _url(url)
, _disablePerMessageDeflate(disablePerMessageDeflate)
, _binaryMode(binaryMode)
{
if (disableAutomaticReconnection)
{
@ -81,7 +80,7 @@ namespace ix
if (pos == std::string::npos) continue;
auto key = token.substr(0, pos);
auto val = token.substr(pos+1);
auto val = token.substr(pos + 1);
std::cerr << key << ": " << val << std::endl;
headers[key] = val;
@ -114,61 +113,58 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const ix::WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) {
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
log("ws_connect: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
log("ws_connect: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
std::cout << it.first << ": " << it.second << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws_connect: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws_connect: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
ss << "ws_connect: received message: "
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
std::cerr << "Received ping" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
std::cerr << "Received pong" << std::endl;
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
ss << "ws_connect: received message: " << msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
std::cerr << "Received ping" << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
std::cerr << "Received pong" << std::endl;
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
@ -237,5 +233,4 @@ namespace ix
return 0;
}
}
} // namespace ix

32
ws/ws_echo_server.cpp
View File

@ -5,8 +5,8 @@
*/
#include <iostream>
#include <sstream>
#include <ixwebsocket/IXWebSocketServer.h>
#include <sstream>
namespace ix
{
@ -18,11 +18,9 @@ namespace ix
server.setOnConnectionCallback(
[greetings](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState)
{
std::shared_ptr<ConnectionState> connectionState) {
webSocket->setOnMessageCallback(
[webSocket, connectionState, greetings](const WebSocketMessagePtr& msg)
{
[webSocket, connectionState, greetings](const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
std::cerr << "New connection" << std::endl;
@ -42,29 +40,25 @@ namespace ix
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason << std::endl;
<< " code " << msg->closeInfo.code << " reason "
<< msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received "
<< msg->wireSize << " bytes"
<< std::endl;
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
webSocket->send(msg->str, msg->binary);
}
}
);
}
);
});
});
auto res = server.listen();
if (!res.first)
@ -78,4 +72,4 @@ namespace ix
return 0;
}
}
} // namespace ix

29
ws/ws_http_client.cpp
View File

@ -4,11 +4,11 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <sstream>
#include <fstream>
#include <iostream>
#include <ixwebsocket/IXHttpClient.h>
#include <ixwebsocket/IXWebSocketHttpHeaders.h>
#include <sstream>
namespace ix
{
@ -19,7 +19,7 @@ namespace ix
idx = path.rfind('/');
if (idx != std::string::npos)
{
std::string filename = path.substr(idx+1);
std::string filename = path.substr(idx + 1);
return filename;
}
else
@ -44,7 +44,7 @@ namespace ix
if (pos == std::string::npos) continue;
auto key = token.substr(0, pos);
auto val = token.substr(pos+1);
auto val = token.substr(pos + 1);
std::cerr << key << ": " << val << std::endl;
headers[key] = val;
@ -73,7 +73,7 @@ namespace ix
if (pos == std::string::npos) continue;
auto key = token.substr(0, pos);
auto val = token.substr(pos+1);
auto val = token.substr(pos + 1);
std::cerr << key << ": " << val << std::endl;
httpParameters[key] = val;
@ -104,14 +104,10 @@ namespace ix
args->maxRedirects = maxRedirects;
args->verbose = verbose;
args->compress = compress;
args->logger = [](const std::string& msg)
{
std::cout << msg;
};
args->onProgressCallback = [](int current, int total) -> bool
{
std::cerr << "\r" << "Downloaded "
<< current << " bytes out of " << total;
args->logger = [](const std::string& msg) { std::cout << msg; };
args->onProgressCallback = [](int current, int total) -> bool {
std::cerr << "\r"
<< "Downloaded " << current << " bytes out of " << total;
return true;
};
@ -160,7 +156,7 @@ namespace ix
std::cout << "Writing to disk: " << filename << std::endl;
std::ofstream out(filename);
out.write((char*)&response->payload.front(), response->payload.size());
out.write((char*) &response->payload.front(), response->payload.size());
out.close();
}
else
@ -173,11 +169,12 @@ namespace ix
{
std::cerr << "Binary output can mess up your terminal." << std::endl;
std::cerr << "Use the -O flag to save the file to disk." << std::endl;
std::cerr << "You can also use the --output option to specify a filename." << std::endl;
std::cerr << "You can also use the --output option to specify a filename."
<< std::endl;
}
}
}
return 0;
}
}
} // namespace ix

8
ws/ws_httpd.cpp
View File

@ -4,12 +4,12 @@
* Copyright (c) 2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <vector>
#include <fstream>
#include <sstream>
#include <iostream>
#include <ixwebsocket/IXHttpServer.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <vector>
namespace ix
{
@ -31,4 +31,4 @@ namespace ix
return 0;
}
}
} // namespace ix

135
ws/ws_ping_pong.cpp
View File

@ -5,33 +5,33 @@
*/
#include <iostream>
#include <sstream>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <sstream>
namespace ix
{
class WebSocketPingPong
{
public:
WebSocketPingPong(const std::string& _url);
public:
WebSocketPingPong(const std::string& _url);
void subscribe(const std::string& channel);
void start();
void stop();
void subscribe(const std::string& channel);
void start();
void stop();
void ping(const std::string& text);
void send(const std::string& text);
void ping(const std::string& text);
void send(const std::string& text);
private:
std::string _url;
ix::WebSocket _webSocket;
private:
std::string _url;
ix::WebSocket _webSocket;
void log(const std::string& msg);
void log(const std::string& msg);
};
WebSocketPingPong::WebSocketPingPong(const std::string& url) :
_url(url)
WebSocketPingPong::WebSocketPingPong(const std::string& url)
: _url(url)
{
;
}
@ -53,63 +53,57 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const ix::WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) {
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
log("ping_pong: connected");
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
log("ping_pong: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
std::cout << it.first << ": " << it.second << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ping_pong: disconnected:"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "ping_pong: received message: "
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
ss << "ping_pong: received ping message: "
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
ss << "ping_pong: received pong message: "
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ping_pong: disconnected:"
<< " code " << msg->closeInfo.code << " reason " << msg->closeInfo.reason
<< msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "ping_pong: received message: " << msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Ping)
{
ss << "ping_pong: received ping message: " << msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Pong)
{
ss << "ping_pong: received pong message: " << msg->str;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
@ -118,7 +112,8 @@ namespace ix
{
if (!_webSocket.ping(text).success)
{
std::cerr << "Failed to send ping message. Message too long (> 125 bytes) or endpoint is disconnected"
std::cerr << "Failed to send ping message. Message too long (> 125 bytes) or endpoint "
"is disconnected"
<< std::endl;
}
}
@ -160,4 +155,4 @@ namespace ix
return 0;
}
}
} // namespace ix

185
ws/ws_receive.cpp
View File

@ -4,19 +4,19 @@
* Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <vector>
#include <condition_variable>
#include <mutex>
#include <chrono>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include <ixcrypto/IXUuid.h>
#include <condition_variable>
#include <fstream>
#include <iostream>
#include <ixcrypto/IXBase64.h>
#include <ixcrypto/IXHash.h>
#include <ixcrypto/IXUuid.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <msgpack11/msgpack11.hpp>
#include <mutex>
#include <sstream>
#include <vector>
using msgpack11::MsgPack;
@ -24,42 +24,40 @@ namespace ix
{
class WebSocketReceiver
{
public:
WebSocketReceiver(const std::string& _url,
bool enablePerMessageDeflate,
int delayMs);
public:
WebSocketReceiver(const std::string& _url, bool enablePerMessageDeflate, int delayMs);
void subscribe(const std::string& channel);
void start();
void stop();
void subscribe(const std::string& channel);
void start();
void stop();
void waitForConnection();
void waitForMessage();
void handleMessage(const std::string& str);
void waitForConnection();
void waitForMessage();
void handleMessage(const std::string& str);
private:
std::string _url;
std::string _id;
ix::WebSocket _webSocket;
bool _enablePerMessageDeflate;
int _delayMs;
int _receivedFragmentCounter;
private:
std::string _url;
std::string _id;
ix::WebSocket _webSocket;
bool _enablePerMessageDeflate;
int _delayMs;
int _receivedFragmentCounter;
std::mutex _conditionVariableMutex;
std::condition_variable _condition;
std::mutex _conditionVariableMutex;
std::condition_variable _condition;
std::string extractFilename(const std::string& path);
void handleError(const std::string& errMsg, const std::string& id);
void log(const std::string& msg);
std::string extractFilename(const std::string& path);
void handleError(const std::string& errMsg, const std::string& id);
void log(const std::string& msg);
};
WebSocketReceiver::WebSocketReceiver(const std::string& url,
bool enablePerMessageDeflate,
int delayMs) :
_url(url),
_enablePerMessageDeflate(enablePerMessageDeflate),
_delayMs(delayMs),
_receivedFragmentCounter(0)
int delayMs)
: _url(url)
, _enablePerMessageDeflate(enablePerMessageDeflate)
, _delayMs(delayMs)
, _receivedFragmentCounter(0)
{
;
}
@ -98,7 +96,7 @@ namespace ix
idx = path.rfind('/');
if (idx != std::string::npos)
{
std::string filename = path.substr(idx+1);
std::string filename = path.substr(idx + 1);
return filename;
}
else
@ -107,8 +105,7 @@ namespace ix
}
}
void WebSocketReceiver::handleError(const std::string& errMsg,
const std::string& id)
void WebSocketReceiver::handleError(const std::string& errMsg, const std::string& id)
{
std::map<MsgPack, MsgPack> pdu;
pdu["kind"] = "error";
@ -156,7 +153,7 @@ namespace ix
std::cout << "Writing to disk: " << filenameTmp << std::endl;
std::ofstream out(filenameTmp);
out.write((char*)&content.front(), content.size());
out.write((char*) &content.front(), content.size());
out.close();
std::cout << "Renaming " << filenameTmp << " to " << filename << std::endl;
@ -182,70 +179,66 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const ix::WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) {
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
_condition.notify_one();
_condition.notify_one();
log("ws_receive: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
log("ws_receive: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
ss << "ws_receive: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
std::cout << it.first << ": " << it.second << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "ws_receive: transfered " << msg->wireSize << " bytes";
log(ss.str());
handleMessage(msg->str);
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
ss << "ws_receive: received fragment " << _receivedFragmentCounter++;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws_receive: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
ss << "ws_receive: transfered " << msg->wireSize << " bytes";
log(ss.str());
handleMessage(msg->str);
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
ss << "ws_receive: received fragment " << _receivedFragmentCounter++;
log(ss.str());
if (_delayMs > 0)
{
// Introduce an arbitrary delay, to simulate a slow connection
std::chrono::duration<double, std::milli> duration(_delayMs);
std::this_thread::sleep_for(duration);
}
}
else if (msg->type == ix::WebSocketMessageType::Error)
if (_delayMs > 0)
{
ss << "ws_receive ";
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
// Introduce an arbitrary delay, to simulate a slow connection
std::chrono::duration<double, std::milli> duration(_delayMs);
std::this_thread::sleep_for(duration);
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "ws_receive ";
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
void wsReceive(const std::string& url,
bool enablePerMessageDeflate,
int delayMs)
void wsReceive(const std::string& url, bool enablePerMessageDeflate, int delayMs)
{
WebSocketReceiver webSocketReceiver(url, enablePerMessageDeflate, delayMs);
webSocketReceiver.start();
@ -261,11 +254,9 @@ namespace ix
webSocketReceiver.stop();
}
int ws_receive_main(const std::string& url,
bool enablePerMessageDeflate,
int delayMs)
int ws_receive_main(const std::string& url, bool enablePerMessageDeflate, int delayMs)
{
wsReceive(url, enablePerMessageDeflate, delayMs);
return 0;
}
}
} // namespace ix

7
ws/ws_redis_publish.cpp
View File

@ -4,9 +4,9 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include "IXRedisClient.h"
#include <iostream>
#include <sstream>
#include "IXRedisClient.h"
namespace ix
{
@ -41,8 +41,7 @@ namespace ix
{
if (!redisClient.publish(channel, message, errMsg))
{
std::cerr << "Error publishing to channel " << channel
<< "error: " << errMsg
std::cerr << "Error publishing to channel " << channel << "error: " << errMsg
<< std::endl;
return 1;
}
@ -50,4 +49,4 @@ namespace ix
return 0;
}
}
} // namespace ix

21
ws/ws_redis_subscribe.cpp
View File

@ -4,12 +4,12 @@
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include "IXRedisClient.h"
#include <atomic>
#include <chrono>
#include <iostream>
#include <sstream>
#include <chrono>
#include <thread>
#include <atomic>
#include "IXRedisClient.h"
namespace ix
{
@ -41,9 +41,7 @@ namespace ix
std::atomic<int> msgPerSeconds(0);
std::atomic<int> msgCount(0);
auto callback = [&msgPerSeconds, &msgCount, verbose]
(const std::string& message)
{
auto callback = [&msgPerSeconds, &msgCount, verbose](const std::string& message) {
if (verbose)
{
std::cout << "received: " << message << std::endl;
@ -53,18 +51,15 @@ namespace ix
msgCount++;
};
auto responseCallback = [](const std::string& redisResponse)
{
auto responseCallback = [](const std::string& redisResponse) {
std::cout << "Redis subscribe response: " << redisResponse << std::endl;
};
auto timer = [&msgPerSeconds, &msgCount]
{
auto timer = [&msgPerSeconds, &msgCount] {
while (true)
{
std::cout << "#messages " << msgCount << " "
<< "msg/s " << msgPerSeconds
<< std::endl;
<< "msg/s " << msgPerSeconds << std::endl;
msgPerSeconds = 0;
auto duration = std::chrono::seconds(1);
@ -83,4 +78,4 @@ namespace ix
return 0;
}
}
} // namespace ix

236
ws/ws_send.cpp
View File

@ -4,19 +4,19 @@
* Copyright (c) 2017-2018 Machine Zone, Inc. All rights reserved.
*/
#include <iostream>
#include <fstream>
#include <sstream>
#include <vector>
#include <condition_variable>
#include <mutex>
#include <chrono>
#include <ixwebsocket/IXWebSocket.h>
#include <ixwebsocket/IXSocket.h>
#include <ixcrypto/IXUuid.h>
#include <condition_variable>
#include <fstream>
#include <iostream>
#include <ixcrypto/IXBase64.h>
#include <ixcrypto/IXHash.h>
#include <ixcrypto/IXUuid.h>
#include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXWebSocket.h>
#include <msgpack11/msgpack11.hpp>
#include <mutex>
#include <sstream>
#include <vector>
using msgpack11::MsgPack;
@ -24,35 +24,33 @@ namespace ix
{
class WebSocketSender
{
public:
WebSocketSender(const std::string& _url,
bool enablePerMessageDeflate);
public:
WebSocketSender(const std::string& _url, bool enablePerMessageDeflate);
void subscribe(const std::string& channel);
void start();
void stop();
void subscribe(const std::string& channel);
void start();
void stop();
void waitForConnection();
void waitForAck();
void waitForConnection();
void waitForAck();
void sendMessage(const std::string& filename, bool throttle);
void sendMessage(const std::string& filename, bool throttle);
private:
std::string _url;
std::string _id;
ix::WebSocket _webSocket;
bool _enablePerMessageDeflate;
private:
std::string _url;
std::string _id;
ix::WebSocket _webSocket;
bool _enablePerMessageDeflate;
std::mutex _conditionVariableMutex;
std::condition_variable _condition;
std::mutex _conditionVariableMutex;
std::condition_variable _condition;
void log(const std::string& msg);
void log(const std::string& msg);
};
WebSocketSender::WebSocketSender(const std::string& url,
bool enablePerMessageDeflate) :
_url(url),
_enablePerMessageDeflate(enablePerMessageDeflate)
WebSocketSender::WebSocketSender(const std::string& url, bool enablePerMessageDeflate)
: _url(url)
, _enablePerMessageDeflate(enablePerMessageDeflate)
{
;
}
@ -95,7 +93,7 @@ namespace ix
file.seekg(0, file.beg);
memblock.resize((size_t) size);
file.read((char*)&memblock.front(), static_cast<std::streamsize>(size));
file.read((char*) &memblock.front(), static_cast<std::streamsize>(size));
return memblock;
}
@ -111,114 +109,110 @@ namespace ix
std::stringstream ss;
log(std::string("Connecting to url: ") + _url);
_webSocket.setOnMessageCallback(
[this](const WebSocketMessagePtr& msg)
_webSocket.setOnMessageCallback([this](const WebSocketMessagePtr& msg) {
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
std::stringstream ss;
if (msg->type == ix::WebSocketMessageType::Open)
{
_condition.notify_one();
_condition.notify_one();
log("ws_send: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cout << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
log("ws_send: connected");
std::cout << "Uri: " << msg->openInfo.uri << std::endl;
std::cout << "Handshake Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
ss << "ws_send: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
std::cout << it.first << ": " << it.second << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
_condition.notify_one();
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
ss << "ws_send: connection closed:";
ss << " code " << msg->closeInfo.code;
ss << " reason " << msg->closeInfo.reason << std::endl;
log(ss.str());
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
_condition.notify_one();
ss << "ws_send: received message (" << msg->wireSize << " bytes)";
log(ss.str());
ss << "ws_send: received message (" << msg->wireSize << " bytes)";
log(ss.str());
std::string errMsg;
MsgPack data = MsgPack::parse(msg->str, errMsg);
if (!errMsg.empty())
{
std::cerr << "Invalid MsgPack response" << std::endl;
return;
}
std::string errMsg;
MsgPack data = MsgPack::parse(msg->str, errMsg);
if (!errMsg.empty())
{
std::cerr << "Invalid MsgPack response" << std::endl;
return;
}
std::string id = data["id"].string_value();
if (_id != id)
{
std::cerr << "Invalid id" << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Error)
std::string id = data["id"].string_value();
if (_id != id)
{
ss << "ws_send ";
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
std::cerr << "Invalid id" << std::endl;
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
ss << "ws_send ";
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
log(ss.str());
}
else
{
ss << "Invalid ix::WebSocketMessageType";
log(ss.str());
}
});
_webSocket.start();
}
class Bench
{
public:
Bench(const std::string& description) :
_description(description),
_start(std::chrono::system_clock::now()),
_reported(false)
public:
Bench(const std::string& description)
: _description(description)
, _start(std::chrono::system_clock::now())
, _reported(false)
{
;
}
~Bench()
{
if (!_reported)
{
;
report();
}
}
~Bench()
{
if (!_reported)
{
report();
}
}
void report()
{
auto now = std::chrono::system_clock::now();
auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - _start);
void report()
{
auto now = std::chrono::system_clock::now();
auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now - _start);
_ms = milliseconds.count();
std::cout << _description << " completed in " << _ms << "ms" << std::endl;
_ms = milliseconds.count();
std::cout << _description << " completed in "
<< _ms << "ms" << std::endl;
_reported = true;
}
_reported = true;
}
uint64_t getDuration() const
{
return _ms;
}
uint64_t getDuration() const
{
return _ms;
}
private:
std::string _description;
std::chrono::time_point<std::chrono::system_clock> _start;
uint64_t _ms;
bool _reported;
private:
std::string _description;
std::chrono::time_point<std::chrono::system_clock> _start;
uint64_t _ms;
bool _reported;
};
void WebSocketSender::sendMessage(const std::string& filename,
bool throttle)
void WebSocketSender::sendMessage(const std::string& filename, bool throttle)
{
std::vector<uint8_t> content;
{
@ -239,9 +233,7 @@ namespace ix
MsgPack msg(pdu);
Bench bench("Sending file through websocket");
_webSocket.sendBinary(msg.dump(),
[throttle](int current, int total) -> bool
{
_webSocket.sendBinary(msg.dump(), [throttle](int current, int total) -> bool {
std::cout << "ws_send: Step " << current << " out of " << total << std::endl;
if (throttle)
@ -256,8 +248,7 @@ namespace ix
do
{
size_t bufferedAmount = _webSocket.bufferedAmount();
std::cout << "ws_send: " << bufferedAmount
<< " bytes left to be sent" << std::endl;
std::cout << "ws_send: " << bufferedAmount << " bytes left to be sent" << std::endl;
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
@ -289,8 +280,7 @@ namespace ix
webSocketSender.stop();
}
int ws_send_main(const std::string& url,
const std::string& path)
int ws_send_main(const std::string& url, const std::string& path)
{
bool throttle = false;
bool enablePerMessageDeflate = false;
@ -298,4 +288,4 @@ namespace ix
wsSend(url, path, enablePerMessageDeflate, throttle);
return 0;
}
}
} // namespace ix

9
ws/ws_snake.cpp
View File

@ -5,10 +5,9 @@
*/
#include "IXSnakeServer.h"
#include <fstream>
#include <iostream>
#include <sstream>
#include <fstream>
namespace
{
@ -24,7 +23,7 @@ namespace
file.seekg(0, file.beg);
memblock.resize((size_t) size);
file.read((char*)&memblock.front(), static_cast<std::streamsize>(size));
file.read((char*) &memblock.front(), static_cast<std::streamsize>(size));
return memblock;
}
@ -34,7 +33,7 @@ namespace
auto vec = load(path);
return std::string(vec.begin(), vec.end());
}
}
} // namespace
namespace ix
{
@ -80,4 +79,4 @@ namespace ix
return 0; // should never reach this
}
}
} // namespace ix

123
ws/ws_transfer.cpp
View File

@ -5,8 +5,8 @@
*/
#include <iostream>
#include <sstream>
#include <ixwebsocket/IXWebSocketServer.h>
#include <sstream>
namespace ix
{
@ -16,76 +16,67 @@ namespace ix
ix::WebSocketServer server(port, hostname);
server.setOnConnectionCallback(
[&server](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState)
{
webSocket->setOnMessageCallback(
[webSocket, connectionState, &server](const WebSocketMessagePtr& msg)
server.setOnConnectionCallback([&server](std::shared_ptr<ix::WebSocket> webSocket,
std::shared_ptr<ConnectionState> connectionState) {
webSocket->setOnMessageCallback([webSocket, connectionState, &server](
const WebSocketMessagePtr& msg) {
if (msg->type == ix::WebSocketMessageType::Open)
{
std::cerr << "New connection" << std::endl;
std::cerr << "id: " << connectionState->getId() << std::endl;
std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
std::cerr << "Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
if (msg->type == ix::WebSocketMessageType::Open)
std::cerr << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code << " reason "
<< msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment " << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
for (auto&& client : server.getClients())
{
if (client != webSocket)
{
std::cerr << "New connection" << std::endl;
std::cerr << "id: " << connectionState->getId() << std::endl;
std::cerr << "Uri: " << msg->openInfo.uri << std::endl;
std::cerr << "Headers:" << std::endl;
for (auto it : msg->openInfo.headers)
{
std::cerr << it.first << ": " << it.second << std::endl;
}
}
else if (msg->type == ix::WebSocketMessageType::Close)
{
std::cerr << "Closed connection"
<< " code " << msg->closeInfo.code
<< " reason " << msg->closeInfo.reason << std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Error)
{
std::stringstream ss;
ss << "Connection error: " << msg->errorInfo.reason << std::endl;
ss << "#retries: " << msg->errorInfo.retries << std::endl;
ss << "Wait time(ms): " << msg->errorInfo.wait_time << std::endl;
ss << "HTTP Status: " << msg->errorInfo.http_status << std::endl;
std::cerr << ss.str();
}
else if (msg->type == ix::WebSocketMessageType::Fragment)
{
std::cerr << "Received message fragment "
<< std::endl;
}
else if (msg->type == ix::WebSocketMessageType::Message)
{
std::cerr << "Received " << msg->wireSize << " bytes" << std::endl;
for (auto&& client : server.getClients())
{
if (client != webSocket)
{
client->send(msg->str,
msg->binary,
[](int current, int total) -> bool
{
std::cerr << "ws_transfer: Step " << current
<< " out of " << total << std::endl;
return true;
});
client->send(msg->str, msg->binary, [](int current, int total) -> bool {
std::cerr << "ws_transfer: Step " << current << " out of " << total
<< std::endl;
return true;
});
do
{
size_t bufferedAmount = client->bufferedAmount();
std::cerr << "ws_transfer: " << bufferedAmount
<< " bytes left to be sent" << std::endl;
do
{
size_t bufferedAmount = client->bufferedAmount();
std::cerr << "ws_transfer: " << bufferedAmount
<< " bytes left to be sent" << std::endl;
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
} while (client->bufferedAmount() != 0);
}
}
std::chrono::duration<double, std::milli> duration(10);
std::this_thread::sleep_for(duration);
} while (client->bufferedAmount() != 0);
}
}
);
}
);
}
});
});
auto res = server.listen();
if (!res.first)
@ -99,4 +90,4 @@ namespace ix
return 0;
}
}
} // namespace ix