such-gitea/wownero-lws
1
0
Fork 1
mirror of https://codeberg.org/wownero/wownero-lws synced 2026-01-09 23:25:16 -08:00
Code Issues Packages Projects Releases Wiki Activity

Switch from epee http client to boost::beast. All HTTP now non-blocking. (#150)

Browse Source
This commit is contained in:
Lee *!* Clagett
2024-12-04 17:25:07 -05:00
committed by Lee *!* Clagett
parent 66b7497a34
commit 29358f1323
22 changed files with 1095 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
src/rpc/client.cpp
View File

@@ -42,7 +42,7 @@
#include "net/zmq_async.h"
#include "scanner.h"
#include "serialization/json_object.h" // monero/src
#include "wire/msgpack.h"
#include "wire/json/write.h"
#if MLWS_RMQ_ENABLED
#include <amqp.h>
#include <amqp_tcp_socket.h>
@@ -50,6 +50,15 @@
namespace lws
{
// Not in `rates.h` - defaulting to JSON output seems odd
std::ostream& operator<<(std::ostream& out, lws::rates const& src)
{
wire::json_stream_writer dest{out};
lws::write_bytes(dest, src);
dest.finish();
return out;
}
namespace rpc
{
namespace http = epee::net_utils::http;
@@ -194,17 +203,16 @@ namespace rpc
, rmq(std::move(rmq))
, daemon_addr(std::move(daemon_addr))
, sub_addr(std::move(sub_addr))
, rates_conn()
, rates_conn(epee::net_utils::ssl_verification_t::system_ca)
, cache_time()
, cache_interval(interval)
, cached{}
, account_counter(0)
, sync_pub()
, sync_rates()
, untrusted_daemon(untrusted_daemon)
, rates_running()
{
if (std::chrono::minutes{0} < cache_interval)
rates_conn.set_server(crypto_compare.host, boost::none, epee::net_utils::ssl_support_t::e_ssl_support_enabled);
rates_running.clear();
}
zcontext comm;
@@ -213,14 +221,14 @@ namespace rpc
rcontext rmq;
const std::string daemon_addr;
const std::string sub_addr;
http::http_simple_client rates_conn;
net::http::client rates_conn;
std::chrono::steady_clock::time_point cache_time;
const std::chrono::minutes cache_interval;
rates cached;
std::atomic<unsigned> account_counter;
boost::mutex sync_pub;
boost::mutex sync_rates;
const bool untrusted_daemon;
std::atomic_flag rates_running;
};
} // detail
@@ -597,37 +605,44 @@ namespace rpc
return do_signal(ctx->signal_pub.get(), abort_process_signal);
}
expect<boost::optional<lws::rates>> context::retrieve_rates()
expect<void> context::retrieve_rates_async(boost::asio::io_context& io)
{
MONERO_PRECOND(ctx != nullptr);
if (ctx->cache_interval <= std::chrono::minutes{0})
return boost::make_optional(false, ctx->cached);
return success();
const auto now = std::chrono::steady_clock::now();
if (now - ctx->cache_time < ctx->cache_interval)
return boost::make_optional(false, ctx->cached);
if (ctx->rates_running.test_and_set())
return success();
expect<rates> fresh{lws::error::exchange_rates_fetch};
auto& self = ctx;
const expect<void> rc = ctx->rates_conn.get_async(
io, crypto_compare.url, [self] (boost::system::error_code error, std::string body)
{
expect<rates> fresh{lws::error::exchange_rates_fetch};
if (!error)
{
fresh = crypto_compare(std::move(body));
if (fresh)
MINFO("Updated exchange rates: " << *fresh);
else
MERROR("Failed to parse exchange rates: " << fresh.error());
}
else
MERROR("Failed to retrieve exchange rates: " << error.message());
const http::http_response_info* info = nullptr;
const bool retrieved =
ctx->rates_conn.invoke_get(crypto_compare.path, std::chrono::seconds{20}, std::string{}, std::addressof(info)) &&
info != nullptr &&
info->m_response_code == 200;
// \TODO Remove copy below
if (retrieved)
fresh = crypto_compare(std::string{info->m_body});
const boost::unique_lock<boost::mutex> lock{ctx->sync_rates};
ctx->cache_time = now;
if (fresh)
{
ctx->cached = *fresh;
return boost::make_optional(*fresh);
}
return fresh.error();
const auto now = std::chrono::steady_clock::now();
if (fresh)
{
const boost::lock_guard<boost::mutex> lock{self->sync_rates};
self->cache_time = now;
self->cached = std::move(*fresh);
}
self->rates_running.clear();
});
if (!rc)
ctx->rates_running.clear();
return rc;
}
} // rpc
} // lws

