Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[WIP] Refactor, stage 2 #113

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

[WIP] Refactor, stage 2 #113

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 59 additions & 2 deletions include/irods/private/amqp_sender.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,11 @@

#include "irods/private/audit_amqp.hpp"

#include <cstdint>
#include <fstream>
#include <string>
#include <string_view>
#include <vector>

#include <proton/connection.hpp>
#include <proton/connection_options.hpp>
Expand All @@ -16,12 +20,49 @@
#include <proton/sender.hpp>
#include <proton/session.hpp>

#include <fmt/core.h>
#include <fmt/compile.h>
#include <nlohmann/json.hpp>

// filesystem
// clang-format off
#ifdef __cpp_lib_filesystem
#include <filesystem>
namespace fs = std::filesystem;
#else
#include <boost/filesystem.hpp>
namespace fs = boost::filesystem;
#endif
Comment on lines +29 to +35
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just curious: Is this feature check for purposes of portability?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay. It made me wonder to what extent we wanted to do things like this. For instance, if this is to support older compilers, do we need to do something similar for, i.e. string_view (>=C++17)? I don't have a problem with this one, was just wondering where we're drawing this particular line, and/or if I missed the point. :)

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's less about the compiler and more about the stdlib. If you think it's unlikely that someone would compile against a stdlib that would have string_view and not std::filesystem I can remove this

// clang-format on

namespace irods::plugin::rule_engine::audit_amqp
{
class amqp_endpoint
{
public:
amqp_endpoint(
const std::string_view& _scheme,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason these std::string_views are refs?

const std::string_view& _host,
const std::uint_fast16_t& _port,
const std::string_view& _topic,
const std::string_view& _user,
const std::string_view& _password);

const std::string url;
const std::string user;
const std::string password;

static constexpr const std::string_view default_scheme{"amqp"};
static constexpr const std::uint_fast16_t default_amqp_port = 5672;
static constexpr const std::uint_fast16_t default_amqps_port = 5671;
static constexpr const std::string_view default_user{""};
static constexpr const std::string_view default_password{""};
};

class send_handler : public proton::messaging_handler
{
public:
send_handler(const proton::message& _message, const std::string& _url);
send_handler(const proton::message& _message, const std::vector<amqp_endpoint>& _endpoints);
void on_container_start(proton::container& _container) override;
void on_sendable(proton::sender& _sender) override;
void on_tracker_accept(proton::tracker& _tracker) override;
Expand All @@ -34,10 +75,26 @@ namespace irods::plugin::rule_engine::audit_amqp
void on_error(const proton::error_condition& _err_cond) override;

private:
const std::string& _amqp_url;
const std::vector<amqp_endpoint>& _endpoints;
std::vector<amqp_endpoint>::size_type _endpoint_idx;
const proton::message& _message;
bool _message_sent;
};

class amqp_sender
{
public:
amqp_sender(const std::vector<amqp_endpoint>& _endpoints);
~amqp_sender();
void enable_test_mode(const fs::path _test_mode_log_path);
void disable_test_mode();
void send_message(const nlohmann::json& _message_body, const std::uint64_t _timestamp_ms);
private:
const std::vector<amqp_endpoint>& _endpoints;
std::vector<amqp_endpoint>::size_type _endpoint_idx;
fs::path _log_file_path;
std::ofstream _log_file_ofstream;
};
} //namespace irods::plugin::rule_engine::audit_amqp

#endif // IRODS_AUDIT_AMQP_SENDER_HPP
85 changes: 82 additions & 3 deletions src/amqp_sender.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,23 @@
#include "irods/private/amqp_sender.hpp"

#include <boost/config.hpp>
#include <fmt/core.h>
#include <fmt/compile.h>

namespace irods::plugin::rule_engine::audit_amqp
{
namespace
{
BOOST_FORCEINLINE std::string build_amqp_url(
const std::string_view& _scheme,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we be using refs with std::string_view?

const std::string_view& _host,
const std::uint_fast16_t& _port,
const std::string_view& _topic)
{
// TODO: use Boost.URL (if available at compile-time) to validate URLs
return fmt::format(FMT_COMPILE("{0:s}://{1:s}:{2:d}/{3:s}"), _scheme, _host, _port, _topic);
}

BOOST_FORCEINLINE void log_proton_error(const proton::error_condition& err_cond, const std::string& log_message)
{
// clang-format off
Expand All @@ -21,17 +33,84 @@ namespace irods::plugin::rule_engine::audit_amqp
}
} // namespace

send_handler::send_handler(const proton::message& _message, const std::string& _url)
: _amqp_url(_url)
amqp_endpoint::amqp_endpoint(
const std::string_view& _scheme,
const std::string_view& _host,
const std::uint_fast16_t& _port,
const std::string_view& _topic,
const std::string_view& _user,
const std::string_view& _password)
: url(build_amqp_url(_scheme, _host, _port, _topic))
, user(_user)
, password(_password)
{
}

amqp_sender::amqp_sender(const std::vector<amqp_endpoint>& _endpoints)
: _endpoints(_endpoints), _endpoint_idx(static_cast<std::vector<amqp_endpoint>::size_type>(0))
{
}

amqp_sender::~amqp_sender()
{
if (_log_file_ofstream.is_open()) {
_log_file_ofstream.close();
}
}

void amqp_sender::enable_test_mode(const fs::path _test_mode_log_path)
{
if (_log_file_ofstream.is_open()) {
_log_file_ofstream.close();
}
_log_file_path.assign(_test_mode_log_path);
}

void amqp_sender::disable_test_mode()
{
_log_file_path.clear();
if (_log_file_ofstream.is_open()) {
_log_file_ofstream.close();
}
}

void amqp_sender::send_message(const nlohmann::json& _message_body, const std::uint64_t _timestamp_ms)
{
const std::string msg_str = _message_body.dump();

proton::message msg(msg_str);
msg.content_type("application/json");
msg.creation_time(proton::timestamp(static_cast<proton::timestamp::numeric_type>(_timestamp_ms)));
send_handler handler(msg, _endpoints);
proton::container(handler).run();

if (!_log_file_path.empty()) {
if (!_log_file_ofstream.is_open()) {
_log_file_ofstream.open(_log_file_path);
}
_log_file_ofstream << msg_str << std::endl;
}
}

send_handler::send_handler(const proton::message& _message, const std::vector<amqp_endpoint>& _endpoints)
: _endpoints(_endpoints)
, _endpoint_idx(static_cast<std::vector<amqp_endpoint>::size_type>(0))
, _message(_message)
, _message_sent(false)
{
}

void send_handler::on_container_start(proton::container& _container)
{
const amqp_endpoint& endpoint = _endpoints[_endpoint_idx];
proton::connection_options conn_opts;
_container.open_sender(_amqp_url, conn_opts);
if (!endpoint.user.empty()) {
conn_opts.user(endpoint.user);
}
if (!endpoint.password.empty()) {
conn_opts.password(endpoint.password);
}
_container.open_sender(endpoint.url, conn_opts);
}

void send_handler::on_sendable(proton::sender& _sender)
Expand Down
Loading