2019-09-06 06:57:05 +02:00
|
|
|
/*
|
|
|
|
* Author: Benjamin Sergeant
|
|
|
|
* Copyright (c) 2018 Machine Zone. All rights reserved.
|
|
|
|
*/
|
|
|
|
|
2019-09-23 19:25:23 +02:00
|
|
|
#include "IXTest.h"
|
2019-09-06 06:57:05 +02:00
|
|
|
#include "catch.hpp"
|
2019-09-23 19:25:23 +02:00
|
|
|
#include <iostream>
|
|
|
|
#include <ixcobra/IXCobraMetricsPublisher.h>
|
|
|
|
#include <ixcrypto/IXUuid.h>
|
2019-09-24 06:04:01 +02:00
|
|
|
#include <ixsnake/IXRedisServer.h>
|
2019-09-23 20:46:16 +02:00
|
|
|
#include <ixsnake/IXSnakeServer.h>
|
2019-09-23 19:25:23 +02:00
|
|
|
#include <set>
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
using namespace ix;
|
|
|
|
|
|
|
|
namespace
|
|
|
|
{
|
2019-09-24 06:04:01 +02:00
|
|
|
std::atomic<size_t> incomingBytes(0);
|
|
|
|
std::atomic<size_t> outgoingBytes(0);
|
|
|
|
|
|
|
|
void setupTrafficTrackerCallback()
|
|
|
|
{
|
|
|
|
ix::CobraConnection::setTrafficTrackerCallback([](size_t size, bool incoming) {
|
|
|
|
if (incoming)
|
|
|
|
{
|
|
|
|
incomingBytes += size;
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
outgoingBytes += size;
|
|
|
|
}
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2019-09-06 06:57:05 +02:00
|
|
|
std::atomic<bool> gStop;
|
|
|
|
std::atomic<bool> gSubscriberConnectedAndSubscribed;
|
2019-09-23 04:22:48 +02:00
|
|
|
std::atomic<size_t> gUniqueMessageIdsCount;
|
2019-09-06 06:57:05 +02:00
|
|
|
std::atomic<int> gMessageCount;
|
|
|
|
|
|
|
|
std::set<std::string> gIds;
|
|
|
|
std::mutex gProtectIds; // std::set is no thread-safe, so protect access with this mutex.
|
|
|
|
|
|
|
|
//
|
|
|
|
// Background thread subscribe to the channel and validates what was sent
|
|
|
|
//
|
2020-03-20 20:21:45 +01:00
|
|
|
void startSubscriber(const ix::CobraConfig& config, const std::string& channel)
|
2019-09-06 06:57:05 +02:00
|
|
|
{
|
|
|
|
gSubscriberConnectedAndSubscribed = false;
|
|
|
|
gUniqueMessageIdsCount = 0;
|
|
|
|
gMessageCount = 0;
|
|
|
|
|
|
|
|
ix::CobraConnection conn;
|
2020-03-20 20:21:45 +01:00
|
|
|
conn.configure(config);
|
2019-09-06 06:57:05 +02:00
|
|
|
conn.connect();
|
|
|
|
|
2020-03-20 20:21:45 +01:00
|
|
|
conn.setEventCallback([&conn, &channel](ix::CobraConnectionEventType eventType,
|
|
|
|
const std::string& errMsg,
|
|
|
|
const ix::WebSocketHttpHeaders& headers,
|
|
|
|
const std::string& subscriptionId,
|
|
|
|
CobraConnection::MsgId msgId) {
|
2019-09-23 19:25:23 +02:00
|
|
|
if (eventType == ix::CobraConnection_EventType_Open)
|
2019-09-06 06:57:05 +02:00
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber connected:";
|
2019-09-24 06:04:01 +02:00
|
|
|
for (auto&& it : headers)
|
|
|
|
{
|
|
|
|
log("Headers " + it.first + " " + it.second);
|
|
|
|
}
|
2019-09-23 19:25:23 +02:00
|
|
|
}
|
|
|
|
if (eventType == ix::CobraConnection_EventType_Error)
|
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber error:" << errMsg;
|
2019-09-23 19:25:23 +02:00
|
|
|
}
|
|
|
|
else if (eventType == ix::CobraConnection_EventType_Authenticated)
|
|
|
|
{
|
|
|
|
log("Subscriber authenticated");
|
|
|
|
std::string filter;
|
2020-03-14 00:06:13 +01:00
|
|
|
std::string position("$");
|
|
|
|
|
2020-03-13 20:49:37 +01:00
|
|
|
conn.subscribe(
|
2020-03-20 20:21:45 +01:00
|
|
|
channel, filter, position, [](const Json::Value& msg, const std::string& /*position*/) {
|
2020-03-13 20:49:37 +01:00
|
|
|
log(msg.toStyledString());
|
2019-09-23 19:25:23 +02:00
|
|
|
|
2020-03-13 20:49:37 +01:00
|
|
|
std::string id = msg["id"].asString();
|
|
|
|
{
|
|
|
|
std::lock_guard<std::mutex> guard(gProtectIds);
|
|
|
|
gIds.insert(id);
|
|
|
|
}
|
2019-09-23 19:25:23 +02:00
|
|
|
|
2020-03-13 20:49:37 +01:00
|
|
|
gMessageCount++;
|
|
|
|
});
|
2019-09-23 19:25:23 +02:00
|
|
|
}
|
|
|
|
else if (eventType == ix::CobraConnection_EventType_Subscribed)
|
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber: subscribed to channel " << subscriptionId;
|
2020-03-20 20:21:45 +01:00
|
|
|
if (subscriptionId == channel)
|
2019-09-23 19:25:23 +02:00
|
|
|
{
|
|
|
|
gSubscriberConnectedAndSubscribed = true;
|
2019-09-06 06:57:05 +02:00
|
|
|
}
|
2019-09-23 19:25:23 +02:00
|
|
|
else
|
2019-09-06 06:57:05 +02:00
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber: unexpected channel " << subscriptionId;
|
2019-09-06 06:57:05 +02:00
|
|
|
}
|
2019-09-23 19:25:23 +02:00
|
|
|
}
|
|
|
|
else if (eventType == ix::CobraConnection_EventType_UnSubscribed)
|
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber: ununexpected from channel " << subscriptionId;
|
2020-03-20 20:21:45 +01:00
|
|
|
if (subscriptionId != channel)
|
2019-09-06 06:57:05 +02:00
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber: unexpected channel " << subscriptionId;
|
2019-09-06 06:57:05 +02:00
|
|
|
}
|
|
|
|
}
|
2019-09-23 19:25:23 +02:00
|
|
|
else if (eventType == ix::CobraConnection_EventType_Published)
|
|
|
|
{
|
2020-03-12 19:15:54 +01:00
|
|
|
TLogger() << "Subscriber: published message acked: " << msgId;
|
2019-09-23 19:25:23 +02:00
|
|
|
}
|
|
|
|
});
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
while (!gStop)
|
|
|
|
{
|
|
|
|
std::chrono::duration<double, std::milli> duration(10);
|
|
|
|
std::this_thread::sleep_for(duration);
|
|
|
|
}
|
|
|
|
|
2020-03-20 20:21:45 +01:00
|
|
|
conn.unsubscribe(channel);
|
2019-09-06 06:57:05 +02:00
|
|
|
conn.disconnect();
|
|
|
|
|
|
|
|
gUniqueMessageIdsCount = gIds.size();
|
|
|
|
}
|
2019-09-24 20:46:54 +02:00
|
|
|
|
|
|
|
// publish 100 messages, during roughly 100ms
|
|
|
|
// this is used to test thread safety of CobraMetricsPublisher::push
|
|
|
|
void runAdditionalPublisher(ix::CobraMetricsPublisher* cobraMetricsPublisher)
|
|
|
|
{
|
|
|
|
Json::Value data;
|
|
|
|
data["foo"] = "bar";
|
|
|
|
|
|
|
|
for (int i = 0; i < 100; ++i)
|
|
|
|
{
|
|
|
|
cobraMetricsPublisher->push("sms_metric_F_id", data);
|
|
|
|
ix::msleep(1);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-09-23 19:25:23 +02:00
|
|
|
} // namespace
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
TEST_CASE("Cobra_Metrics_Publisher", "[cobra]")
|
|
|
|
{
|
2019-09-06 07:03:47 +02:00
|
|
|
int port = getFreePort();
|
2020-03-20 20:21:45 +01:00
|
|
|
bool preferTLS = false;
|
|
|
|
snake::AppConfig appConfig = makeSnakeServerConfig(port, preferTLS);
|
2019-09-24 06:04:01 +02:00
|
|
|
|
|
|
|
// Start a redis server
|
|
|
|
ix::RedisServer redisServer(appConfig.redisPort);
|
|
|
|
auto res = redisServer.listen();
|
|
|
|
REQUIRE(res.first);
|
|
|
|
redisServer.start();
|
|
|
|
|
|
|
|
// Start a snake server
|
2019-09-06 06:57:05 +02:00
|
|
|
snake::SnakeServer snakeServer(appConfig);
|
|
|
|
snakeServer.run();
|
|
|
|
|
2019-09-24 06:04:01 +02:00
|
|
|
setupTrafficTrackerCallback();
|
|
|
|
|
2020-03-20 20:21:45 +01:00
|
|
|
std::string channel = ix::generateSessionId();
|
|
|
|
std::string endpoint = makeCobraEndpoint(port, preferTLS);
|
|
|
|
std::string appkey("FC2F10139A2BAc53BB72D9db967b024f");
|
|
|
|
std::string role = "_sub";
|
|
|
|
std::string secret = "66B1dA3ED5fA074EB5AE84Dd8CE3b5ba";
|
2019-09-06 07:03:47 +02:00
|
|
|
|
2020-03-20 20:21:45 +01:00
|
|
|
ix::CobraConfig config;
|
|
|
|
config.endpoint = endpoint;
|
|
|
|
config.appkey = appkey;
|
|
|
|
config.rolename = role;
|
|
|
|
config.rolesecret = secret;
|
|
|
|
config.socketTLSOptions = makeClientTLSOptions();
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
gStop = false;
|
2020-03-20 20:21:45 +01:00
|
|
|
std::thread subscriberThread(&startSubscriber, config, channel);
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
int timeout = 10 * 1000; // 10s
|
2019-09-12 20:43:52 +02:00
|
|
|
|
2019-09-06 06:57:05 +02:00
|
|
|
// Wait until the subscriber is ready (authenticated + subscription successful)
|
|
|
|
while (!gSubscriberConnectedAndSubscribed)
|
|
|
|
{
|
|
|
|
std::chrono::duration<double, std::milli> duration(10);
|
|
|
|
std::this_thread::sleep_for(duration);
|
2019-09-12 20:43:52 +02:00
|
|
|
|
2019-09-06 06:57:05 +02:00
|
|
|
timeout -= 10;
|
|
|
|
if (timeout <= 0)
|
|
|
|
{
|
2019-09-24 06:04:01 +02:00
|
|
|
snakeServer.stop();
|
|
|
|
redisServer.stop();
|
2019-09-06 06:57:05 +02:00
|
|
|
REQUIRE(false); // timeout
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
ix::CobraMetricsPublisher cobraMetricsPublisher;
|
2020-03-20 20:21:45 +01:00
|
|
|
cobraMetricsPublisher.configure(config, channel);
|
2019-09-06 06:57:05 +02:00
|
|
|
cobraMetricsPublisher.setSession(uuid4());
|
2020-03-20 20:21:45 +01:00
|
|
|
cobraMetricsPublisher.enable(true);
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
Json::Value data;
|
|
|
|
data["foo"] = "bar";
|
|
|
|
|
|
|
|
// (1) Publish without restrictions
|
|
|
|
cobraMetricsPublisher.push("sms_metric_A_id", data); // (msg #1)
|
|
|
|
cobraMetricsPublisher.push("sms_metric_B_id", data); // (msg #2)
|
|
|
|
|
|
|
|
// (2) Restrict what is sent using a blacklist
|
|
|
|
// Add one entry to the blacklist
|
|
|
|
// (will send msg #3)
|
|
|
|
cobraMetricsPublisher.setBlacklist({
|
|
|
|
"sms_metric_B_id" // this id will not be sent
|
|
|
|
});
|
|
|
|
// (msg #4)
|
|
|
|
cobraMetricsPublisher.push("sms_metric_A_id", data);
|
|
|
|
// ...
|
|
|
|
cobraMetricsPublisher.push("sms_metric_B_id", data); // this won't be sent
|
|
|
|
|
|
|
|
// Reset the blacklist
|
|
|
|
// (msg #5)
|
|
|
|
cobraMetricsPublisher.setBlacklist({}); // 4.
|
|
|
|
|
|
|
|
// (3) Restrict what is sent using rate control
|
|
|
|
|
|
|
|
// (msg #6)
|
|
|
|
cobraMetricsPublisher.setRateControl({
|
2019-09-23 19:25:23 +02:00
|
|
|
{"sms_metric_C_id", 1}, // published once per minute (60 seconds) max
|
2019-09-06 06:57:05 +02:00
|
|
|
});
|
|
|
|
// (msg #7)
|
|
|
|
cobraMetricsPublisher.push("sms_metric_C_id", data);
|
|
|
|
cobraMetricsPublisher.push("sms_metric_C_id", data); // this won't be sent
|
|
|
|
|
|
|
|
ix::msleep(1400);
|
|
|
|
|
|
|
|
// (msg #8)
|
|
|
|
cobraMetricsPublisher.push("sms_metric_C_id", data); // now this will be sent
|
|
|
|
|
|
|
|
ix::msleep(600); // wait a bit so that the last message is sent and can be received
|
|
|
|
|
|
|
|
log("Testing suspend/resume now, which will disconnect the cobraMetricsPublisher.");
|
|
|
|
|
|
|
|
// Test suspend + resume
|
2019-09-23 19:25:23 +02:00
|
|
|
for (int i = 0; i < 3; ++i)
|
2019-09-06 06:57:05 +02:00
|
|
|
{
|
|
|
|
cobraMetricsPublisher.suspend();
|
|
|
|
ix::msleep(500);
|
|
|
|
REQUIRE(!cobraMetricsPublisher.isConnected()); // Check that we are not connected anymore
|
|
|
|
|
|
|
|
cobraMetricsPublisher.push("sms_metric_D_id", data); // will not be sent this time
|
|
|
|
|
|
|
|
cobraMetricsPublisher.resume();
|
2019-09-23 19:25:23 +02:00
|
|
|
ix::msleep(2000); // give cobra 2s to connect
|
2019-09-06 06:57:05 +02:00
|
|
|
REQUIRE(cobraMetricsPublisher.isConnected()); // Check that we are connected now
|
|
|
|
|
|
|
|
cobraMetricsPublisher.push("sms_metric_E_id", data);
|
|
|
|
}
|
|
|
|
|
|
|
|
ix::msleep(500);
|
|
|
|
|
2019-09-24 20:46:54 +02:00
|
|
|
// Test multi-threaded publish
|
|
|
|
std::thread bgPublisher1(&runAdditionalPublisher, &cobraMetricsPublisher);
|
|
|
|
std::thread bgPublisher2(&runAdditionalPublisher, &cobraMetricsPublisher);
|
|
|
|
std::thread bgPublisher3(&runAdditionalPublisher, &cobraMetricsPublisher);
|
|
|
|
std::thread bgPublisher4(&runAdditionalPublisher, &cobraMetricsPublisher);
|
|
|
|
std::thread bgPublisher5(&runAdditionalPublisher, &cobraMetricsPublisher);
|
|
|
|
|
|
|
|
bgPublisher1.join();
|
|
|
|
bgPublisher2.join();
|
|
|
|
bgPublisher3.join();
|
|
|
|
bgPublisher4.join();
|
|
|
|
bgPublisher5.join();
|
|
|
|
|
2019-09-06 06:57:05 +02:00
|
|
|
// Now stop the thread
|
|
|
|
gStop = true;
|
2020-03-20 20:21:45 +01:00
|
|
|
subscriberThread.join();
|
2019-09-06 06:57:05 +02:00
|
|
|
|
|
|
|
//
|
|
|
|
// Validate that we received all message kinds, and the correct number of messages
|
|
|
|
//
|
|
|
|
CHECK(gIds.count("sms_metric_A_id") == 1);
|
|
|
|
CHECK(gIds.count("sms_metric_B_id") == 1);
|
|
|
|
CHECK(gIds.count("sms_metric_C_id") == 1);
|
|
|
|
CHECK(gIds.count("sms_metric_D_id") == 1);
|
|
|
|
CHECK(gIds.count("sms_metric_E_id") == 1);
|
2019-09-24 20:46:54 +02:00
|
|
|
CHECK(gIds.count("sms_metric_F_id") == 1);
|
2019-09-06 06:57:05 +02:00
|
|
|
CHECK(gIds.count("sms_set_rate_control_id") == 1);
|
|
|
|
CHECK(gIds.count("sms_set_blacklist_id") == 1);
|
|
|
|
|
2019-09-24 06:51:55 +02:00
|
|
|
spdlog::info("Incoming bytes {}", incomingBytes);
|
|
|
|
spdlog::info("Outgoing bytes {}", outgoingBytes);
|
2019-09-24 06:04:01 +02:00
|
|
|
|
2019-09-24 06:51:55 +02:00
|
|
|
spdlog::info("Stopping snake server...");
|
2019-09-06 06:57:05 +02:00
|
|
|
snakeServer.stop();
|
2019-09-24 06:04:01 +02:00
|
|
|
|
2019-09-24 06:51:55 +02:00
|
|
|
spdlog::info("Stopping redis server...");
|
2019-09-24 06:04:01 +02:00
|
|
|
redisServer.stop();
|
2019-09-06 06:57:05 +02:00
|
|
|
}
|