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

minor refactoring to delete files which are not needed

Browse Source
This commit is contained in:
Benjamin Sergeant 2020-03-12 12:13:31 -07:00
parent 7642ccc99e
commit 83ae105edb
9 changed files with 17 additions and 329 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ixbots/ixbots/IXCobraToSentryBot.cpp
View File

@ -17,7 +17,7 @@
namespace ix namespace ix
{ {
int cobra_to_sentry_bot(const ix::CobraConfig& config, int cobra_to_sentry_bot(const CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& dsn, const std::string& dsn,

2
ixbots/ixbots/IXCobraToSentryBot.h
View File

@ -10,7 +10,7 @@
namespace ix namespace ix
{ {
int cobra_to_sentry_bot(const ix::CobraConfig& config, int cobra_to_sentry_bot(const CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& dsn, const std::string& dsn,

10
ixbots/ixbots/IXCobraToStatsdBot.h
View File

@ -14,9 +14,9 @@ namespace ix
int cobra_to_statsd_bot(const ix::CobraConfig& config, int cobra_to_statsd_bot(const ix::CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,
const std::string& dsn, const std::string& host,
bool verbose, int port,
bool strict, const std::string& prefix,
int jobs, const std::string& fields,
size_t maxQueueSize); bool verbose);
} // namespace ix } // namespace ix

4
test/IXCobraToSentryBotTest.cpp
View File

@ -153,7 +153,9 @@ TEST_CASE("Cobra_to_sentry_bot", "[foo]")
// https://xxxxx:yyyyyy@sentry.io/1234567 // https://xxxxx:yyyyyy@sentry.io/1234567
std::stringstream oss; std::stringstream oss;
oss << "http://xxxxxxx:yyyyyyy@localhost:" << sentryPort << "/1234567"; std::string scheme("http://");
oss << scheme << "xxxxxxx:yyyyyyy@localhost:" << sentryPort << "/1234567";
std::string dsn = oss.str(); std::string dsn = oss.str();
// Only run the bot for 3 seconds // Only run the bot for 3 seconds

2
ws/CMakeLists.txt
View File

@ -58,8 +58,6 @@ add_executable(ws
ws_cobra_subscribe.cpp ws_cobra_subscribe.cpp
ws_cobra_metrics_publish.cpp ws_cobra_metrics_publish.cpp
ws_cobra_publish.cpp ws_cobra_publish.cpp
ws_cobra_to_statsd.cpp
ws_cobra_to_sentry.cpp
ws_cobra_metrics_to_redis.cpp ws_cobra_metrics_to_redis.cpp
ws_httpd.cpp ws_httpd.cpp
ws_autobahn.cpp ws_autobahn.cpp

10
ws/ws.cpp
View File

@ -15,6 +15,8 @@
#include <ixwebsocket/IXNetSystem.h> #include <ixwebsocket/IXNetSystem.h>
#include <ixwebsocket/IXSocket.h> #include <ixwebsocket/IXSocket.h>
#include <ixwebsocket/IXUserAgent.h> #include <ixwebsocket/IXUserAgent.h>
#include <ixbots/IXCobraToSentryBot.h>
#include <ixbots/IXCobraToStatsdBot.h>
#include <spdlog/spdlog.h> #include <spdlog/spdlog.h>
#include <sstream> #include <sstream>
#include <string> #include <string>
@ -440,13 +442,15 @@ int main(int argc, char** argv)
} }
else if (app.got_subcommand("cobra_to_statsd")) else if (app.got_subcommand("cobra_to_statsd"))
{ {
ret = ix::ws_cobra_to_statsd_main( ret = ix::cobra_to_statsd_bot(
cobraConfig, channel, filter, hostname, statsdPort, prefix, fields, verbose); cobraConfig, channel, filter, hostname, statsdPort, prefix, fields, verbose);
} }
else if (app.got_subcommand("cobra_to_sentry")) else if (app.got_subcommand("cobra_to_sentry"))
{ {
ret = ix::ws_cobra_to_sentry_main( bool enableHeartbeat = true;
cobraConfig, channel, filter, dsn, verbose, strict, jobs, maxQueueSize); int runtime = -1;
ret = ix::cobra_to_sentry_bot(
cobraConfig, channel, filter, dsn, verbose, strict, jobs, maxQueueSize, enableHeartbeat, runtime);
} }
else if (app.got_subcommand("cobra_metrics_to_redis")) else if (app.got_subcommand("cobra_metrics_to_redis"))
{ {

18
ws/ws.h
View File

@ -90,24 +90,6 @@ namespace ix
const std::string& path, const std::string& path,
bool stress); bool stress);
int ws_cobra_to_statsd_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& host,
int port,
const std::string& prefix,
const std::string& fields,
bool verbose);
int ws_cobra_to_sentry_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& dsn,
bool verbose,
bool strict,
int jobs,
size_t maxQueueSize);
int ws_cobra_metrics_to_redis(const ix::CobraConfig& config, int ws_cobra_metrics_to_redis(const ix::CobraConfig& config,
const std::string& channel, const std::string& channel,
const std::string& filter, const std::string& filter,

25
ws/ws_cobra_to_sentry.cpp
View File

@ -1,25 +0,0 @@
/*
* ws_cobra_to_sentry.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <ixbots/IXCobraToSentryBot.h>
namespace ix
{
int ws_cobra_to_sentry_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& dsn,
bool verbose,
bool strict,
int jobs,
size_t maxQueueSize)
{
bool enableHeartbeat = true;
int runtime = -1;
return cobra_to_sentry_bot(
config, channel, filter, dsn, verbose, strict, jobs, maxQueueSize, enableHeartbeat, runtime);
}
} // namespace ix

273
ws/ws_cobra_to_statsd.cpp
View File

@ -1,273 +0,0 @@
/*
* ws_cobra_to_statsd.cpp
* Author: Benjamin Sergeant
* Copyright (c) 2019 Machine Zone, Inc. All rights reserved.
*/
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <ixcobra/IXCobraConnection.h>
#include <spdlog/spdlog.h>
#include <sstream>
#include <thread>
#include <vector>
#ifndef _WIN32
#include <statsd_client.h>
#endif
namespace
{
class QueueManager
{
public:
QueueManager(size_t maxQueueSize, std::atomic<bool>& stop)
: _maxQueueSize(maxQueueSize)
, _stop(stop)
{
}
Json::Value pop();
void add(Json::Value msg);
private:
std::queue<Json::Value> _queue;
std::mutex _mutex;
std::condition_variable _condition;
size_t _maxQueueSize;
std::atomic<bool>& _stop;
};
Json::Value QueueManager::pop()
{
std::unique_lock<std::mutex> lock(_mutex);
if (_queue.empty())
{
Json::Value val;
return val;
}
_condition.wait(lock, [this] { return !_stop; });
auto msg = _queue.front();
_queue.pop();
return msg;
}
void QueueManager::add(Json::Value msg)
{
std::unique_lock<std::mutex> lock(_mutex);
// if the sending is not fast enough there is no point
// in queuing too many events.
if (_queue.size() < _maxQueueSize)
{
_queue.push(msg);
_condition.notify_one();
}
}
} // namespace
namespace ix
{
// fields are command line argument that can be specified multiple times
std::vector<std::string> parseFields(const std::string& fields)
{
std::vector<std::string> tokens;
// Split by \n
std::string token;
std::stringstream tokenStream(fields);
while (std::getline(tokenStream, token))
{
tokens.push_back(token);
}
return tokens;
}
//
// Extract an attribute from a Json Value.
// extractAttr("foo.bar", {"foo": {"bar": "baz"}}) => baz
//
std::string extractAttr(const std::string& attr, const Json::Value& jsonValue)
{
// Split by .
std::string token;
std::stringstream tokenStream(attr);
Json::Value val(jsonValue);
while (std::getline(tokenStream, token, '.'))
{
val = val[token];
}
return val.asString();
}
int ws_cobra_to_statsd_main(const ix::CobraConfig& config,
const std::string& channel,
const std::string& filter,
const std::string& host,
int port,
const std::string& prefix,
const std::string& fields,
bool verbose)
{
ix::CobraConnection conn;
conn.configure(config);
conn.connect();
auto tokens = parseFields(fields);
Json::FastWriter jsonWriter;
std::atomic<uint64_t> sentCount(0);
std::atomic<uint64_t> receivedCount(0);
std::atomic<bool> stop(false);
size_t maxQueueSize = 1000;
QueueManager queueManager(maxQueueSize, stop);
auto timer = [&sentCount, &receivedCount] {
while (true)
{
spdlog::info("messages received {} sent {}", receivedCount, sentCount);
auto duration = std::chrono::seconds(1);
std::this_thread::sleep_for(duration);
}
};
std::thread t1(timer);
auto heartbeat = [&sentCount, &receivedCount] {
std::string state("na");
while (true)
{
std::stringstream ss;
ss << "messages received " << receivedCount;
ss << "messages sent " << sentCount;
std::string currentState = ss.str();
if (currentState == state)
{
spdlog::error("no messages received or sent for 1 minute, exiting");
exit(1);
}
state = currentState;
auto duration = std::chrono::minutes(1);
std::this_thread::sleep_for(duration);
}
};
std::thread t2(heartbeat);
auto statsdSender = [&queueManager, &host, &port, &sentCount, &tokens, &prefix, &stop] {
// statsd client
// test with netcat as a server: `nc -ul 8125`
bool statsdBatch = true;
#ifndef _WIN32
statsd::StatsdClient statsdClient(host, port, prefix, statsdBatch);
#else
int statsdClient;
#endif
while (true)
{
Json::Value msg = queueManager.pop();
if (msg.isNull()) continue;
if (stop) return;
std::string id;
for (auto&& attr : tokens)
{
id += ".";
id += extractAttr(attr, msg);
}
sentCount += 1;
#ifndef _WIN32
statsdClient.count(id, 1);
#endif
}
};
std::thread t3(statsdSender);
conn.setEventCallback(
[&conn, &channel, &filter, &jsonWriter, verbose, &queueManager, &receivedCount](
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");
for (auto it : headers)
{
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, &queueManager, verbose, &receivedCount](
const Json::Value& msg) {
if (verbose)
{
spdlog::info(jsonWriter.write(msg));
}
receivedCount++;
++receivedCount;
queueManager.add(msg);
});
}
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);
}
else if (eventType == ix::CobraConnection_EventType_Pong)
{
spdlog::info("Received websocket pong");
}
});
while (true)
{
std::chrono::duration<double, std::milli> duration(1000);
std::this_thread::sleep_for(duration);
}
return 0;
}
} // namespace ix