15
src/rpc/client.h
View File

@@ -148,10 +148,10 @@ namespace rpc
*/
expect<void> send(epee::byte_slice message, std::chrono::seconds timeout) noexcept;
//! Publish `payload` to ZMQ external pub socket.
//! Publish `payload` to ZMQ external pub socket. Blocks iff RMQ.
expect<void> publish(epee::byte_slice payload) const;
//! Publish `data` after `topic` to ZMQ external pub socket.
//! Publish `data` after `topic` to ZMQ external pub socket. Blocks iff RMQ.
template<typename F, typename T>
expect<void> publish(const boost::string_ref topic, const T& data) const
{
@@ -250,14 +250,11 @@ namespace rpc
expect<void> raise_abort_process() noexcept;
/*!
Retrieve exchange rates, if enabled and past cache interval. Not
thread-safe (this can be invoked from one thread only, but this is
thread-safe with `client::get_rates()`). All clients will see new rates
immediately.
Retrieve exchange rates, if enabled. Thread-safe. All clients will see
new rates once completed.
\return Rates iff they were updated.
*/
expect<boost::optional<lws::rates>> retrieve_rates();
\return `success()` if HTTP GET was queued. */
expect<void> retrieve_rates_async(boost::asio::io_context& io);
};
} // rpc
} // lws

8
src/rpc/rates.cpp
View File

