chatterino2
Code Issues

changed to 80 max column

Browse Source
This commit is contained in:
fourtf
2018-08-06 21:17:03 +02:00
parent defa7e41fa
commit f71ff08e68
203 changed files with 3792 additions and 2405 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/providers/twitch/PubsubClient.cpp
+98 -64
View File
@@ -24,7 +24,8 @@ static std::map<QString, std::string> sentMessages;
namespace detail {
PubSubClient::PubSubClient(WebsocketClient &websocketClient, WebsocketHandle handle)
PubSubClient::PubSubClient(WebsocketClient &websocketClient,
WebsocketHandle handle)
: websocketClient_(websocketClient)
, handle_(handle)
{
@@ -58,7 +59,8 @@ bool PubSubClient::listen(rapidjson::Document &message)
this->numListens_ += numRequestedListens;
for (const auto &topic : message["data"]["topics"].GetArray()) {
this->listeners_.emplace_back(Listener{topic.GetString(), false, false, false});
this->listeners_.emplace_back(
Listener{topic.GetString(), false, false, false});
}
auto uuid = CreateUUID();
@@ -135,34 +137,38 @@ void PubSubClient::ping()
auto self = this->shared_from_this();
runAfter(this->websocketClient_.get_io_service(), std::chrono::seconds(15), [self](auto timer) {
if (!self->started_) {
return;
}
runAfter(this->websocketClient_.get_io_service(), std::chrono::seconds(15),
[self](auto timer) {
if (!self->started_) {
return;
}
if (self->awaitingPong_) {
Log("No pong respnose, disconnect!");
// TODO(pajlada): Label this connection as "disconnect me"
}
});
if (self->awaitingPong_) {
Log("No pong respnose, disconnect!");
// TODO(pajlada): Label this connection as "disconnect me"
}
});
runAfter(this->websocketClient_.get_io_service(), std::chrono::minutes(5), [self](auto timer) {
if (!self->started_) {
return;
}
runAfter(this->websocketClient_.get_io_service(), std::chrono::minutes(5),
[self](auto timer) {
if (!self->started_) {
return;
}
self->ping(); //
});
self->ping(); //
});
}
bool PubSubClient::send(const char *payload)
{
WebsocketErrorCode ec;
this->websocketClient_.send(this->handle_, payload, websocketpp::frame::opcode::text, ec);
this->websocketClient_.send(this->handle_, payload,
websocketpp::frame::opcode::text, ec);
if (ec) {
Log("Error sending message {}: {}", payload, ec.message());
// TODO(pajlada): Check which error code happened and maybe gracefully handle it
// TODO(pajlada): Check which error code happened and maybe gracefully
// handle it
return false;
}
@@ -176,13 +182,15 @@ PubSub::PubSub()
{
qDebug() << "init PubSub";
this->moderationActionHandlers["clear"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["clear"] = [this](const auto &data,
const auto &roomID) {
ClearChatAction action(data, roomID);
this->signals_.moderation.chatCleared.invoke(action);
};
this->moderationActionHandlers["slowoff"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["slowoff"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::Slow;
@@ -191,7 +199,8 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["slow"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["slow"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::Slow;
@@ -228,7 +237,8 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["r9kbetaoff"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["r9kbetaoff"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::R9K;
@@ -237,7 +247,8 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["r9kbeta"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["r9kbeta"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::R9K;
@@ -246,17 +257,18 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["subscribersoff"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
this->moderationActionHandlers["subscribersoff"] =
[this](const auto &data, const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::SubscribersOnly;
action.state = ModeChangedAction::State::Off;
action.mode = ModeChangedAction::Mode::SubscribersOnly;
action.state = ModeChangedAction::State::Off;
this->signals_.moderation.modeChanged.invoke(action);
};
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["subscribers"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["subscribers"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::SubscribersOnly;
@@ -265,16 +277,18 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["emoteonlyoff"] = [this](const auto &data, const auto &roomID) {
ModeChangedAction action(data, roomID);
this->moderationActionHandlers["emoteonlyoff"] =
[this](const auto &data, const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::EmoteOnly;
action.state = ModeChangedAction::State::Off;
action.mode = ModeChangedAction::Mode::EmoteOnly;
action.state = ModeChangedAction::State::Off;
this->signals_.moderation.modeChanged.invoke(action);
};
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["emoteonly"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["emoteonly"] = [this](const auto &data,
const auto &roomID) {
ModeChangedAction action(data, roomID);
action.mode = ModeChangedAction::Mode::EmoteOnly;
@@ -283,7 +297,8 @@ PubSub::PubSub()
this->signals_.moderation.modeChanged.invoke(action);
};
this->moderationActionHandlers["unmod"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["unmod"] = [this](const auto &data,
const auto &roomID) {
ModerationStateAction action(data, roomID);
getTargetUser(data, action.target);
@@ -307,7 +322,8 @@ PubSub::PubSub()
this->signals_.moderation.moderationStateChanged.invoke(action);
};
this->moderationActionHandlers["mod"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["mod"] = [this](const auto &data,
const auto &roomID) {
ModerationStateAction action(data, roomID);
getTargetUser(data, action.target);
@@ -331,7 +347,8 @@ PubSub::PubSub()
this->signals_.moderation.moderationStateChanged.invoke(action);
};
this->moderationActionHandlers["timeout"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["timeout"] = [this](const auto &data,
const auto &roomID) {
BanAction action(data, roomID);
getCreatedByUser(data, action.source);
@@ -367,7 +384,8 @@ PubSub::PubSub()
}
};
this->moderationActionHandlers["ban"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["ban"] = [this](const auto &data,
const auto &roomID) {
BanAction action(data, roomID);
getCreatedByUser(data, action.source);
@@ -396,7 +414,8 @@ PubSub::PubSub()
}
};
this->moderationActionHandlers["unban"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["unban"] = [this](const auto &data,
const auto &roomID) {
UnbanAction action(data, roomID);
getCreatedByUser(data, action.source);
@@ -421,7 +440,8 @@ PubSub::PubSub()
}
};
this->moderationActionHandlers["untimeout"] = [this](const auto &data, const auto &roomID) {
this->moderationActionHandlers["untimeout"] = [this](const auto &data,
const auto &roomID) {
UnbanAction action(data, roomID);
getCreatedByUser(data, action.source);
@@ -447,16 +467,21 @@ PubSub::PubSub()
};
this->websocketClient.set_access_channels(websocketpp::log::alevel::all);
this->websocketClient.clear_access_channels(websocketpp::log::alevel::frame_payload);
this->websocketClient.clear_access_channels(
websocketpp::log::alevel::frame_payload);
this->websocketClient.init_asio();
// SSL Handshake
this->websocketClient.set_tls_init_handler(bind(&PubSub::onTLSInit, this, ::_1));
this->websocketClient.set_tls_init_handler(
bind(&PubSub::onTLSInit, this, ::_1));
this->websocketClient.set_message_handler(bind(&PubSub::onMessage, this, ::_1, ::_2));
this->websocketClient.set_open_handler(bind(&PubSub::onConnectionOpen, this, ::_1));
this->websocketClient.set_close_handler(bind(&PubSub::onConnectionClose, this, ::_1));
this->websocketClient.set_message_handler(
bind(&PubSub::onMessage, this, ::_1, ::_2));
this->websocketClient.set_open_handler(
bind(&PubSub::onConnectionOpen, this, ::_1));
this->websocketClient.set_close_handler(
bind(&PubSub::onConnectionClose, this, ::_1));
// Add an initial client
this->addClient();
@@ -477,7 +502,8 @@ void PubSub::addClient()
void PubSub::start()
{
this->mainThread.reset(new std::thread(std::bind(&PubSub::runThread, this)));
this->mainThread.reset(
new std::thread(std::bind(&PubSub::runThread, this)));
}
void PubSub::listenToWhispers(std::shared_ptr<TwitchAccount> account)
@@ -507,8 +533,8 @@ void PubSub::unlistenAllModerationActions()
}
}
void PubSub::listenToChannelModerationActions(const QString &channelID,
std::shared_ptr<TwitchAccount> account)
void PubSub::listenToChannelModerationActions(
const QString &channelID, std::shared_ptr<TwitchAccount> account)
{
assert(!channelID.isEmpty());
assert(account != nullptr);
@@ -527,7 +553,8 @@ void PubSub::listenToChannelModerationActions(const QString &channelID,
this->listenToTopic(topic, account);
}
void PubSub::listenToTopic(const std::string &topic, std::shared_ptr<TwitchAccount> account)
void PubSub::listenToTopic(const std::string &topic,
std::shared_ptr<TwitchAccount> account)
{
auto message = createListenMessage({topic}, account);
@@ -542,7 +569,8 @@ void PubSub::listen(rapidjson::Document &&msg)
}
Log("Added to the back of the queue");
this->requests.emplace_back(std::make_unique<rapidjson::Document>(std::move(msg)));
this->requests.emplace_back(
std::make_unique<rapidjson::Document>(std::move(msg)));
}
bool PubSub::tryListen(rapidjson::Document &msg)
@@ -570,7 +598,8 @@ bool PubSub::isListeningToTopic(const std::string &topic)
return false;
}
void PubSub::onMessage(websocketpp::connection_hdl hdl, WebsocketMessagePtr websocketMessage)
void PubSub::onMessage(websocketpp::connection_hdl hdl,
WebsocketMessagePtr websocketMessage)
{
const std::string &payload = websocketMessage->get_payload();
@@ -585,7 +614,9 @@ void PubSub::onMessage(websocketpp::connection_hdl hdl, WebsocketMessagePtr webs
}
if (!msg.IsObject()) {
Log("Error parsing message '{}' from PubSub. Root object is not an object", payload);
Log("Error parsing message '{}' from PubSub. Root object is not an "
"object",
payload);
return;
}
@@ -615,8 +646,8 @@ void PubSub::onMessage(websocketpp::connection_hdl hdl, WebsocketMessagePtr webs
} else if (type == "PONG") {
auto clientIt = this->clients.find(hdl);
// If this assert goes off, there's something wrong with the connection creation/preserving
// code KKona
// If this assert goes off, there's something wrong with the connection
// creation/preserving code KKona
assert(clientIt != this->clients.end());
auto &client = *clientIt;
@@ -629,9 +660,11 @@ void PubSub::onMessage(websocketpp::connection_hdl hdl, WebsocketMessagePtr webs
void PubSub::onConnectionOpen(WebsocketHandle hdl)
{
auto client = std::make_shared<detail::PubSubClient>(this->websocketClient, hdl);
auto client =
std::make_shared<detail::PubSubClient>(this->websocketClient, hdl);
// We separate the starting from the constructor because we will want to use shared_from_this
// We separate the starting from the constructor because we will want to use
// shared_from_this
client->start();
this->clients.emplace(hdl, client);
@@ -643,8 +676,8 @@ void PubSub::onConnectionClose(WebsocketHandle hdl)
{
auto clientIt = this->clients.find(hdl);
// If this assert goes off, there's something wrong with the connection creation/preserving
// code KKona
// If this assert goes off, there's something wrong with the connection
// creation/preserving code KKona
assert(clientIt != this->clients.end());
auto &client = clientIt->second;
@@ -658,7 +691,8 @@ void PubSub::onConnectionClose(WebsocketHandle hdl)
PubSub::WebsocketContextPtr PubSub::onTLSInit(websocketpp::connection_hdl hdl)
{
WebsocketContextPtr ctx(new boost::asio::ssl::context(boost::asio::ssl::context::tlsv1));
WebsocketContextPtr ctx(
new boost::asio::ssl::context(boost::asio::ssl::context::tlsv1));
try {
ctx->set_options(boost::asio::ssl::context::default_workarounds |