glguy/xbot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

restructure the "threads"

Browse Source
This commit is contained in:
Eric Mertens 2023-11-27 14:12:20 -08:00
parent 7c6ef18a6c
commit e23fb33d89
19 changed files with 266 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -39,7 +39,7 @@ add_custom_command(
add_executable(xbot
main.cpp irc_commands.inc ircmsg.cpp settings.cpp connection.cpp
thread.cpp snote_thread.cpp watchdog_thread.cpp write_irc.cpp
snote_thread.cpp watchdog_thread.cpp write_irc.cpp
ping_thread.cpp irc_parse_thread.cpp registration_thread.cpp
self_thread.cpp command_thread.cpp)
target_include_directories(xbot PRIVATE ${CMAKE_CURRENT_BINARY_DIR})

2
command_thread.hpp
View File

@ -1,6 +1,6 @@
#pragma once
#include "thread.hpp"
#include "event.hpp"
#include <string>

2
connection.cpp
View File

@ -81,6 +81,7 @@ auto Connection::connect(
break;
}
buffer.add_bytes(n, [this](char * line) {
std::cout << "RECV: " << line << std::endl;
make_event<LineEvent>(line);
});
}
@ -90,6 +91,7 @@ auto Connection::connect(
auto Connection::write_line(std::string message) -> void
{
std::cout << "SEND: " << message << std::endl;
message += "\r\n";
auto const need_cancel = write_strings_.empty();
write_strings_.push_back(std::move(message));

2
connection.hpp
View File

@ -1,8 +1,8 @@
#pragma once
#include "event.hpp"
#include "linebuffer.hpp"
#include "settings.hpp"
#include "thread.hpp"
#include <eventpp/eventdispatcher.h>
#include <eventpp/utilities/argumentadapter.h>

2
thread.hpp → event.hpp
View File

@ -6,6 +6,6 @@
#include <vector>
struct Event {
virtual ~Event() {}
virtual ~Event() = default;
bool handled_ = false;
};

2
irc_commands.gperf
View File

@ -162,7 +162,7 @@ struct RecognizedCommand {
438, IrcCommand::ERR_NICKTOOFAST
440, IrcCommand::ERR_SERVICESDOWN
441, IrcCommand::ERR_USERNOTINCHANNEL
442, IrcCommand::ERR_NOTONCHANNEL
442, IrcCommand::ERR_NOTONCHANNEL, 3, 3
443, IrcCommand::ERR_USERONCHANNEL
444, IrcCommand::ERR_NOLOGIN
445, IrcCommand::ERR_SUMMONDISABLED

7
irc_parse_thread.cpp
View File

@ -10,7 +10,7 @@ namespace {
} // namespace
auto irc_parse_thread(Connection& connection) -> void
auto IrcParseThread::start(Connection& connection) -> void
{
connection.add_listener<LineEvent>([&connection](LineEvent const& event)
{
@ -21,6 +21,11 @@ auto irc_parse_thread(Connection& connection) -> void
&& recognized->min_args <= msg.args.size()
&& recognized->max_args >= msg.args.size()
? recognized->command : IrcCommand::UNKNOWN;
if (IrcCommand::UNKNOWN == command)
{
std::cout << "Unrecognized command: " << msg.command << " " << msg.args.size() << std::endl;
}
connection.make_event<IrcMsgEvent>(command, msg);
});
}

7
irc_parse_thread.hpp
View File

@ -1,6 +1,6 @@
#pragma once
#include "thread.hpp"
#include "event.hpp"
#include <string_view>
@ -293,4 +293,7 @@ struct IrcMsgEvent : Event
IrcMsg const& irc;
};
auto irc_parse_thread(Connection& connection) -> void;
struct IrcParseThread
{
static auto start(Connection& connection) -> void;
};

51
main.cpp
View File

@ -1,19 +1,19 @@
#include <boost/asio.hpp>
#include "connection.hpp"
#include "event.hpp"
#include "ircmsg.hpp"
#include "linebuffer.hpp"
#include "settings.hpp"
#include "thread.hpp"
#include "write_irc.hpp"
#include "irc_parse_thread.hpp"
#include "registration_thread.hpp"
#include "ping_thread.hpp"
#include "watchdog_thread.hpp"
#include "snote_thread.hpp"
#include "self_thread.hpp"
#include "command_thread.hpp"
#include "irc_parse_thread.hpp"
#include "ping_thread.hpp"
#include "registration_thread.hpp"
#include "self_thread.hpp"
#include "snote_thread.hpp"
#include "watchdog_thread.hpp"
#include <algorithm>
#include <chrono>
@ -34,23 +34,6 @@
using namespace std::chrono_literals;
auto unhandled_message_thread(Connection& connection) -> void
{
connection.add_listener<IrcMsgEvent>([](IrcMsgEvent const& event)
{
if (IrcCommand::UNKNOWN == event.command)
{
auto& irc = event.irc;
std::cout << "Unknown message " << irc.command;
for (auto const arg : irc.args)
{
std::cout << " " << arg;
}
std::cout << "\n";
}
});
}
auto echo_thread(Connection& connection) -> void
{
connection.add_listener<CommandEvent>([&connection](CommandEvent& event)
@ -59,9 +42,7 @@ auto echo_thread(Connection& connection) -> void
and "glguy" == event.oper
and "glguy" == event.account)
{
auto txt = std::string{event.arg};
txt += "\r\n";
connection.write_line(std::move(txt));
connection.write_line(std::string{event.arg});
event.handled_ = true;
send_notice(connection, event.nick, "ack");
}
@ -72,20 +53,14 @@ auto start(boost::asio::io_context & io, Settings const& settings) -> void
{
auto connection = std::make_shared<Connection>(io);
watchdog_thread(*connection);
irc_parse_thread(*connection);
ping_thread(*connection);
auto const self_thread = SelfThread::start(*connection);
registration_thread(*connection, settings.password, settings.username, settings.realname, settings.nickname);
WatchdogThread::start(*connection);
IrcParseThread::start(*connection);
PingThread::start(*connection);
SelfThread::start(*connection);
RegistrationThread::start(*connection, settings.password, settings.username, settings.realname, settings.nickname);
SnoteThread::start(*connection);
CommandThread::start(*connection);
echo_thread(*connection);
unhandled_message_thread(*connection);
connection->add_listener<SnoteEvent>([](SnoteEvent& event)
{
std::cout << "Snote match " << static_cast<int>(event.tag) << std::endl;
});
boost::asio::co_spawn(
io,

2
ping_thread.cpp
View File

@ -3,7 +3,7 @@
#include "irc_parse_thread.hpp"
#include "write_irc.hpp"
auto ping_thread(Connection& connection) -> void
auto PingThread::start(Connection& connection) -> void
{
connection.add_listener<IrcMsgEvent>([&connection](IrcMsgEvent& event)
{

7
ping_thread.hpp
View File

@ -1,6 +1,9 @@
#pragma once
#include "connection.hpp"
#include "thread.hpp"
#include "event.hpp"
auto ping_thread(Connection& connection) -> void;
struct PingThread
{
static auto start(Connection& connection) -> void;
};

210
registration_thread.cpp
View File

@ -4,48 +4,6 @@
#include <unordered_set>
#include <unordered_map>
namespace {
struct RegistrationThread : std::enable_shared_from_this<RegistrationThread>
{
Connection& connection_;
std::string password_;
std::string username_;
std::string realname_;
std::string nickname_;
std::unordered_map<std::string, std::string> caps;
std::unordered_set<std::string> outstanding;
Connection::Handle<ConnectEvent> connect_handle_;
Connection::Handle<IrcMsgEvent> message_handle_;
enum class Stage
{
LsReply,
AckReply,
};
Stage stage_;
RegistrationThread(
Connection& connection_,
std::string password,
std::string username,
std::string realname,
std::string nickname
);
auto on_connect() -> void;
auto send_req() -> void;
auto capack(IrcMsg const& msg) -> void;
auto capls(IrcMsg const& msg) -> void;
auto on_msg(IrcMsg const& msg) -> void;
};
RegistrationThread::RegistrationThread(
Connection& connection,
std::string password,
@ -58,8 +16,8 @@ RegistrationThread::RegistrationThread(
, username_{username}
, realname_{realname}
, nickname_{nickname}
, stage_{Stage::LsReply}
{}
{
}
auto RegistrationThread::on_connect() -> void
{
@ -100,117 +58,121 @@ auto RegistrationThread::send_req() -> void
outstanding.insert(cap);
}
}
connection_.remove_listener(message_handle_);
if (not outstanding.empty())
{
request.pop_back();
send_cap_req(connection_, request);
stage_ = Stage::AckReply;
listen_for_cap_ack();
}
else
{
send_cap_end(connection_);
}
}
auto RegistrationThread::on_msg_cap_ack(IrcMsg const& msg) -> void
{
auto in = std::istringstream{std::string{msg.args[2]}};
std::for_each(
std::istream_iterator<std::string>{in},
std::istream_iterator<std::string>{},
[this](std::string x) {
outstanding.erase(x);
}
);
if (outstanding.empty())
{
send_cap_end(connection_);
connection_.remove_listener(message_handle_);
}
}
auto RegistrationThread::capack(IrcMsg const& msg) -> void
auto RegistrationThread::on_msg_cap_ls(IrcMsg const& msg) -> void
{
auto const n = msg.args.size();
if (n >= 2 && "*" == msg.args[0] && "ACK" == msg.args[1])
std::string_view const* kvs;
bool last;
if (3 == msg.args.size())
{
auto in = std::istringstream{std::string{msg.args[2]}};
std::for_each(
std::istream_iterator<std::string>{in},
std::istream_iterator<std::string>{},
[this](std::string x) {
outstanding.erase(x);
kvs = &msg.args[2];
last = true;
}
else if (4 == msg.args.size() && "*" == msg.args[2])
{
kvs = &msg.args[3];
last = false;
}
else
{
return;
}
auto in = std::istringstream{std::string{*kvs}};
std::for_each(
std::istream_iterator<std::string>{in},
std::istream_iterator<std::string>{},
[this](std::string x) {
auto const eq = x.find('=');
if (eq == x.npos)
{
caps.emplace(x, std::string{});
}
);
if (outstanding.empty())
{
send_cap_end(connection_);
connection_.remove_listener(message_handle_);
}
}
}
auto RegistrationThread::capls(IrcMsg const& msg) -> void
{
auto const n = msg.args.size();
if (n >= 2 && "*" == msg.args[0] && "LS" == msg.args[1])
{
std::string_view const* kvs;
bool last;
if (3 == n)
{
kvs = &msg.args[2];
last = true;
}
else if (4 == n && "*" == msg.args[2])
{
kvs = &msg.args[3];
last = false;
}
else
{
return;
}
auto in = std::istringstream{std::string{*kvs}};
std::for_each(
std::istream_iterator<std::string>{in},
std::istream_iterator<std::string>{},
[this](std::string x) {
auto const eq = x.find('=');
if (eq == x.npos)
{
caps.emplace(x, std::string{});
}
else
{
caps.emplace(std::string{x, 0, eq}, std::string{x, eq+1, x.npos});
}
else
{
caps.emplace(std::string{x, 0, eq}, std::string{x, eq+1, x.npos});
}
);
if (last)
{
send_req();
}
}
}
);
auto RegistrationThread::on_msg(IrcMsg const& msg) -> void
{
switch (stage_)
if (last)
{
case Stage::LsReply: capls(msg); return;
case Stage::AckReply: capack(msg); return;
send_req();
}
}
} // namespace
auto registration_thread(
auto RegistrationThread::start(
Connection& connection,
std::string password,
std::string username,
std::string realname,
std::string nickname
) -> void
) -> std::shared_ptr<RegistrationThread>
{
auto const thread = std::make_shared<RegistrationThread>(connection, password, username, realname, nickname);
thread->message_handle_ = connection.add_listener<IrcMsgEvent>([thread](IrcMsgEvent const& event)
{
if (IrcCommand::CAP == event.command)
{
thread->on_msg(event.irc);
}
});
thread->listen_for_cap_ls();
thread->connect_handle_ = connection.add_listener<ConnectEvent>([thread](ConnectEvent const&)
{
thread->on_connect();
});
return thread;
}
auto RegistrationThread::listen_for_cap_ack() -> void
{
message_handle_ = connection_.add_listener<IrcMsgEvent>([thread = shared_from_this()](IrcMsgEvent const& event)
{
if (IrcCommand::CAP == event.command && event.irc.args.size() >= 2 && "*" == event.irc.args[0] && "ACK" == event.irc.args[1])
{
thread->on_msg_cap_ack(event.irc);
}
});
}
auto RegistrationThread::listen_for_cap_ls() -> void
{
message_handle_ = connection_.add_listener<IrcMsgEvent>([thread = shared_from_this()](IrcMsgEvent const& event)
{
if (IrcCommand::CAP == event.command && event.irc.args.size() >= 2 && "*" == event.irc.args[0] && "LS" == event.irc.args[1])
{
thread->on_msg_cap_ls(event.irc);
}
});
}

51
registration_thread.hpp
View File

@ -1,18 +1,53 @@
#pragma once
#include "thread.hpp"
#include "connection.hpp"
#include "event.hpp"
#include "irc_parse_thread.hpp"
#include "write_irc.hpp"
#include <eventpp/eventdispatcher.h>
#include <memory>
#include <string>
#include <unordered_map>
#include <unordered_set>
auto registration_thread(
Connection& connection,
std::string password,
std::string username,
std::string realname,
std::string nickname
) -> void;
class RegistrationThread : public std::enable_shared_from_this<RegistrationThread>
{
Connection& connection_;
std::string password_;
std::string username_;
std::string realname_;
std::string nickname_;
std::unordered_map<std::string, std::string> caps;
std::unordered_set<std::string> outstanding;
Connection::Handle<ConnectEvent> connect_handle_;
Connection::Handle<IrcMsgEvent> message_handle_;
auto on_connect() -> void;
auto send_req() -> void;
auto on_msg_cap_ls(IrcMsg const& msg) -> void;
auto on_msg_cap_ack(IrcMsg const& msg) -> void;
auto listen_for_cap_ack() -> void;
auto listen_for_cap_ls() -> void;
public:
RegistrationThread(
Connection& connection_,
std::string password,
std::string username,
std::string realname,
std::string nickname
);
static auto start(
Connection& connection,
std::string password,
std::string username,
std::string realname,
std::string nickname
) -> std::shared_ptr<RegistrationThread>;
};

1
self_thread.hpp
View File

@ -15,4 +15,3 @@ public:
SelfThread(Connection& connection) : connection_{connection} {}
static auto start(Connection&) -> std::shared_ptr<SelfThread>;
};

49
snote_thread.cpp
View File

@ -16,14 +16,12 @@ struct SnotePattern
SnotePattern(SnoteTag tag, char const* expression, unsigned flags = 0)
: tag{tag}
, expression{expression}
, flags{flags}
, regex{expression, std::regex_constants::ECMAScript | std::regex_constants::optimize}
{
}
SnoteTag tag;
char const* expression;
unsigned flags;
std::regex regex;
};
@ -34,31 +32,44 @@ SnotePattern const patterns[] =
{SnoteTag::ClientExiting,
R"(^Client exiting: ([^ ]+) \(([^@ ]+)@([^) ]+)\) \[(.*)\] \[(.*)\]$)"},
{SnoteTag::RejectingKlined,
R"(^Rejecting K-Lined user ([^ ]+)\[([^@]+)@([^\]]+)\] \[([^\] ]+)\] \((.*)\)$)"},
{SnoteTag::NickChange,
R"(^Nick change: From ([^ ]+) to ([^ ]+) \[([^@]+)@([^ ]+)\]$)"},
{SnoteTag::CreateChannel,
R"(^([^ ]+) is creating new channel ([^ ]+)$)"},
{SnoteTag::TemporaryKlineExpired,
R"(^Temporary K-line for \[([^ ]+)\] expired$)"},
{SnoteTag::DisconnectingKlined,
R"(^Disconnecting K-Lined user ([^ ]+)\[([^@]+)@([^ ]+)\] \((.*)\)$)"},
};
auto setup_database() -> std::unique_ptr<hs_database_t, SnoteThread::DbDeleter>
auto setup_database() -> hs_database_t*
{
auto const n = std::size(patterns);
std::vector<char const*> expressions;
std::vector<unsigned> flags;
std::vector<unsigned> flags(n, 0);
std::vector<unsigned> ids;
expressions.reserve(std::size(patterns));
flags.reserve(std::size(patterns));
ids.reserve(std::size(patterns));
unsigned id = 0;
for (auto const& pattern : patterns)
for (std::size_t i = 0; i < n; i++)
{
expressions.push_back(pattern.expression);
flags.push_back(pattern.flags);
ids.push_back(id++);
expressions.push_back(patterns[i].expression);
flags.push_back(0);
ids.push_back(i);
}
hs_database_t* db;
hs_compile_error *error;
switch (hs_compile_multi(expressions.data(), flags.data(), ids.data(), expressions.size(), HS_MODE_BLOCK, nullptr, &db, &error))
hs_platform_info_t *platform = nullptr; // target current platform
switch (hs_compile_multi(expressions.data(), flags.data(), ids.data(), expressions.size(), HS_MODE_BLOCK, platform, &db, &error))
{
case HS_COMPILER_ERROR:
{
@ -71,7 +82,7 @@ auto setup_database() -> std::unique_ptr<hs_database_t, SnoteThread::DbDeleter>
default:
abort();
}
return std::unique_ptr<hs_database_t, SnoteThread::DbDeleter>{db};
return db;
}
} // namespace
@ -80,14 +91,14 @@ auto SnoteThread::start(Connection& connection) -> std::shared_ptr<SnoteThread>
{
auto thread = std::make_shared<SnoteThread>();
thread->db_ = setup_database();
thread->db_.reset(setup_database());
hs_scratch_t* scratch = nullptr;
if (HS_SUCCESS != hs_alloc_scratch(thread->db_.get(), &scratch))
{
abort();
}
thread->scratch_ = std::unique_ptr<hs_scratch_t, ScratchDeleter>{scratch};
thread->scratch_.reset(scratch);
static char const* const prefix = "*** Notice -- ";
connection.add_listener<IrcMsgEvent>([&connection, thread](IrcMsgEvent& event)
@ -114,7 +125,11 @@ auto SnoteThread::start(Connection& connection) -> std::shared_ptr<SnoteThread>
abort();
}
if (match_id != -1)
if (match_id == -1)
{
std::cout << "Unknown snote: " << message << std::endl;
}
else
{
auto& pattern = patterns[match_id];
std::match_results<std::string_view::const_iterator> results;

7
snote_thread.hpp
View File

@ -1,6 +1,6 @@
#pragma once
#include "thread.hpp"
#include "event.hpp"
#include <hs.h>
@ -12,6 +12,11 @@ enum class SnoteTag
{
ClientConnecting,
ClientExiting,
RejectingKlined,
NickChange,
CreateChannel,
TemporaryKlineExpired,
DisconnectingKlined,
};
struct SnoteEvent : Event

1
thread.cpp
View File

@ -1 +0,0 @@
#include "thread.hpp"

89
watchdog_thread.cpp
View File

@ -11,71 +11,60 @@
using namespace std::chrono_literals;
namespace {
struct WatchdogThread : std::enable_shared_from_this<WatchdogThread>
WatchdogThread::WatchdogThread(Connection& connection)
: connection_{connection}
, timer_{connection.get_executor()}
, tried_ping{false}
{
WatchdogThread(Connection& connection)
: connection_{connection}
, timer_{connection.get_executor()}
, tried_ping{false}
{
}
}
Connection& connection_;
boost::asio::steady_timer timer_;
bool tried_ping;
auto WatchdogThread::on_activity() -> void
{
tried_ping = false;
timer_.expires_from_now(30s);
}
auto on_activity() -> void
auto WatchdogThread::timeout_token()
{
return [weak = weak_from_this()](auto const& error)
{
tried_ping = false;
timer_.expires_from_now(30s);
}
auto timeout_token()
{
return [weak = weak_from_this()](auto const& error)
if (not error)
{
if (not error)
if (auto self = weak.lock())
{
if (auto self = weak.lock())
{
self->on_timeout();
}
self->on_timeout();
}
};
}
auto on_timeout() -> void
{
if (tried_ping)
{
connection_.close();
}
else
{
send_ping(connection_, "watchdog");
tried_ping = true;
timer_.expires_from_now(30s);
timer_.async_wait(timeout_token());
}
}
};
}
auto on_connect() -> void
auto WatchdogThread::on_timeout() -> void
{
if (tried_ping)
{
on_activity();
connection_.close();
}
else
{
send_ping(connection_, "watchdog");
tried_ping = true;
timer_.expires_from_now(30s);
timer_.async_wait(timeout_token());
}
}
auto on_disconnect() -> void
{
timer_.cancel();
}
};
auto WatchdogThread::on_connect() -> void
{
on_activity();
timer_.async_wait(timeout_token());
}
} // namespace
auto WatchdogThread::on_disconnect() -> void
{
timer_.cancel();
}
auto watchdog_thread(Connection& connection) -> void
auto WatchdogThread::start(Connection& connection) -> void
{
auto const thread = std::make_shared<WatchdogThread>(connection);
connection.add_listener<ConnectEvent>([thread](auto&)

22
watchdog_thread.hpp
View File

@ -1,4 +1,24 @@
#pragma once
#include <boost/asio/steady_timer.hpp>
#include <memory>
class Connection;
auto watchdog_thread(Connection& connection) -> void;
class WatchdogThread : std::enable_shared_from_this<WatchdogThread>
{
Connection& connection_;
boost::asio::steady_timer timer_;
bool tried_ping;
auto on_activity() -> void;
auto timeout_token();
auto on_timeout() -> void;
auto on_connect() -> void;
auto on_disconnect() -> void;
public:
WatchdogThread(Connection& connection);
static auto start(Connection& connection) -> void;
};