@@ -65,10 +65,10 @@ namespace lws
namespace rpc
{
const char crypto_compare_::host[] = "https://min-api.cryptocompare.com:443";
const char crypto_compare_::path[] =
"/data/price?fsym=XMR&tsyms=AUD,BRL,BTC,CAD,CHF,CNY,EUR,GBP,"
"HKD,INR,JPY,KRW,MXN,NOK,NZD,SEK,SGD,TRY,USD,RUB,ZAR";
const char crypto_compare_::url[] =
"https://min-api.cryptocompare.com"
"/data/price?fsym=XMR&tsyms=AUD,BRL,BTC,CAD,CHF,CNY,EUR,GBP,"
"HKD,INR,JPY,KRW,MXN,NOK,NZD,SEK,SGD,TRY,USD,RUB,ZAR";
expect<lws::rates> crypto_compare_::operator()(std::string&& body) const
{

3
src/rpc/rates.h
View File

@@ -64,8 +64,7 @@ namespace lws
{
struct crypto_compare_
{
static const char host[];
static const char path[];
static const char url[];
expect<lws::rates> operator()(std::string&& body) const;
};

7
src/rpc/scanner/server.cpp
View File

@@ -458,7 +458,7 @@ namespace lws { namespace rpc { namespace scanner
};
}
server::server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, ssl_verification_t webhook_verify)
server::server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl)
: strand_(io),
check_timer_(io),
acceptor_(io),
@@ -467,11 +467,11 @@ namespace lws { namespace rpc { namespace scanner
active_(std::move(active)),
disk_(std::move(disk)),
zclient_(std::move(zclient)),
webhook_(std::move(ssl)),
accounts_cur_{},
next_thread_(0),
pass_hashed_(),
pass_salt_(),
webhook_verify_(webhook_verify),
stop_(false)
{
std::sort(active_.begin(), active_.end());
@@ -563,8 +563,7 @@ namespace lws { namespace rpc { namespace scanner
self->strand_,
[self, users = std::move(users), blocks = std::move(blocks)] ()
{
const lws::scanner_options opts{self->webhook_verify_, false, false};
if (!lws::user_data::store(self->disk_, self->zclient_, epee::to_span(blocks), epee::to_span(users), nullptr, opts))
if (!lws::user_data::store(self->strand_.context(), self->disk_, self->zclient_, self->webhook_, epee::to_span(blocks), epee::to_span(users), nullptr))
{
self->do_stop();
self->strand_.context().stop();

5
src/rpc/scanner/server.h
View File

@@ -38,6 +38,7 @@
#include "db/fwd.h"
#include "db/storage.h"
#include "net/http/client.h"
#include "net/net_ssl.h" // monero/contrib/epee/include
#include "rpc/client.h"
#include "rpc/scanner/queue.h"
@@ -65,11 +66,11 @@ namespace lws { namespace rpc { namespace scanner
std::vector<db::account_id> active_;
db::storage disk_;
rpc::client zclient_;
net::http::client webhook_;
db::cursor::accounts accounts_cur_;
std::size_t next_thread_;
std::array<unsigned char, 32> pass_hashed_;
std::array<unsigned char, crypto_pwhash_SALTBYTES> pass_salt_;
const ssl_verification_t webhook_verify_;
bool stop_;
//! Async acceptor routine
@@ -85,7 +86,7 @@ namespace lws { namespace rpc { namespace scanner
public:
static boost::asio::ip::tcp::endpoint get_endpoint(const std::string& address);
explicit server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, ssl_verification_t webhook_verify);
explicit server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl);
server(const server&) = delete;
server(server&&) = delete;

192
src/rpc/webhook.h
View File

@@ -27,161 +27,103 @@
#pragma once
#include <boost/asio/io_context.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/utility/string_ref.hpp>
#include <chrono>
#include <string>
#include "byte_slice.h" // monero/contrib/epee/include
#include "misc_log_ex.h" // monero/contrib/epee/include
#include "net/http_client.h" // monero/contrib/epee/include
#include "net/http/client.h"
#include "span.h"
#include "wire/json.h"
#include "wire/msgpack.h"
namespace lws { namespace rpc
{
namespace net = epee::net_utils;
template<typename T>
void http_send(net::http::http_simple_client& client, boost::string_ref uri, const T& event, const net::http::fields_list& params, const std::chrono::milliseconds timeout)
template<typename T>
void http_async(boost::asio::io_context& io, net::http::client& client, const epee::span<const T> events)
{
for (const auto& event : events)
{
if (uri.empty())
uri = "/";
epee::byte_slice bytes{};
const std::string& url = event.value.second.url;
const std::error_code json_error = wire::json::to_bytes(bytes, event);
const net::http::http_response_info* info = nullptr;
if (json_error)
if (event.value.second.url != "zmq")
{
MERROR("Failed to generate webhook JSON: " << json_error.message());
return;
}
MINFO("Sending webhook to " << url);
if (!client.invoke(uri, "POST", std::string{bytes.begin(), bytes.end()}, timeout, std::addressof(info), params))
{
MERROR("Failed to invoke http request to " << url);
return;
}
if (!info)
{
MERROR("Failed to invoke http request to " << url << ", internal error (null response ptr)");
return;
}
if (info->m_response_code != 200 && info->m_response_code != 201)
{
MERROR("Failed to invoke http request to " << url << ", wrong response code: " << info->m_response_code);
return;
epee::byte_slice bytes{};
const std::error_code json_error = wire::json::to_bytes(bytes, event);
if (!json_error)
{
MINFO("Sending webhook to " << event.value.second.url);
const expect<void> rc =
client.post_async(io, event.value.second.url, std::move(bytes));
if (!rc)
MERROR("Failed to send HTTP webhook to " << event.value.second.url << ": " << rc.error().message());
}
else
MERROR("Failed to generate webhook JSON: " << json_error.message());
}
}
}
template<typename T>
void http_send(const epee::span<const T> events, const std::chrono::milliseconds timeout, net::ssl_verification_t verify_mode)
template<typename T>
struct zmq_index_single
{
const std::uint64_t index;
const T& event;
};
template<typename T>
void write_bytes(wire::writer& dest, const zmq_index_single<T>& self)
{
wire::object(dest, WIRE_FIELD(index), WIRE_FIELD(event));
}
template<typename T>
void zmq_send(const rpc::client& client, const epee::span<const T> events, const boost::string_ref json_topic, const boost::string_ref msgpack_topic)
{
// Each `T` should have a unique count. This is desired.
struct zmq_order
{
if (events.empty())
return;
std::uint64_t current;
boost::mutex sync;
net::http::url_content url{};
net::http::http_simple_client client{};
zmq_order()
: current(0), sync()
{}
};
net::http::fields_list params;
params.emplace_back("Content-Type", "application/json; charset=utf-8");
static zmq_order ordering{};
//! \TODO monitor XPUB to cull the serialization
if (!events.empty() && client.has_publish())
{
// make sure the event is queued to zmq in order.
const boost::unique_lock<boost::mutex> guard{ordering.sync};
for (const auto& event : events)
{
if (event.value.second.url.empty() || !net::parse_url(event.value.second.url, url))
{
MERROR("Bad URL for webhook event: " << event.value.second.url);
continue;
}
const bool https = (url.schema == "https");
if (!https && url.schema != "http")
{
MERROR("Only http or https connections: " << event.value.second.url);
continue;
}
const net::ssl_support_t ssl_mode = https ?
net::ssl_support_t::e_ssl_support_enabled : net::ssl_support_t::e_ssl_support_disabled;
net::ssl_options_t ssl_options{ssl_mode};
if (https)
ssl_options.verification = verify_mode;
if (url.port == 0)
url.port = https ? 443 : 80;
client.set_server(url.host, std::to_string(url.port), boost::none, std::move(ssl_options));
if (client.connect(timeout))
http_send(client, url.uri, event, params, timeout);
else
MERROR("Unable to send webhook to " << event.value.second.url);
client.disconnect();
const zmq_index_single<T> index{ordering.current++, event};
MINFO("Sending ZMQ-PUB topics " << json_topic << " and " << msgpack_topic);
expect<void> result = success();
if (!(result = client.publish<wire::json>(json_topic, index)))
MERROR("Failed to serialize+send " << json_topic << " " << result.error().message());
if (!(result = client.publish<wire::msgpack>(msgpack_topic, index)))
MERROR("Failed to serialize+send " << msgpack_topic << " " << result.error().message());
}
}
}
template<typename T>
struct zmq_index_single
{
const std::uint64_t index;
const T& event;
};
template<typename T>
void write_bytes(wire::writer& dest, const zmq_index_single<T>& self)
{
wire::object(dest, WIRE_FIELD(index), WIRE_FIELD(event));
}
template<typename T>
void zmq_send(const rpc::client& client, const epee::span<const T> events, const boost::string_ref json_topic, const boost::string_ref msgpack_topic)
{
// Each `T` should have a unique count. This is desired.
struct zmq_order
{
std::uint64_t current;
boost::mutex sync;
zmq_order()
: current(0), sync()
{}
};
static zmq_order ordering{};
//! \TODO monitor XPUB to cull the serialization
if (!events.empty() && client.has_publish())
{
// make sure the event is queued to zmq in order.
const boost::unique_lock<boost::mutex> guard{ordering.sync};
for (const auto& event : events)
{
const zmq_index_single<T> index{ordering.current++, event};
MINFO("Sending ZMQ-PUB topics " << json_topic << " and " << msgpack_topic);
expect<void> result = success();
if (!(result = client.publish<wire::json>(json_topic, index)))
MERROR("Failed to serialize+send " << json_topic << " " << result.error().message());
if (!(result = client.publish<wire::msgpack>(msgpack_topic, index)))
MERROR("Failed to serialize+send " << msgpack_topic << " " << result.error().message());
}
}
}
template<typename T>
void send_webhook(
template<typename T>
void send_webhook_async(
boost::asio::io_context& io,
const rpc::client& client,
net::http::client& http_client,
const epee::span<const T> events,
const boost::string_ref json_topic,
const boost::string_ref msgpack_topic,
const std::chrono::seconds timeout,
epee::net_utils::ssl_verification_t verify_mode)
const boost::string_ref msgpack_topic)
{
http_send(events, timeout, verify_mode);
http_async(io, http_client, events);
// ZMQ PUB sockets never block, but RMQ does. No easy way around this.
zmq_send(client, events, json_topic, msgpack_topic);
}
}} // lws // rpc