ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-11-30 16:45:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cover LoadBalancer with tests (#1394)

Browse Source 
Fixes #680. Fixes #1222.
This commit is contained in:
Sergey Kuznetsov
2024-05-15 14:02:36 +01:00
committed by GitHub
parent f74b89cc8d
commit da10535bc0
61 changed files with 2769 additions and 1466 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/data/BackendInterface.cpp
View File

@@ -235,7 +235,8 @@ BackendInterface::fetchBookOffers(
LOG(gLog.debug()) << "Fetching " << std::to_string(keys.size()) << " offers took "
<< std::to_string(getMillis(mid - begin)) << " milliseconds. Fetching next dir took "
<< std::to_string(succMillis) << " milliseonds. Fetched next dir " << std::to_string(numSucc)
<< " times" << " Fetching next page of dir took " << std::to_string(pageMillis) << " milliseconds"
<< " times"
<< " Fetching next page of dir took " << std::to_string(pageMillis) << " milliseconds"
<< ". num pages = " << std::to_string(numPages) << ". Fetching all objects took "
<< std::to_string(getMillis(end - mid))
<< " milliseconds. total time = " << std::to_string(getMillis(end - begin)) << " milliseconds"

4
src/data/CassandraBackend.hpp
View File

@@ -792,8 +792,8 @@ public:
void
writeSuccessor(std::string&& key, std::uint32_t const seq, std::string&& successor) override
{
LOG(log_.trace()) << "Writing successor. key = " << key.size() << " bytes. " << " seq = " << std::to_string(seq)
<< " successor = " << successor.size() << " bytes.";
LOG(log_.trace()) << "Writing successor. key = " << key.size() << " bytes. "
<< " seq = " << std::to_string(seq) << " successor = " << successor.size() << " bytes.";
ASSERT(!key.empty(), "Key must not be empty");
ASSERT(!successor.empty(), "Successor must not be empty");

3
src/data/cassandra/impl/ExecutionStrategy.hpp
View File

@@ -484,7 +484,8 @@ private:
{
std::unique_lock<std::mutex> lck(throttleMutex_);
if (!canAddWriteRequest()) {
LOG(log_.trace()) << "Max outstanding requests reached. " << "Waiting for other requests to finish";
LOG(log_.trace()) << "Max outstanding requests reached. "
<< "Waiting for other requests to finish";
throttleCv_.wait(lck, [this]() { return canAddWriteRequest(); });
}
}

6
src/etl/CMakeLists.txt
View File

@@ -2,11 +2,13 @@ add_library(clio_etl)
target_sources(
clio_etl
PRIVATE NFTHelpers.cpp
PRIVATE CacheLoaderSettings.cpp
ETLHelpers.cpp
ETLService.cpp
ETLState.cpp
LoadBalancer.cpp
CacheLoaderSettings.cpp
NetworkValidatedLedgers.cpp
NFTHelpers.cpp
Source.cpp
impl/ForwardingCache.cpp
impl/ForwardingSource.cpp

46
src/etl/ETLHelpers.cpp Normal file
View File

@@ -0,0 +1,46 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include "etl/ETLHelpers.hpp"
#include "util/Assert.hpp"
#include <ripple/basics/base_uint.h>
#include <cstddef>
#include <vector>
namespace etl {
std::vector<ripple::uint256>
getMarkers(size_t numMarkers)
{
ASSERT(numMarkers <= 256, "Number of markers must be <= 256. Got: {}", numMarkers);
unsigned char const incr = 256 / numMarkers;
std::vector<ripple::uint256> markers;
markers.reserve(numMarkers);
ripple::uint256 base{0};
for (size_t i = 0; i < numMarkers; ++i) {
markers.push_back(base);
base.data()[0] += incr;
}
return markers;
}
} // namespace etl

97
src/etl/ETLHelpers.hpp
View File

@@ -20,11 +20,8 @@
/** @file */
#pragma once
#include "util/Assert.hpp"
#include <ripple/basics/base_uint.h>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
@@ -35,83 +32,6 @@
#include <vector>
namespace etl {
/**
* @brief This datastructure is used to keep track of the sequence of the most recent ledger validated by the network.
*
* There are two methods that will wait until certain conditions are met. This datastructure is able to be "stopped".
* When the datastructure is stopped, any threads currently waiting are unblocked.
* Any later calls to methods of this datastructure will not wait. Once the datastructure is stopped, the datastructure
* remains stopped for the rest of its lifetime.
*/
class NetworkValidatedLedgers {
// max sequence validated by network
std::optional<uint32_t> max_;
mutable std::mutex m_;
std::condition_variable cv_;
public:
/**
* @brief A factory function for NetworkValidatedLedgers
*
* @return A shared pointer to a new instance of NetworkValidatedLedgers
*/
static std::shared_ptr<NetworkValidatedLedgers>
make_ValidatedLedgers()
{
return std::make_shared<NetworkValidatedLedgers>();
}
/**
* @brief Notify the datastructure that idx has been validated by the network.
*
* @param idx Sequence validated by network
*/
void
push(uint32_t idx)
{
std::lock_guard const lck(m_);
if (!max_ || idx > *max_)
max_ = idx;
cv_.notify_all();
}
/**
* @brief Get most recently validated sequence.
*
* If no ledgers are known to have been validated, this function waits until the next ledger is validated
*
* @return Sequence of most recently validated ledger. empty optional if the datastructure has been stopped
*/
std::optional<uint32_t>
getMostRecent()
{
std::unique_lock lck(m_);
cv_.wait(lck, [this]() { return max_; });
return max_;
}
/**
* @brief Waits for the sequence to be validated by the network.
*
* @param sequence The sequence to wait for
* @param maxWaitMs Maximum time to wait for the sequence to be validated. If empty, wait indefinitely
* @return true if sequence was validated, false otherwise a return value of false means the datastructure has been
* stopped
*/
bool
waitUntilValidatedByNetwork(uint32_t sequence, std::optional<uint32_t> maxWaitMs = {})
{
std::unique_lock lck(m_);
auto pred = [sequence, this]() -> bool { return (max_ && sequence <= *max_); };
if (maxWaitMs) {
cv_.wait_for(lck, std::chrono::milliseconds(*maxWaitMs));
} else {
cv_.wait(lck, pred);
}
return pred();
}
};
// TODO: does the note make sense? lockfree queues provide the same blocking behaviour just without mutex, don't they?
/**
@@ -228,20 +148,7 @@ public:
* @param numMarkers Total markers to partition for
* @return The markers
*/
inline std::vector<ripple::uint256>
getMarkers(size_t numMarkers)
{
ASSERT(numMarkers <= 256, "Number of markers must be <= 256. Got: {}", numMarkers);
std::vector<ripple::uint256>
getMarkers(size_t numMarkers);
unsigned char const incr = 256 / numMarkers;
std::vector<ripple::uint256> markers;
markers.reserve(numMarkers);
ripple::uint256 base{0};
for (size_t i = 0; i < numMarkers; ++i) {
markers.push_back(base);
base.data()[0] += incr;
}
return markers;
}
} // namespace etl

6
src/etl/ETLService.cpp
View File

@@ -22,6 +22,8 @@
#include "data/BackendInterface.hpp"
#include "data/LedgerCache.hpp"
#include "etl/CorruptionDetector.hpp"
#include "etl/ETLHelpers.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Assert.hpp"
#include "util/Constants.hpp"
#include "util/config/Config.hpp"
@@ -263,9 +265,9 @@ ETLService::ETLService(
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<SubscriptionManagerType> subscriptions,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<LoadBalancerType> balancer,
std::shared_ptr<NetworkValidatedLedgersType> ledgers
std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
)
: backend_(backend)
, loadBalancer_(balancer)

18
src/etl/ETLService.hpp
View File

@@ -33,7 +33,7 @@
#include "etl/impl/LedgerLoader.hpp"
#include "etl/impl/LedgerPublisher.hpp"
#include "etl/impl/Transformer.hpp"
#include "feed/SubscriptionManager.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/log/Logger.hpp"
#include <boost/asio/io_context.hpp>
@@ -77,16 +77,14 @@ namespace etl {
*/
class ETLService {
// TODO: make these template parameters in ETLService
using SubscriptionManagerType = feed::SubscriptionManager;
using LoadBalancerType = LoadBalancer;
using NetworkValidatedLedgersType = NetworkValidatedLedgers;
using DataPipeType = etl::impl::ExtractionDataPipe<org::xrpl::rpc::v1::GetLedgerResponse>;
using CacheType = data::LedgerCache;
using CacheLoaderType = etl::CacheLoader<CacheType>;
using LedgerFetcherType = etl::impl::LedgerFetcher<LoadBalancerType>;
using ExtractorType = etl::impl::Extractor<DataPipeType, NetworkValidatedLedgersType, LedgerFetcherType>;
using ExtractorType = etl::impl::Extractor<DataPipeType, LedgerFetcherType>;
using LedgerLoaderType = etl::impl::LedgerLoader<LoadBalancerType, LedgerFetcherType>;
using LedgerPublisherType = etl::impl::LedgerPublisher<SubscriptionManagerType, CacheType>;
using LedgerPublisherType = etl::impl::LedgerPublisher<CacheType>;
using AmendmentBlockHandlerType = etl::impl::AmendmentBlockHandler<>;
using TransformerType =
etl::impl::Transformer<DataPipeType, LedgerLoaderType, LedgerPublisherType, AmendmentBlockHandlerType>;
@@ -95,7 +93,7 @@ class ETLService {
std::shared_ptr<BackendInterface> backend_;
std::shared_ptr<LoadBalancerType> loadBalancer_;
std::shared_ptr<NetworkValidatedLedgersType> networkValidatedLedgers_;
std::shared_ptr<NetworkValidatedLedgersInterface> networkValidatedLedgers_;
std::uint32_t extractorThreads_ = 1;
std::thread worker_;
@@ -128,9 +126,9 @@ public:
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<SubscriptionManagerType> subscriptions,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<LoadBalancerType> balancer,
std::shared_ptr<NetworkValidatedLedgersType> ledgers
std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
);
/**
@@ -151,9 +149,9 @@ public:
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<SubscriptionManagerType> subscriptions,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<LoadBalancerType> balancer,
std::shared_ptr<NetworkValidatedLedgersType> ledgers
std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
)
{
auto etl = std::make_shared<ETLService>(config, ioc, backend, subscriptions, balancer, ledgers);

18
src/etl/ETLState.cpp
View File

@@ -27,21 +27,23 @@
#include <ripple/protocol/jss.h>
#include <cstdint>
#include <optional>
namespace etl {
ETLState
tag_invoke(boost::json::value_to_tag<ETLState>, boost::json::value const& jv)
std::optional<ETLState>
tag_invoke(boost::json::value_to_tag<std::optional<ETLState>>, boost::json::value const& jv)
{
ETLState state;
auto const& jsonObject = jv.as_object();
if (!jsonObject.contains(JS(error))) {
if (jsonObject.contains(JS(result)) && jsonObject.at(JS(result)).as_object().contains(JS(info))) {
auto const rippledInfo = jsonObject.at(JS(result)).as_object().at(JS(info)).as_object();
if (rippledInfo.contains(JS(network_id)))
state.networkID.emplace(boost::json::value_to<int64_t>(rippledInfo.at(JS(network_id))));
}
if (jsonObject.contains(JS(error)))
return std::nullopt;
if (jsonObject.contains(JS(result)) && jsonObject.at(JS(result)).as_object().contains(JS(info))) {
auto const rippledInfo = jsonObject.at(JS(result)).as_object().at(JS(info)).as_object();
if (rippledInfo.contains(JS(network_id)))
state.networkID.emplace(boost::json::value_to<int64_t>(rippledInfo.at(JS(network_id))));
}
return state;

6
src/etl/ETLState.hpp
View File

@@ -51,7 +51,7 @@ struct ETLState {
});
if (serverInfoRippled)
return boost::json::value_to<ETLState>(boost::json::value(*serverInfoRippled));
return boost::json::value_to<std::optional<ETLState>>(boost::json::value(*serverInfoRippled));
return std::nullopt;
}
@@ -63,7 +63,7 @@ struct ETLState {
* @param jv The json value to convert
* @return The ETLState
*/
ETLState
tag_invoke(boost::json::value_to_tag<ETLState>, boost::json::value const& jv);
std::optional<ETLState>
tag_invoke(boost::json::value_to_tag<std::optional<ETLState>>, boost::json::value const& jv);
} // namespace etl

112
src/etl/LoadBalancer.cpp
View File

@@ -21,9 +21,10 @@
#include "data/BackendInterface.hpp"
#include "etl/ETLHelpers.hpp"
#include "etl/ETLService.hpp"
#include "etl/ETLState.hpp"
#include "etl/Source.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Assert.hpp"
#include "util/Constants.hpp"
#include "util/Random.hpp"
#include "util/log/Logger.hpp"
@@ -57,19 +58,23 @@ LoadBalancer::make_LoadBalancer(
Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceFactory sourceFactory
)
{
return std::make_shared<LoadBalancer>(config, ioc, backend, subscriptions, validatedLedgers);
return std::make_shared<LoadBalancer>(
config, ioc, std::move(backend), std::move(subscriptions), std::move(validatedLedgers), std::move(sourceFactory)
);
}
LoadBalancer::LoadBalancer(
Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceFactory sourceFactory
)
{
auto const forwardingCacheTimeout = config.valueOr<float>("forwarding_cache_timeout", 0.f);
@@ -81,7 +86,8 @@ LoadBalancer::LoadBalancer(
static constexpr std::uint32_t MAX_DOWNLOAD = 256;
if (auto value = config.maybeValue<uint32_t>("num_markers"); value) {
downloadRanges_ = std::clamp(*value, 1u, MAX_DOWNLOAD);
ASSERT(*value > 0 and *value <= MAX_DOWNLOAD, "'num_markers' value in config must be in range 1-256");
downloadRanges_ = *value;
} else if (backend->fetchLedgerRange()) {
downloadRanges_ = 4;
}
@@ -98,7 +104,7 @@ LoadBalancer::LoadBalancer(
};
for (auto const& entry : config.array("etl_sources")) {
auto source = make_Source(
auto source = sourceFactory(
entry,
ioc,
backend,
@@ -113,12 +119,12 @@ LoadBalancer::LoadBalancer(
);
// checking etl node validity
auto const stateOpt = ETLState::fetchETLStateFromSource(source);
auto const stateOpt = ETLState::fetchETLStateFromSource(*source);
if (!stateOpt) {
checkOnETLFailure(fmt::format(
"Failed to fetch ETL state from source = {} Please check the configuration and network",
source.toString()
source->toString()
));
} else if (etlState_ && etlState_->networkID && stateOpt->networkID &&
etlState_->networkID != stateOpt->networkID) {
@@ -132,7 +138,7 @@ LoadBalancer::LoadBalancer(
}
sources_.push_back(std::move(source));
LOG(log_.info()) << "Added etl source - " << sources_.back().toString();
LOG(log_.info()) << "Added etl source - " << sources_.back()->toString();
}
if (sources_.empty())
@@ -140,8 +146,8 @@ LoadBalancer::LoadBalancer(
// This is made separate from source creation to prevent UB in case one of the sources will call
// chooseForwardingSource while we are still filling the sources_ vector
for (auto& source : sources_) {
source.run();
for (auto const& source : sources_) {
source->run();
}
}
@@ -150,53 +156,57 @@ LoadBalancer::~LoadBalancer()
sources_.clear();
}
std::pair<std::vector<std::string>, bool>
LoadBalancer::loadInitialLedger(uint32_t sequence, bool cacheOnly)
std::vector<std::string>
LoadBalancer::loadInitialLedger(uint32_t sequence, bool cacheOnly, std::chrono::steady_clock::duration retryAfter)
{
std::vector<std::string> response;
auto const success = execute(
execute(
[this, &response, &sequence, cacheOnly](auto& source) {
auto [data, res] = source.loadInitialLedger(sequence, downloadRanges_, cacheOnly);
auto [data, res] = source->loadInitialLedger(sequence, downloadRanges_, cacheOnly);
if (!res) {
LOG(log_.error()) << "Failed to download initial ledger. Sequence = " << sequence
<< " source = " << source.toString();
LOG(log_.error()) << "Failed to download initial ledger."
<< " Sequence = " << sequence << " source = " << source->toString();
} else {
response = std::move(data);
}
return res;
},
sequence
sequence,
retryAfter
);
return {std::move(response), success};
return response;
}
LoadBalancer::OptionalGetLedgerResponseType
LoadBalancer::fetchLedger(uint32_t ledgerSequence, bool getObjects, bool getObjectNeighbors)
LoadBalancer::fetchLedger(
uint32_t ledgerSequence,
bool getObjects,
bool getObjectNeighbors,
std::chrono::steady_clock::duration retryAfter
)
{
GetLedgerResponseType response;
bool const success = execute(
execute(
[&response, ledgerSequence, getObjects, getObjectNeighbors, log = log_](auto& source) {
auto [status, data] = source.fetchLedger(ledgerSequence, getObjects, getObjectNeighbors);
auto [status, data] = source->fetchLedger(ledgerSequence, getObjects, getObjectNeighbors);
response = std::move(data);
if (status.ok() && response.validated()) {
LOG(log.info()) << "Successfully fetched ledger = " << ledgerSequence
<< " from source = " << source.toString();
<< " from source = " << source->toString();
return true;
}
LOG(log.warn()) << "Could not fetch ledger " << ledgerSequence << ", Reply: " << response.DebugString()
<< ", error_code: " << status.error_code() << ", error_msg: " << status.error_message()
<< ", source = " << source.toString();
<< ", source = " << source->toString();
return false;
},
ledgerSequence
ledgerSequence,
retryAfter
);
if (success) {
return response;
}
return {};
return response;
}
std::optional<boost::json::object>
@@ -212,15 +222,14 @@ LoadBalancer::forwardToRippled(
}
}
std::size_t sourceIdx = 0;
if (!sources_.empty())
sourceIdx = util::Random::uniform(0ul, sources_.size() - 1);
ASSERT(not sources_.empty(), "ETL sources must be configured to forward requests.");
std::size_t sourceIdx = util::Random::uniform(0ul, sources_.size() - 1);
auto numAttempts = 0u;
std::optional<boost::json::object> response;
while (numAttempts < sources_.size()) {
if (auto res = sources_[sourceIdx].forwardToRippled(request, clientIp, yield)) {
if (auto res = sources_[sourceIdx]->forwardToRippled(request, clientIp, yield)) {
response = std::move(res);
break;
}
@@ -240,42 +249,41 @@ LoadBalancer::toJson() const
{
boost::json::array ret;
for (auto& src : sources_)
ret.push_back(src.toJson());
ret.push_back(src->toJson());
return ret;
}
template <typename Func>
bool
LoadBalancer::execute(Func f, uint32_t ledgerSequence)
void
LoadBalancer::execute(Func f, uint32_t ledgerSequence, std::chrono::steady_clock::duration retryAfter)
{
std::size_t sourceIdx = 0;
if (!sources_.empty())
sourceIdx = util::Random::uniform(0ul, sources_.size() - 1);
ASSERT(not sources_.empty(), "ETL sources must be configured to execute functions.");
size_t sourceIdx = util::Random::uniform(0ul, sources_.size() - 1);
auto numAttempts = 0;
size_t numAttempts = 0;
while (true) {
auto& source = sources_[sourceIdx];
LOG(log_.debug()) << "Attempting to execute func. ledger sequence = " << ledgerSequence
<< " - source = " << source.toString();
<< " - source = " << source->toString();
// Originally, it was (source->hasLedger(ledgerSequence) || true)
/* Sometimes rippled has ledger but doesn't actually know. However,
but this does NOT happen in the normal case and is safe to remove
This || true is only needed when loading full history standalone */
if (source.hasLedger(ledgerSequence)) {
if (source->hasLedger(ledgerSequence)) {
bool const res = f(source);
if (res) {
LOG(log_.debug()) << "Successfully executed func at source = " << source.toString()
LOG(log_.debug()) << "Successfully executed func at source = " << source->toString()
<< " - ledger sequence = " << ledgerSequence;
break;
}
LOG(log_.warn()) << "Failed to execute func at source = " << source.toString()
LOG(log_.warn()) << "Failed to execute func at source = " << source->toString()
<< " - ledger sequence = " << ledgerSequence;
} else {
LOG(log_.warn()) << "Ledger not present at source = " << source.toString()
LOG(log_.warn()) << "Ledger not present at source = " << source->toString()
<< " - ledger sequence = " << ledgerSequence;
}
sourceIdx = (sourceIdx + 1) % sources_.size();
@@ -283,10 +291,9 @@ LoadBalancer::execute(Func f, uint32_t ledgerSequence)
if (numAttempts % sources_.size() == 0) {
LOG(log_.info()) << "Ledger sequence " << ledgerSequence
<< " is not yet available from any configured sources. Sleeping and trying again";
std::this_thread::sleep_for(std::chrono::seconds(2));
std::this_thread::sleep_for(retryAfter);
}
}
return true;
}
std::optional<ETLState>
@@ -304,10 +311,11 @@ LoadBalancer::chooseForwardingSource()
{
hasForwardingSource_ = false;
for (auto& source : sources_) {
if (source.isConnected()) {
source.setForwarding(true);
if (not hasForwardingSource_ and source->isConnected()) {
source->setForwarding(true);
hasForwardingSource_ = true;
return;
} else {
source->setForwarding(false);
}
}
}

54
src/etl/LoadBalancer.hpp
View File

@@ -24,7 +24,7 @@
#include "etl/ETLState.hpp"
#include "etl/Source.hpp"
#include "etl/impl/ForwardingCache.hpp"
#include "feed/SubscriptionManager.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/config/Config.hpp"
#include "util/log/Logger.hpp"
@@ -39,6 +39,7 @@
#include <ripple/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
#include <atomic>
#include <chrono>
#include <cstdint>
#include <memory>
#include <optional>
@@ -46,14 +47,6 @@
#include <utility>
#include <vector>
namespace etl {
class ProbingSource;
} // namespace etl
namespace feed {
class SubscriptionManager;
} // namespace feed
namespace etl {
/**
@@ -73,9 +66,9 @@ private:
static constexpr std::uint32_t DEFAULT_DOWNLOAD_RANGES = 16;
util::Logger log_{"ETL"};
// Forwarding cache must be destroyed after sources because sources have a callnack to invalidate cache
// Forwarding cache must be destroyed after sources because sources have a callback to invalidate cache
std::optional<impl::ForwardingCache> forwardingCache_;
std::vector<Source> sources_;
std::vector<SourcePtr> sources_;
std::optional<ETLState> etlState_;
std::uint32_t downloadRanges_ =
DEFAULT_DOWNLOAD_RANGES; /*< The number of markers to use when downloading initial ledger */
@@ -90,13 +83,15 @@ public:
* @param backend BackendInterface implementation
* @param subscriptions Subscription manager
* @param validatedLedgers The network validated ledgers datastructure
* @param sourceFactory A factory function to create a source
*/
LoadBalancer(
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceFactory sourceFactory = make_Source
);
/**
@@ -107,6 +102,7 @@ public:
* @param backend BackendInterface implementation
* @param subscriptions Subscription manager
* @param validatedLedgers The network validated ledgers datastructure
* @param sourceFactory A factory function to create a source
* @return A shared pointer to a new instance of LoadBalancer
*/
static std::shared_ptr<LoadBalancer>
@@ -114,22 +110,28 @@ public:
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceFactory sourceFactory = make_Source
);
~LoadBalancer();
/**
* @brief Load the initial ledger, writing data to the queue.
* @note This function will retry indefinitely until the ledger is downloaded.
*
* @param sequence Sequence of ledger to download
* @param cacheOnly Whether to only write to cache and not to the DB; defaults to false
* @return A std::pair<std::vector<std::string>, bool> The ledger data and a bool indicating whether the download
* was successful
* @param retryAfter Time to wait between retries (2 seconds by default)
* @return A std::vector<std::string> The ledger data
*/
std::pair<std::vector<std::string>, bool>
loadInitialLedger(uint32_t sequence, bool cacheOnly = false);
std::vector<std::string>
loadInitialLedger(
uint32_t sequence,
bool cacheOnly = false,
std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2}
);
/**
* @brief Fetch data for a specific ledger.
@@ -140,11 +142,17 @@ public:
* @param ledgerSequence Sequence of the ledger to fetch
* @param getObjects Whether to get the account state diff between this ledger and the prior one
* @param getObjectNeighbors Whether to request object neighbors
* @param retryAfter Time to wait between retries (2 seconds by default)
* @return The extracted data, if extraction was successful. If the ledger was found
* in the database or the server is shutting down, the optional will be empty
*/
OptionalGetLedgerResponseType
fetchLedger(uint32_t ledgerSequence, bool getObjects, bool getObjectNeighbors);
fetchLedger(
uint32_t ledgerSequence,
bool getObjects,
bool getObjectNeighbors,
std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2}
);
/**
* @brief Represent the state of this load balancer as a JSON object
@@ -186,12 +194,12 @@ private:
*
* @param f Function to execute. This function takes the ETL source as an argument, and returns a bool
* @param ledgerSequence f is executed for each Source that has this ledger
* @return true if f was eventually executed successfully. false if the ledger was found in the database or the
* @param retryAfter Time to wait between retries (2 seconds by default)
* server is shutting down
*/
template <typename Func>
bool
execute(Func f, uint32_t ledgerSequence);
void
execute(Func f, uint32_t ledgerSequence, std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2});
/**
* @brief Choose a new source to forward requests

65
src/etl/NetworkValidatedLedgers.cpp Normal file
View File

@@ -0,0 +1,65 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include "etl/NetworkValidatedLedgers.hpp"
#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
namespace etl {
std::shared_ptr<NetworkValidatedLedgers>
NetworkValidatedLedgers::make_ValidatedLedgers()
{
return std::make_shared<NetworkValidatedLedgers>();
}
void
NetworkValidatedLedgers::push(uint32_t idx)
{
std::lock_guard const lck(m_);
if (!max_ || idx > *max_)
max_ = idx;
cv_.notify_all();
}
std::optional<uint32_t>
NetworkValidatedLedgers::getMostRecent()
{
std::unique_lock lck(m_);
cv_.wait(lck, [this]() { return max_; });
return max_;
}
bool
NetworkValidatedLedgers::waitUntilValidatedByNetwork(uint32_t sequence, std::optional<uint32_t> maxWaitMs)
{
std::unique_lock lck(m_);
auto pred = [sequence, this]() -> bool { return (max_ && sequence <= *max_); };
if (maxWaitMs) {
cv_.wait_for(lck, std::chrono::milliseconds(*maxWaitMs));
} else {
cv_.wait(lck, pred);
}
return pred();
}
} // namespace etl

86
src/etl/NetworkValidatedLedgers.hpp Normal file
View File

@@ -0,0 +1,86 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2022, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#pragma once
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include <condition_variable>
#include <cstdint>
#include <memory>
#include <mutex>
#include <optional>
namespace etl {
/**
* @brief This datastructure is used to keep track of the sequence of the most recent ledger validated by the network.
*
* There are two methods that will wait until certain conditions are met. This datastructure is able to be "stopped".
* When the datastructure is stopped, any threads currently waiting are unblocked.
* Any later calls to methods of this datastructure will not wait. Once the datastructure is stopped, the datastructure
* remains stopped for the rest of its lifetime.
*/
class NetworkValidatedLedgers : public NetworkValidatedLedgersInterface {
// max sequence validated by network
std::optional<uint32_t> max_;
mutable std::mutex m_;
std::condition_variable cv_;
public:
/**
* @brief A factory function for NetworkValidatedLedgers
*
* @return A shared pointer to a new instance of NetworkValidatedLedgers
*/
static std::shared_ptr<NetworkValidatedLedgers>
make_ValidatedLedgers();
/**
* @brief Notify the datastructure that idx has been validated by the network.
*
* @param idx Sequence validated by network
*/
void
push(uint32_t idx) final;
/**
* @brief Get most recently validated sequence.
*
* If no ledgers are known to have been validated, this function waits until the next ledger is validated
*
* @return Sequence of most recently validated ledger. empty optional if the datastructure has been stopped
*/
std::optional<uint32_t>
getMostRecent() final;
/**
* @brief Waits for the sequence to be validated by the network.
*
* @param sequence The sequence to wait for
* @param maxWaitMs Maximum time to wait for the sequence to be validated. If empty, wait indefinitely
* @return true if sequence was validated, false otherwise a return value of false means the datastructure has been
* stopped
*/
bool
waitUntilValidatedByNetwork(uint32_t sequence, std::optional<uint32_t> maxWaitMs = {}) final;
};
} // namespace etl

64
src/etl/NetworkValidatedLedgersInterface.hpp Normal file
View File

@@ -0,0 +1,64 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2022, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
/** @file */
#pragma once
#include <cstdint>
#include <optional>
namespace etl {
/**
* @brief An interface for NetworkValidatedLedgers
*/
class NetworkValidatedLedgersInterface {
public:
virtual ~NetworkValidatedLedgersInterface() = default;
/**
* @brief Notify the datastructure that idx has been validated by the network.
*
* @param idx Sequence validated by network
*/
virtual void
push(uint32_t idx) = 0;
/**
* @brief Get most recently validated sequence.
*
* If no ledgers are known to have been validated, this function waits until the next ledger is validated
*
* @return Sequence of most recently validated ledger. empty optional if the datastructure has been stopped
*/
virtual std::optional<uint32_t>
getMostRecent() = 0;
/**
* @brief Waits for the sequence to be validated by the network.
*
* @param sequence The sequence to wait for
* @param maxWaitMs Maximum time to wait for the sequence to be validated. If empty, wait indefinitely
* @return true if sequence was validated, false otherwise a return value of false means the datastructure has been
* stopped
*/
virtual bool
waitUntilValidatedByNetwork(uint32_t sequence, std::optional<uint32_t> maxWaitMs = {}) = 0;
};
} // namespace etl

24
src/etl/Source.cpp
View File

@@ -21,7 +21,11 @@
#include "data/BackendInterface.hpp"
#include "etl/ETLHelpers.hpp"
#include "feed/SubscriptionManager.hpp"
#include "etl/impl/ForwardingSource.hpp"
#include "etl/impl/GrpcSource.hpp"
#include "etl/impl/SourceImpl.hpp"
#include "etl/impl/SubscriptionSource.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/config/Config.hpp"
#include <boost/asio/io_context.hpp>
@@ -32,18 +36,16 @@
namespace etl {
template class SourceImpl<>;
Source
SourcePtr
make_Source(
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers,
Source::OnDisconnectHook onDisconnect,
Source::OnConnectHook onConnect,
Source::OnLedgerClosedHook onLedgerClosed
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceBase::OnConnectHook onConnect,
SourceBase::OnDisconnectHook onDisconnect,
SourceBase::OnLedgerClosedHook onLedgerClosed
)
{
auto const ip = config.valueOr<std::string>("ip", {});
@@ -63,9 +65,9 @@ make_Source(
std::move(onLedgerClosed)
);
return Source{
return std::make_unique<impl::SourceImpl<>>(
ip, wsPort, grpcPort, std::move(grpcSource), std::move(subscriptionSource), std::move(forwardingSource)
};
);
}
} // namespace etl

168
src/etl/Source.hpp
View File

@@ -20,11 +20,8 @@
#pragma once
#include "data/BackendInterface.hpp"
#include "etl/ETLHelpers.hpp"
#include "etl/impl/ForwardingSource.hpp"
#include "etl/impl/GrpcSource.hpp"
#include "etl/impl/SubscriptionSource.hpp"
#include "feed/SubscriptionManager.hpp"
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/config/Config.hpp"
#include "util/log/Logger.hpp"
@@ -35,8 +32,8 @@
#include <grpcpp/support/status.h>
#include <org/xrpl/rpc/v1/get_ledger.pb.h>
#include <chrono>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <string>
@@ -48,122 +45,48 @@ namespace etl {
/**
* @brief Provides an implementation of a ETL source
*
* @tparam GrpcSourceType The type of the gRPC source
* @tparam SubscriptionSourceTypePtr The type of the subscription source
* @tparam ForwardingSourceType The type of the forwarding source
*/
template <
typename GrpcSourceType = impl::GrpcSource,
typename SubscriptionSourceTypePtr = std::unique_ptr<impl::SubscriptionSource>,
typename ForwardingSourceType = impl::ForwardingSource>
class SourceImpl {
std::string ip_;
std::string wsPort_;
std::string grpcPort_;
GrpcSourceType grpcSource_;
SubscriptionSourceTypePtr subscriptionSource_;
ForwardingSourceType forwardingSource_;
class SourceBase {
public:
using OnConnectHook = impl::SubscriptionSource::OnConnectHook;
using OnDisconnectHook = impl::SubscriptionSource::OnDisconnectHook;
using OnLedgerClosedHook = impl::SubscriptionSource::OnLedgerClosedHook;
using OnConnectHook = std::function<void()>;
using OnDisconnectHook = std::function<void()>;
using OnLedgerClosedHook = std::function<void()>;
/**
* @brief Construct a new SourceImpl object
*
* @param ip The IP of the source
* @param wsPort The web socket port of the source
* @param grpcPort The gRPC port of the source
* @param grpcSource The gRPC source
* @param subscriptionSource The subscription source
* @param forwardingSource The forwarding source
*/
template <typename SomeGrpcSourceType, typename SomeForwardingSourceType>
requires std::is_same_v<GrpcSourceType, SomeGrpcSourceType> and
std::is_same_v<ForwardingSourceType, SomeForwardingSourceType>
SourceImpl(
std::string ip,
std::string wsPort,
std::string grpcPort,
SomeGrpcSourceType&& grpcSource,
SubscriptionSourceTypePtr subscriptionSource,
SomeForwardingSourceType&& forwardingSource
)
: ip_(std::move(ip))
, wsPort_(std::move(wsPort))
, grpcPort_(std::move(grpcPort))
, grpcSource_(std::forward<SomeGrpcSourceType>(grpcSource))
, subscriptionSource_(std::move(subscriptionSource))
, forwardingSource_(std::forward<SomeForwardingSourceType>(forwardingSource))
{
}
virtual ~SourceBase() = default;
/**
* @brief Run subscriptions loop of the source
*/
void
run()
{
subscriptionSource_->run();
}
virtual void
run() = 0;
/**
* @brief Check if source is connected
*
* @return true if source is connected; false otherwise
*/
bool
isConnected() const
{
return subscriptionSource_->isConnected();
}
virtual bool
isConnected() const = 0;
/**
* @brief Set the forwarding state of the source.
*
* @param isForwarding Whether to forward or not
*/
void
setForwarding(bool isForwarding)
{
subscriptionSource_->setForwarding(isForwarding);
}
virtual void
setForwarding(bool isForwarding) = 0;
/**
* @brief Represent the source as a JSON object
*
* @return JSON representation of the source
*/
boost::json::object
toJson() const
{
boost::json::object res;
res["validated_range"] = subscriptionSource_->validatedRange();
res["is_connected"] = std::to_string(static_cast<int>(subscriptionSource_->isConnected()));
res["ip"] = ip_;
res["ws_port"] = wsPort_;
res["grpc_port"] = grpcPort_;
auto last = subscriptionSource_->lastMessageTime();
if (last.time_since_epoch().count() != 0) {
res["last_msg_age_seconds"] = std::to_string(
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - last).count()
);
}
return res;
}
virtual boost::json::object
toJson() const = 0;
/** @return String representation of the source (for debug) */
std::string
toString() const
{
return "{validated range: " + subscriptionSource_->validatedRange() + ", ip: " + ip_ +
", web socket port: " + wsPort_ + ", grpc port: " + grpcPort_ + "}";
}
virtual std::string
toString() const = 0;
/**
* @brief Check if ledger is known by this source.
@@ -171,11 +94,8 @@ public:
* @param sequence The ledger sequence to check
* @return true if ledger is in the range of this source; false otherwise
*/
bool
hasLedger(uint32_t sequence) const
{
return subscriptionSource_->hasLedger(sequence);
}
virtual bool
hasLedger(uint32_t sequence) const = 0;
/**
* @brief Fetch data for a specific ledger.
@@ -188,11 +108,8 @@ public:
* @param getObjectNeighbors Whether to request object neighbors; defaults to false
* @return A std::pair of the response status and the response itself
*/
std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
fetchLedger(uint32_t sequence, bool getObjects = true, bool getObjectNeighbors = false)
{
return grpcSource_.fetchLedger(sequence, getObjects, getObjectNeighbors);
}
virtual std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
fetchLedger(uint32_t sequence, bool getObjects = true, bool getObjectNeighbors = false) = 0;
/**
* @brief Download a ledger in full.
@@ -202,11 +119,8 @@ public:
* @param cacheOnly Only insert into cache, not the DB; defaults to false
* @return A std::pair of the data and a bool indicating whether the download was successful
*/
std::pair<std::vector<std::string>, bool>
loadInitialLedger(uint32_t sequence, std::uint32_t numMarkers, bool cacheOnly = false)
{
return grpcSource_.loadInitialLedger(sequence, numMarkers, cacheOnly);
}
virtual std::pair<std::vector<std::string>, bool>
loadInitialLedger(uint32_t sequence, std::uint32_t numMarkers, bool cacheOnly = false) = 0;
/**
* @brief Forward a request to rippled.
@@ -216,20 +130,26 @@ public:
* @param yield The coroutine context
* @return Response wrapped in an optional on success; nullopt otherwise
*/
std::optional<boost::json::object>
virtual std::optional<boost::json::object>
forwardToRippled(
boost::json::object const& request,
std::optional<std::string> const& forwardToRippledClientIp,
boost::asio::yield_context yield
) const
{
return forwardingSource_.forwardToRippled(request, forwardToRippledClientIp, yield);
}
) const = 0;
};
extern template class SourceImpl<>;
using SourcePtr = std::unique_ptr<SourceBase>;
using Source = SourceImpl<>;
using SourceFactory = std::function<SourcePtr(
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceBase::OnConnectHook onConnect,
SourceBase::OnDisconnectHook onDisconnect,
SourceBase::OnLedgerClosedHook onLedgerClosed
)>;
/**
* @brief Create a source
@@ -239,22 +159,22 @@ using Source = SourceImpl<>;
* @param backend BackendInterface implementation
* @param subscriptions Subscription manager
* @param validatedLedgers The network validated ledgers data structure
* @param onDisconnect The hook to call on disconnect
* @param onConnect The hook to call on connect
* @param onDisconnect The hook to call on disconnect
* @param onLedgerClosed The hook to call on ledger closed. This is called when a ledger is closed and the source is set
* as forwarding.
* @return The created source
*/
Source
SourcePtr
make_Source(
util::Config const& config,
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
std::shared_ptr<feed::SubscriptionManager> subscriptions,
std::shared_ptr<NetworkValidatedLedgers> validatedLedgers,
Source::OnDisconnectHook onDisconnect,
Source::OnConnectHook onConnect,
Source::OnLedgerClosedHook onLedgerClosed
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
SourceBase::OnConnectHook onConnect,
SourceBase::OnDisconnectHook onDisconnect,
SourceBase::OnLedgerClosedHook onLedgerClosed
);
} // namespace etl

3
src/etl/impl/AsyncData.hpp
View File

@@ -99,7 +99,8 @@ public:
bool cacheOnly = false
)
{
LOG(log_.trace()) << "Processing response. " << "Marker prefix = " << getMarkerPrefix();
LOG(log_.trace()) << "Processing response. "
<< "Marker prefix = " << getMarkerPrefix();
if (abort) {
LOG(log_.error()) << "AsyncCallData aborted";
return CallStatus::ERRORED;

7
src/etl/impl/Extractor.hpp
View File

@@ -19,6 +19,7 @@
#pragma once
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/SystemState.hpp"
#include "util/Assert.hpp"
#include "util/Profiler.hpp"
@@ -39,12 +40,12 @@ namespace etl::impl {
/**
* @brief Extractor thread that is fetching GRPC data and enqueue it on the DataPipeType
*/
template <typename DataPipeType, typename NetworkValidatedLedgersType, typename LedgerFetcherType>
template <typename DataPipeType, typename LedgerFetcherType>
class Extractor {
util::Logger log_{"ETL"};
std::reference_wrapper<DataPipeType> pipe_;
std::shared_ptr<NetworkValidatedLedgersType> networkValidatedLedgers_;
std::shared_ptr<NetworkValidatedLedgersInterface> networkValidatedLedgers_;
std::reference_wrapper<LedgerFetcherType> ledgerFetcher_;
uint32_t startSequence_;
std::optional<uint32_t> finishSequence_;
@@ -55,7 +56,7 @@ class Extractor {
public:
Extractor(
DataPipeType& pipe,
std::shared_ptr<NetworkValidatedLedgersType> networkValidatedLedgers,
std::shared_ptr<NetworkValidatedLedgersInterface> networkValidatedLedgers,
LedgerFetcherType& ledgerFetcher,
uint32_t startSequence,
std::optional<uint32_t> finishSequence,

79
src/etl/impl/LedgerLoader.hpp
View File

@@ -197,58 +197,55 @@ public:
// asyncWriter consumes from the queue and inserts the data into the
// Ledger object. Once the below call returns, all data has been pushed
// into the queue
auto [edgeKeys, success] = loadBalancer_->loadInitialLedger(sequence);
auto edgeKeys = loadBalancer_->loadInitialLedger(sequence);
if (success) {
size_t numWrites = 0;
backend_->cache().setFull();
size_t numWrites = 0;
backend_->cache().setFull();
auto seconds = ::util::timed<std::chrono::seconds>([this, keys = &edgeKeys, sequence, &numWrites] {
for (auto& key : *keys) {
LOG(log_.debug()) << "Writing edge key = " << ripple::strHex(key);
auto succ = backend_->cache().getSuccessor(*ripple::uint256::fromVoidChecked(key), sequence);
if (succ)
backend_->writeSuccessor(std::move(key), sequence, uint256ToString(succ->key));
}
auto seconds = ::util::timed<std::chrono::seconds>([this, keys = std::move(edgeKeys), sequence, &numWrites](
) mutable {
for (auto& key : keys) {
LOG(log_.debug()) << "Writing edge key = " << ripple::strHex(key);
auto succ = backend_->cache().getSuccessor(*ripple::uint256::fromVoidChecked(key), sequence);
if (succ)
backend_->writeSuccessor(std::move(key), sequence, uint256ToString(succ->key));
}
ripple::uint256 prev = data::firstKey;
while (auto cur = backend_->cache().getSuccessor(prev, sequence)) {
ASSERT(cur.has_value(), "Succesor for key {} must exist", ripple::strHex(prev));
if (prev == data::firstKey)
backend_->writeSuccessor(uint256ToString(prev), sequence, uint256ToString(cur->key));
ripple::uint256 prev = data::firstKey;
while (auto cur = backend_->cache().getSuccessor(prev, sequence)) {
ASSERT(cur.has_value(), "Succesor for key {} must exist", ripple::strHex(prev));
if (prev == data::firstKey)
backend_->writeSuccessor(uint256ToString(prev), sequence, uint256ToString(cur->key));
if (isBookDir(cur->key, cur->blob)) {
auto base = getBookBase(cur->key);
// make sure the base is not an actual object
if (!backend_->cache().get(base, sequence)) {
auto succ = backend_->cache().getSuccessor(base, sequence);
ASSERT(succ.has_value(), "Book base {} must have a successor", ripple::strHex(base));
if (succ->key == cur->key) {
LOG(log_.debug()) << "Writing book successor = " << ripple::strHex(base) << " - "
<< ripple::strHex(cur->key);
if (isBookDir(cur->key, cur->blob)) {
auto base = getBookBase(cur->key);
// make sure the base is not an actual object
if (!backend_->cache().get(base, sequence)) {
auto succ = backend_->cache().getSuccessor(base, sequence);
ASSERT(succ.has_value(), "Book base {} must have a successor", ripple::strHex(base));
if (succ->key == cur->key) {
LOG(log_.debug()) << "Writing book successor = " << ripple::strHex(base) << " - "
<< ripple::strHex(cur->key);
backend_->writeSuccessor(
uint256ToString(base), sequence, uint256ToString(cur->key)
);
}
backend_->writeSuccessor(uint256ToString(base), sequence, uint256ToString(cur->key));
}
++numWrites;
}
prev = cur->key;
static constexpr std::size_t LOG_INTERVAL = 100000;
if (numWrites % LOG_INTERVAL == 0 && numWrites != 0)
LOG(log_.info()) << "Wrote " << numWrites << " book successors";
++numWrites;
}
backend_->writeSuccessor(uint256ToString(prev), sequence, uint256ToString(data::lastKey));
++numWrites;
});
prev = cur->key;
static constexpr std::size_t LOG_INTERVAL = 100000;
if (numWrites % LOG_INTERVAL == 0 && numWrites != 0)
LOG(log_.info()) << "Wrote " << numWrites << " book successors";
}
LOG(log_.info()) << "Looping through cache and submitting all writes took " << seconds
<< " seconds. numWrites = " << std::to_string(numWrites);
}
backend_->writeSuccessor(uint256ToString(prev), sequence, uint256ToString(data::lastKey));
++numWrites;
});
LOG(log_.info()) << "Looping through cache and submitting all writes took " << seconds
<< " seconds. numWrites = " << std::to_string(numWrites);
LOG(log_.debug()) << "Loaded initial ledger";

18
src/etl/impl/LedgerPublisher.hpp
View File

@@ -23,6 +23,7 @@
#include "data/DBHelpers.hpp"
#include "data/Types.hpp"
#include "etl/SystemState.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Assert.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Counter.hpp"
@@ -63,7 +64,7 @@ namespace etl::impl {
* includes reading all of the transactions from the database) is done from the application wide asio io_service, and a
* strand is used to ensure ledgers are published in order.
*/
template <typename SubscriptionManagerType, typename CacheType>
template <typename CacheType>
class LedgerPublisher {
util::Logger log_{"ETL"};
@@ -71,7 +72,7 @@ class LedgerPublisher {
std::shared_ptr<BackendInterface> backend_;
std::reference_wrapper<CacheType> cache_;
std::shared_ptr<SubscriptionManagerType> subscriptions_;
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
std::reference_wrapper<SystemState const> state_; // shared state for ETL
std::chrono::time_point<ripple::NetClock> lastCloseTime_;
@@ -94,7 +95,7 @@ public:
boost::asio::io_context& ioc,
std::shared_ptr<BackendInterface> backend,
CacheType& cache,
std::shared_ptr<SubscriptionManagerType> subscriptions,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
SystemState const& state
)
: publishStrand_{boost::asio::make_strand(ioc)}
@@ -110,11 +111,16 @@ public:
* stream.
*
* @param ledgerSequence the sequence of the ledger to publish
* @param maxAttempts the number of times to attempt to read the ledger from the database. 1 attempt per second
* @param maxAttempts the number of times to attempt to read the ledger from the database
* @param attemptsDelay the delay between attempts to read the ledger from the database
* @return Whether the ledger was found in the database and published
*/
bool
publish(uint32_t ledgerSequence, std::optional<uint32_t> maxAttempts)
publish(
uint32_t ledgerSequence,
std::optional<uint32_t> maxAttempts,
std::chrono::steady_clock::duration attemptsDelay = std::chrono::seconds{1}
)
{
LOG(log_.info()) << "Attempting to publish ledger = " << ledgerSequence;
size_t numAttempts = 0;
@@ -130,7 +136,7 @@ public:
LOG(log_.debug()) << "Failed to publish ledger after " << numAttempts << " attempts.";
return false;
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::this_thread::sleep_for(attemptsDelay);
continue;
}

219
src/etl/impl/SourceImpl.hpp Normal file
View File

@@ -0,0 +1,219 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#pragma once
#include "etl/Source.hpp"
#include "etl/impl/ForwardingSource.hpp"
#include "etl/impl/GrpcSource.hpp"
#include "etl/impl/SubscriptionSource.hpp"
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <grpcpp/support/status.h>
#include <org/xrpl/rpc/v1/get_ledger.pb.h>
#include <chrono>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>
namespace etl::impl {
/**
* @brief Provides an implementation of a ETL source
*
* @tparam GrpcSourceType The type of the gRPC source
* @tparam SubscriptionSourceTypePtr The type of the subscription source
* @tparam ForwardingSourceType The type of the forwarding source
*/
template <
typename GrpcSourceType = GrpcSource,
typename SubscriptionSourceTypePtr = std::unique_ptr<SubscriptionSource>,
typename ForwardingSourceType = ForwardingSource>
class SourceImpl : public SourceBase {
std::string ip_;
std::string wsPort_;
std::string grpcPort_;
GrpcSourceType grpcSource_;
SubscriptionSourceTypePtr subscriptionSource_;
ForwardingSourceType forwardingSource_;
public:
/**
* @brief Construct a new SourceImpl object
*
* @param ip The IP of the source
* @param wsPort The web socket port of the source
* @param grpcPort The gRPC port of the source
* @param grpcSource The gRPC source
* @param subscriptionSource The subscription source
* @param forwardingSource The forwarding source
*/
template <typename SomeGrpcSourceType, typename SomeForwardingSourceType>
requires std::is_same_v<GrpcSourceType, SomeGrpcSourceType> and
std::is_same_v<ForwardingSourceType, SomeForwardingSourceType>
SourceImpl(
std::string ip,
std::string wsPort,
std::string grpcPort,
SomeGrpcSourceType&& grpcSource,
SubscriptionSourceTypePtr subscriptionSource,
SomeForwardingSourceType&& forwardingSource
)
: ip_(std::move(ip))
, wsPort_(std::move(wsPort))
, grpcPort_(std::move(grpcPort))
, grpcSource_(std::forward<SomeGrpcSourceType>(grpcSource))
, subscriptionSource_(std::move(subscriptionSource))
, forwardingSource_(std::forward<SomeForwardingSourceType>(forwardingSource))
{
}
/**
* @brief Run subscriptions loop of the source
*/
void
run() final
{
subscriptionSource_->run();
}
/**
* @brief Check if source is connected
*
* @return true if source is connected; false otherwise
*/
bool
isConnected() const final
{
return subscriptionSource_->isConnected();
}
/**
* @brief Set the forwarding state of the source.
*
* @param isForwarding Whether to forward or not
*/
void
setForwarding(bool isForwarding) final
{
subscriptionSource_->setForwarding(isForwarding);
}
/**
* @brief Represent the source as a JSON object
*
* @return JSON representation of the source
*/
boost::json::object
toJson() const final
{
boost::json::object res;
res["validated_range"] = subscriptionSource_->validatedRange();
res["is_connected"] = std::to_string(static_cast<int>(subscriptionSource_->isConnected()));
res["ip"] = ip_;
res["ws_port"] = wsPort_;
res["grpc_port"] = grpcPort_;
auto last = subscriptionSource_->lastMessageTime();
if (last.time_since_epoch().count() != 0) {
res["last_msg_age_seconds"] = std::to_string(
std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - last).count()
);
}
return res;
}
/** @return String representation of the source (for debug) */
std::string
toString() const final
{
return "{validated range: " + subscriptionSource_->validatedRange() + ", ip: " + ip_ +
", web socket port: " + wsPort_ + ", grpc port: " + grpcPort_ + "}";
}
/**
* @brief Check if ledger is known by this source.
*
* @param sequence The ledger sequence to check
* @return true if ledger is in the range of this source; false otherwise
*/
bool
hasLedger(uint32_t sequence) const final
{
return subscriptionSource_->hasLedger(sequence);
}
/**
* @brief Fetch data for a specific ledger.
*
* This function will continuously try to fetch data for the specified ledger until the fetch succeeds, the ledger
* is found in the database, or the server is shutting down.
*
* @param sequence Sequence of the ledger to fetch
* @param getObjects Whether to get the account state diff between this ledger and the prior one; defaults to true
* @param getObjectNeighbors Whether to request object neighbors; defaults to false
* @return A std::pair of the response status and the response itself
*/
std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
fetchLedger(uint32_t sequence, bool getObjects = true, bool getObjectNeighbors = false) final
{
return grpcSource_.fetchLedger(sequence, getObjects, getObjectNeighbors);
}
/**
* @brief Download a ledger in full.
*
* @param sequence Sequence of the ledger to download
* @param numMarkers Number of markers to generate for async calls
* @param cacheOnly Only insert into cache, not the DB; defaults to false
* @return A std::pair of the data and a bool indicating whether the download was successful
*/
std::pair<std::vector<std::string>, bool>
loadInitialLedger(uint32_t sequence, std::uint32_t numMarkers, bool cacheOnly = false) final
{
return grpcSource_.loadInitialLedger(sequence, numMarkers, cacheOnly);
}
/**
* @brief Forward a request to rippled.
*
* @param request The request to forward
* @param forwardToRippledClientIp IP of the client forwarding this request if known
* @param yield The coroutine context
* @return Response wrapped in an optional on success; nullopt otherwise
*/
std::optional<boost::json::object>
forwardToRippled(
boost::json::object const& request,
std::optional<std::string> const& forwardToRippledClientIp,
boost::asio::yield_context yield
) const final
{
return forwardingSource_.forwardToRippled(request, forwardToRippledClientIp, yield);
}
};
} // namespace etl::impl

48
src/etl/impl/SubscriptionSource.cpp
View File

@@ -19,6 +19,8 @@
#include "etl/impl/SubscriptionSource.hpp"
#include "etl/ETLHelpers.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/JS.hpp"
#include "util/Retry.hpp"
#include "util/log/Logger.hpp"
@@ -28,8 +30,11 @@
#include <boost/algorithm/string/split.hpp>
#include <boost/asio/error.hpp>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/strand.hpp>
#include <boost/asio/use_future.hpp>
#include <boost/beast/http/field.hpp>
#include <boost/json/object.hpp>
#include <boost/json/parse.hpp>
#include <boost/json/serialize.hpp>
@@ -52,6 +57,33 @@
namespace etl::impl {
SubscriptionSource::SubscriptionSource(
boost::asio::io_context& ioContext,
std::string const& ip,
std::string const& wsPort,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
OnConnectHook onConnect,
OnDisconnectHook onDisconnect,
OnLedgerClosedHook onLedgerClosed,
std::chrono::steady_clock::duration const connectionTimeout,
std::chrono::steady_clock::duration const retryDelay
)
: log_(fmt::format("GrpcSource[{}:{}]", ip, wsPort))
, wsConnectionBuilder_(ip, wsPort)
, validatedLedgers_(std::move(validatedLedgers))
, subscriptions_(std::move(subscriptions))
, strand_(boost::asio::make_strand(ioContext))
, retry_(util::makeRetryExponentialBackoff(retryDelay, RETRY_MAX_DELAY, strand_))
, onConnect_(std::move(onConnect))
, onDisconnect_(std::move(onDisconnect))
, onLedgerClosed_(std::move(onLedgerClosed))
{
wsConnectionBuilder_.addHeader({boost::beast::http::field::user_agent, "clio-client"})
.addHeader({"X-User", "clio-client"})
.setConnectionTimeout(connectionTimeout);
}
SubscriptionSource::~SubscriptionSource()
{
stop();
@@ -91,6 +123,12 @@ SubscriptionSource::isConnected() const
return isConnected_;
}
bool
SubscriptionSource::isForwarding() const
{
return isForwarding_;
}
void
SubscriptionSource::setForwarding(bool isForwarding)
{
@@ -203,18 +241,18 @@ SubscriptionSource::handleMessage(std::string const& message)
} else {
if (isForwarding_) {
if (object.contains(JS(transaction))) {
dependencies_.forwardProposedTransaction(object);
subscriptions_->forwardProposedTransaction(object);
} else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ValidationReceived) {
dependencies_.forwardValidation(object);
subscriptions_->forwardValidation(object);
} else if (object.contains(JS(type)) && object.at(JS(type)) == JS_ManifestReceived) {
dependencies_.forwardManifest(object);
subscriptions_->forwardManifest(object);
}
}
}
if (ledgerIndex != 0) {
LOG(log_.trace()) << "Pushing ledger sequence = " << ledgerIndex;
dependencies_.pushValidatedLedger(ledgerIndex);
validatedLedgers_->push(ledgerIndex);
}
return std::nullopt;
@@ -228,9 +266,9 @@ void
SubscriptionSource::handleError(util::requests::RequestError const& error, boost::asio::yield_context yield)
{
isConnected_ = false;
isForwarding_ = false;
if (not stop_) {
onDisconnect_();
isForwarding_ = false;
}
if (wsConnection_ != nullptr) {

43
src/etl/impl/SubscriptionSource.hpp
View File

@@ -19,7 +19,9 @@
#pragma once
#include "etl/impl/SubscriptionSourceDependencies.hpp"
#include "etl/ETLHelpers.hpp"
#include "etl/Source.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/Mutex.hpp"
#include "util/Retry.hpp"
#include "util/log/Logger.hpp"
@@ -35,7 +37,6 @@
#include <atomic>
#include <chrono>
#include <cstdint>
#include <functional>
#include <future>
#include <memory>
#include <optional>
@@ -50,9 +51,9 @@ namespace etl::impl {
*/
class SubscriptionSource {
public:
using OnConnectHook = std::function<void()>;
using OnDisconnectHook = std::function<void()>;
using OnLedgerClosedHook = std::function<void()>;
using OnConnectHook = SourceBase::OnConnectHook;
using OnDisconnectHook = SourceBase::OnDisconnectHook;
using OnLedgerClosedHook = SourceBase::OnLedgerClosedHook;
private:
util::Logger log_;
@@ -65,7 +66,8 @@ private:
};
util::Mutex<ValidatedLedgersData> validatedLedgersData_;
SubscriptionSourceDependencies dependencies_;
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers_;
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
boost::asio::strand<boost::asio::io_context::executor_type> strand_;
@@ -92,7 +94,6 @@ public:
* @brief Construct a new Subscription Source object
*
* @tparam NetworkValidatedLedgersType The type of the network validated ledgers object
* @tparam SubscriptionManagerType The type of the subscription manager object
* @param ioContext The io_context to use
* @param ip The ip address of the source
* @param wsPort The port of the source
@@ -105,32 +106,18 @@ public:
* @param connectionTimeout The connection timeout. Defaults to 30 seconds
* @param retryDelay The retry delay. Defaults to 1 second
*/
template <typename NetworkValidatedLedgersType, typename SubscriptionManagerType>
SubscriptionSource(
boost::asio::io_context& ioContext,
std::string const& ip,
std::string const& wsPort,
std::shared_ptr<NetworkValidatedLedgersType> validatedLedgers,
std::shared_ptr<SubscriptionManagerType> subscriptions,
std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
OnConnectHook onConnect,
OnDisconnectHook onDisconnect,
OnLedgerClosedHook onLedgerClosed,
std::chrono::steady_clock::duration const connectionTimeout = CONNECTION_TIMEOUT,
std::chrono::steady_clock::duration const retryDelay = RETRY_DELAY
)
: log_(fmt::format("GrpcSource[{}:{}]", ip, wsPort))
, wsConnectionBuilder_(ip, wsPort)
, dependencies_(std::move(validatedLedgers), std::move(subscriptions))
, strand_(boost::asio::make_strand(ioContext))
, retry_(util::makeRetryExponentialBackoff(retryDelay, RETRY_MAX_DELAY, strand_))
, onConnect_(std::move(onConnect))
, onDisconnect_(std::move(onDisconnect))
, onLedgerClosed_(std::move(onLedgerClosed))
{
wsConnectionBuilder_.addHeader({boost::beast::http::field::user_agent, "clio-client"})
.addHeader({"X-User", "clio-client"})
.setConnectionTimeout(connectionTimeout);
}
);
/**
* @brief Destroy the Subscription Source object
@@ -162,6 +149,14 @@ public:
bool
isConnected() const;
/**
* @brief Get whether the source is forwarding
*
* @return true if the source is forwarding, false otherwise
*/
bool
isForwarding() const;
/**
* @brief Set source forwarding
*

118
src/etl/impl/SubscriptionSourceDependencies.hpp
View File

@@ -1,118 +0,0 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#pragma once
#include <boost/json/object.hpp>
#include <cstdint>
#include <memory>
namespace etl::impl {
class SubscriptionSourceDependencies {
struct Concept;
std::unique_ptr<Concept> pImpl_;
public:
template <typename NetworkValidatedLedgersType, typename SubscriptionManagerType>
SubscriptionSourceDependencies(
std::shared_ptr<NetworkValidatedLedgersType> networkValidatedLedgers,
std::shared_ptr<SubscriptionManagerType> subscriptions
)
: pImpl_{std::make_unique<Model<NetworkValidatedLedgersType, SubscriptionManagerType>>(
std::move(networkValidatedLedgers),
std::move(subscriptions)
)}
{
}
void
forwardProposedTransaction(boost::json::object const& receivedTxJson)
{
pImpl_->forwardProposedTransaction(receivedTxJson);
}
void
forwardValidation(boost::json::object const& validationJson) const
{
pImpl_->forwardValidation(validationJson);
}
void
forwardManifest(boost::json::object const& manifestJson) const
{
pImpl_->forwardManifest(manifestJson);
}
void
pushValidatedLedger(uint32_t const idx)
{
pImpl_->pushValidatedLedger(idx);
}
private:
struct Concept {
virtual ~Concept() = default;
virtual void
forwardProposedTransaction(boost::json::object const& receivedTxJson) = 0;
virtual void
forwardValidation(boost::json::object const& validationJson) const = 0;
virtual void
forwardManifest(boost::json::object const& manifestJson) const = 0;
virtual void
pushValidatedLedger(uint32_t idx) = 0;
};
template <typename SomeNetworkValidatedLedgersType, typename SomeSubscriptionManagerType>
class Model : public Concept {
std::shared_ptr<SomeNetworkValidatedLedgersType> networkValidatedLedgers_;
std::shared_ptr<SomeSubscriptionManagerType> subscriptions_;
public:
Model(
std::shared_ptr<SomeNetworkValidatedLedgersType> networkValidatedLedgers,
std::shared_ptr<SomeSubscriptionManagerType> subscriptions
)
: networkValidatedLedgers_{std::move(networkValidatedLedgers)}, subscriptions_{std::move(subscriptions)}
{
}
void
forwardProposedTransaction(boost::json::object const& receivedTxJson) override
{
subscriptions_->forwardProposedTransaction(receivedTxJson);
}
void
forwardValidation(boost::json::object const& validationJson) const override
{
subscriptions_->forwardValidation(validationJson);
}
void
forwardManifest(boost::json::object const& manifestJson) const override
{
subscriptions_->forwardManifest(manifestJson);
}
void
pushValidatedLedger(uint32_t idx) override
{
networkValidatedLedgers_->push(idx);
}
};
};
} // namespace etl::impl

16
src/feed/SubscriptionManager.cpp
View File

@@ -186,4 +186,20 @@ SubscriptionManager::pubTransaction(data::TransactionAndMetadata const& txMeta,
transactionFeed_.pub(txMeta, lgrInfo, backend_);
}
boost::json::object
SubscriptionManager::report() const
{
return {
{"ledger", ledgerFeed_.count()},
{"transactions", transactionFeed_.transactionSubCount()},
{"transactions_proposed", proposedTransactionFeed_.transactionSubcount()},
{"manifests", manifestFeed_.count()},
{"validations", validationsFeed_.count()},
{"account", transactionFeed_.accountSubCount()},
{"accounts_proposed", proposedTransactionFeed_.accountSubCount()},
{"books", transactionFeed_.bookSubCount()},
{"book_changes", bookChangesFeed_.count()},
};
}
} // namespace feed

66
src/feed/SubscriptionManager.hpp
View File

@@ -21,6 +21,7 @@
#include "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "feed/Types.hpp"
#include "feed/impl/BookChangesFeed.hpp"
#include "feed/impl/ForwardFeed.hpp"
@@ -55,7 +56,7 @@ namespace feed {
/**
* @brief A subscription manager is responsible for managing the subscriptions and publishing the feeds
*/
class SubscriptionManager {
class SubscriptionManager : public SubscriptionManagerInterface {
std::reference_wrapper<boost::asio::io_context> ioContext_;
std::shared_ptr<data::BackendInterface const> backend_;
@@ -93,14 +94,14 @@ public:
* @param subscriber
*/
void
subBookChanges(SubscriberSharedPtr const& subscriber);
subBookChanges(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the book changes feed.
* @param subscriber
*/
void
unsubBookChanges(SubscriberSharedPtr const& subscriber);
unsubBookChanges(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Publish the book changes feed.
@@ -109,21 +110,21 @@ public:
*/
void
pubBookChanges(ripple::LedgerHeader const& lgrInfo, std::vector<data::TransactionAndMetadata> const& transactions)
const;
const final;
/**
* @brief Subscribe to the proposed transactions feed.
* @param subscriber
*/
void
subProposedTransactions(SubscriberSharedPtr const& subscriber);
subProposedTransactions(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the proposed transactions feed.
* @param subscriber
*/
void
unsubProposedTransactions(SubscriberSharedPtr const& subscriber);
unsubProposedTransactions(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Subscribe to the proposed transactions feed, only receive the feed when particular account is affected.
@@ -131,7 +132,7 @@ public:
* @param subscriber
*/
void
subProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
subProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the proposed transactions feed for particular account.
@@ -139,14 +140,14 @@ public:
* @param subscriber
*/
void
unsubProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
unsubProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Forward the proposed transactions feed.
* @param receivedTxJson The proposed transaction json.
*/
void
forwardProposedTransaction(boost::json::object const& receivedTxJson);
forwardProposedTransaction(boost::json::object const& receivedTxJson) final;
/**
* @brief Subscribe to the ledger feed.
@@ -155,14 +156,14 @@ public:
* @return The ledger feed
*/
boost::json::object
subLedger(boost::asio::yield_context yield, SubscriberSharedPtr const& subscriber);
subLedger(boost::asio::yield_context yield, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the ledger feed.
* @param subscriber
*/
void
unsubLedger(SubscriberSharedPtr const& subscriber);
unsubLedger(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Publish the ledger feed.
@@ -177,63 +178,63 @@ public:
ripple::Fees const& fees,
std::string const& ledgerRange,
std::uint32_t txnCount
) const;
) const final;
/**
* @brief Subscribe to the manifest feed.
* @param subscriber
*/
void
subManifest(SubscriberSharedPtr const& subscriber);
subManifest(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the manifest feed.
* @param subscriber
*/
void
unsubManifest(SubscriberSharedPtr const& subscriber);
unsubManifest(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Forward the manifest feed.
* @param manifestJson The manifest json to forward.
*/
void
forwardManifest(boost::json::object const& manifestJson) const;
forwardManifest(boost::json::object const& manifestJson) const final;
/**
* @brief Subscribe to the validation feed.
* @param subscriber
*/
void
subValidation(SubscriberSharedPtr const& subscriber);
subValidation(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the validation feed.
* @param subscriber
*/
void
unsubValidation(SubscriberSharedPtr const& subscriber);
unsubValidation(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Forward the validation feed.
* @param validationJson The validation feed json to forward.
*/
void
forwardValidation(boost::json::object const& validationJson) const;
forwardValidation(boost::json::object const& validationJson) const final;
/**
* @brief Subscribe to the transactions feed.
* @param subscriber
*/
void
subTransactions(SubscriberSharedPtr const& subscriber);
subTransactions(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the transactions feed.
* @param subscriber
*/
void
unsubTransactions(SubscriberSharedPtr const& subscriber);
unsubTransactions(SubscriberSharedPtr const& subscriber) final;
/**
* @brief Subscribe to the transactions feed, only receive the feed when particular account is affected.
@@ -241,7 +242,7 @@ public:
* @param subscriber
*/
void
subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the transactions feed for particular account.
@@ -249,7 +250,7 @@ public:
* @param subscriber The subscriber to unsubscribe
*/
void
unsubAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
unsubAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Subscribe to the transactions feed, only receive feed when particular order book is affected.
@@ -257,7 +258,7 @@ public:
* @param subscriber
*/
void
subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber);
subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Unsubscribe to the transactions feed for particular order book.
@@ -265,7 +266,7 @@ public:
* @param subscriber
*/
void
unsubBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber);
unsubBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) final;
/**
* @brief Forward the transactions feed.
@@ -273,7 +274,7 @@ public:
* @param lgrInfo The ledger header.
*/
void
pubTransaction(data::TransactionAndMetadata const& txMeta, ripple::LedgerHeader const& lgrInfo);
pubTransaction(data::TransactionAndMetadata const& txMeta, ripple::LedgerHeader const& lgrInfo) final;
/**
* @brief Get the number of subscribers.
@@ -281,20 +282,7 @@ public:
* @return The report of the number of subscribers
*/
boost::json::object
report() const
{
return {
{"ledger", ledgerFeed_.count()},
{"transactions", transactionFeed_.transactionSubCount()},
{"transactions_proposed", proposedTransactionFeed_.transactionSubcount()},
{"manifests", manifestFeed_.count()},
{"validations", validationsFeed_.count()},
{"account", transactionFeed_.accountSubCount()},
{"accounts_proposed", proposedTransactionFeed_.accountSubCount()},
{"books", transactionFeed_.bookSubCount()},
{"book_changes", bookChangesFeed_.count()},
};
}
report() const final;
};
/**

244
src/feed/SubscriptionManagerInterface.hpp Normal file
View File

@@ -0,0 +1,244 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#pragma once
#include "data/Types.hpp"
#include "feed/Types.hpp"
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <ripple/protocol/AccountID.h>
#include <ripple/protocol/Book.h>
#include <ripple/protocol/Fees.h>
#include <ripple/protocol/LedgerHeader.h>
#include <cstdint>
#include <string>
#include <vector>
namespace feed {
/**
* @brief Interface of subscription manager.
* A subscription manager is responsible for managing the subscriptions and publishing the feeds.
*/
class SubscriptionManagerInterface {
public:
virtual ~SubscriptionManagerInterface() = default;
/**
* @brief Subscribe to the book changes feed.
* @param subscriber
*/
virtual void
subBookChanges(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the book changes feed.
* @param subscriber
*/
virtual void
unsubBookChanges(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Publish the book changes feed.
* @param lgrInfo The current ledger header.
* @param transactions The transactions in the current ledger.
*/
virtual void
pubBookChanges(ripple::LedgerHeader const& lgrInfo, std::vector<data::TransactionAndMetadata> const& transactions)
const = 0;
/**
* @brief Subscribe to the proposed transactions feed.
* @param subscriber
*/
virtual void
subProposedTransactions(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the proposed transactions feed.
* @param subscriber
*/
virtual void
unsubProposedTransactions(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Subscribe to the proposed transactions feed, only receive the feed when particular account is affected.
* @param account The account to watch.
* @param subscriber
*/
virtual void
subProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the proposed transactions feed for particular account.
* @param account The account to stop watching.
* @param subscriber
*/
virtual void
unsubProposedAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Forward the proposed transactions feed.
* @param receivedTxJson The proposed transaction json.
*/
virtual void
forwardProposedTransaction(boost::json::object const& receivedTxJson) = 0;
/**
* @brief Subscribe to the ledger feed.
* @param yield The coroutine context
* @param subscriber The subscriber to the ledger feed
* @return The ledger feed
*/
virtual boost::json::object
subLedger(boost::asio::yield_context yield, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the ledger feed.
* @param subscriber
*/
virtual void
unsubLedger(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Publish the ledger feed.
* @param lgrInfo The ledger header.
* @param fees The fees.
* @param ledgerRange The ledger range.
* @param txnCount The transaction count.
*/
virtual void
pubLedger(
ripple::LedgerHeader const& lgrInfo,
ripple::Fees const& fees,
std::string const& ledgerRange,
std::uint32_t txnCount
) const = 0;
/**
* @brief Subscribe to the manifest feed.
* @param subscriber
*/
virtual void
subManifest(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the manifest feed.
* @param subscriber
*/
virtual void
unsubManifest(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Forward the manifest feed.
* @param manifestJson The manifest json to forward.
*/
virtual void
forwardManifest(boost::json::object const& manifestJson) const = 0;
/**
* @brief Subscribe to the validation feed.
* @param subscriber
*/
virtual void
subValidation(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the validation feed.
* @param subscriber
*/
virtual void
unsubValidation(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Forward the validation feed.
* @param validationJson The validation feed json to forward.
*/
virtual void
forwardValidation(boost::json::object const& validationJson) const = 0;
/**
* @brief Subscribe to the transactions feed.
* @param subscriber
*/
virtual void
subTransactions(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the transactions feed.
* @param subscriber
*/
virtual void
unsubTransactions(SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Subscribe to the transactions feed, only receive the feed when particular account is affected.
* @param account The account to watch.
* @param subscriber
*/
virtual void
subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the transactions feed for particular account.
* @param account The account to stop watching
* @param subscriber The subscriber to unsubscribe
*/
virtual void
unsubAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Subscribe to the transactions feed, only receive feed when particular order book is affected.
* @param book The book to watch.
* @param subscriber
*/
virtual void
subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Unsubscribe to the transactions feed for particular order book.
* @param book The book to watch.
* @param subscriber
*/
virtual void
unsubBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber) = 0;
/**
* @brief Forward the transactions feed.
* @param txMeta The transaction and metadata.
* @param lgrInfo The ledger header.
*/
virtual void
pubTransaction(data::TransactionAndMetadata const& txMeta, ripple::LedgerHeader const& lgrInfo) = 0;
/**
* @brief Get the number of subscribers.
*
* @return The report of the number of subscribers
*/
virtual boost::json::object
report() const = 0;
};
} // namespace feed

3
src/main/Main.cpp
View File

@@ -18,8 +18,8 @@
//==============================================================================
#include "data/BackendFactory.hpp"
#include "etl/ETLHelpers.hpp"
#include "etl/ETLService.hpp"
#include "etl/NetworkValidatedLedgers.hpp"
#include "feed/SubscriptionManager.hpp"
#include "main/Build.hpp"
#include "rpc/Counters.hpp"
@@ -34,7 +34,6 @@
#include "web/IntervalSweepHandler.hpp"
#include "web/RPCServerHandler.hpp"
#include "web/Server.hpp"
#include "web/WhitelistHandler.hpp"
#include <boost/asio/buffer.hpp>
#include <boost/asio/io_context.hpp>

2
src/rpc/CMakeLists.txt
View File

@@ -39,7 +39,9 @@ target_sources(
handlers/NFTSellOffers.cpp
handlers/NoRippleCheck.cpp
handlers/Random.cpp
handlers/Subscribe.cpp
handlers/TransactionEntry.cpp
handlers/Unsubscribe.cpp
)
target_link_libraries(clio_rpc PRIVATE clio_util)

14
src/rpc/handlers/ServerInfo.hpp
View File

@@ -21,6 +21,7 @@
#include "data/BackendInterface.hpp"
#include "data/DBHelpers.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "main/Build.hpp"
#include "rpc/Errors.hpp"
#include "rpc/JS.hpp"
@@ -51,9 +52,6 @@ namespace etl {
class ETLService;
class LoadBalancer;
} // namespace etl
namespace feed {
class SubscriptionManager;
} // namespace feed
namespace rpc {
class Counters;
} // namespace rpc
@@ -63,17 +61,16 @@ namespace rpc {
/**
* @brief Contains common functionality for handling the `server_info` command
*
* @tparam SubscriptionManagerType The type of the subscription manager
* @tparam LoadBalancerType The type of the load balancer
* @tparam ETLServiceType The type of the ETL service
* @tparam CountersType The type of the counters
*/
template <typename SubscriptionManagerType, typename LoadBalancerType, typename ETLServiceType, typename CountersType>
template <typename LoadBalancerType, typename ETLServiceType, typename CountersType>
class BaseServerInfoHandler {
static constexpr auto BACKEND_COUNTERS_KEY = "backend_counters";
std::shared_ptr<BackendInterface> backend_;
std::shared_ptr<SubscriptionManagerType> subscriptions_;
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
std::shared_ptr<LoadBalancerType> balancer_;
std::shared_ptr<ETLServiceType const> etl_;
std::reference_wrapper<CountersType const> counters_;
@@ -159,7 +156,7 @@ public:
*/
BaseServerInfoHandler(
std::shared_ptr<BackendInterface> const& backend,
std::shared_ptr<SubscriptionManagerType> const& subscriptions,
std::shared_ptr<feed::SubscriptionManagerInterface> const& subscriptions,
std::shared_ptr<LoadBalancerType> const& balancer,
std::shared_ptr<ETLServiceType const> const& etl,
CountersType const& counters
@@ -352,7 +349,6 @@ private:
*
* For more details see: https://xrpl.org/server_info-clio.html
*/
using ServerInfoHandler =
BaseServerInfoHandler<feed::SubscriptionManager, etl::LoadBalancer, etl::ETLService, Counters>;
using ServerInfoHandler = BaseServerInfoHandler<etl::LoadBalancer, etl::ETLService, Counters>;
} // namespace rpc

303
src/rpc/handlers/Subscribe.cpp Normal file
View File

@@ -0,0 +1,303 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include "rpc/handlers/Subscribe.hpp"
#include "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/Errors.hpp"
#include "rpc/JS.hpp"
#include "rpc/RPCHelpers.hpp"
#include "rpc/common/Checkers.hpp"
#include "rpc/common/MetaProcessors.hpp"
#include "rpc/common/Specs.hpp"
#include "rpc/common/Types.hpp"
#include "rpc/common/Validators.hpp"
#include <boost/asio/spawn.hpp>
#include <boost/json/array.hpp>
#include <boost/json/conversion.hpp>
#include <boost/json/object.hpp>
#include <boost/json/value.hpp>
#include <boost/json/value_to.hpp>
#include <ripple/beast/utility/Zero.h>
#include <ripple/protocol/Book.h>
#include <ripple/protocol/ErrorCodes.h>
#include <ripple/protocol/jss.h>
#include <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <variant>
#include <vector>
namespace rpc {
SubscribeHandler::SubscribeHandler(
std::shared_ptr<BackendInterface> const& sharedPtrBackend,
std::shared_ptr<feed::SubscriptionManagerInterface> const& subscriptions
)
: sharedPtrBackend_(sharedPtrBackend), subscriptions_(subscriptions)
{
}
RpcSpecConstRef
SubscribeHandler::spec([[maybe_unused]] uint32_t apiVersion)
{
static auto const booksValidator =
validation::CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
for (auto const& book : value.as_array()) {
if (!book.is_object())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "ItemNotObject"}};
if (book.as_object().contains("both") && !book.as_object().at("both").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "bothNotBool"}};
if (book.as_object().contains("snapshot") && !book.as_object().at("snapshot").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "snapshotNotBool"}};
if (book.as_object().contains("taker")) {
if (auto err = meta::WithCustomError(
validation::AccountValidator,
Status{RippledError::rpcBAD_ISSUER, "Issuer account malformed."}
)
.verify(book.as_object(), "taker");
!err)
return err;
}
auto const parsedBook = parseBook(book.as_object());
if (auto const status = std::get_if<Status>(&parsedBook))
return Error(*status);
}
return MaybeError{};
}};
static auto const rpcSpec = RpcSpec{
{JS(streams), validation::SubscribeStreamValidator},
{JS(accounts), validation::SubscribeAccountsValidator},
{JS(accounts_proposed), validation::SubscribeAccountsValidator},
{JS(books), booksValidator},
{"user", check::Deprecated{}},
{JS(password), check::Deprecated{}},
{JS(rt_accounts), check::Deprecated{}}
};
return rpcSpec;
}
SubscribeHandler::Result
SubscribeHandler::process(Input input, Context const& ctx) const
{
auto output = Output{};
// Mimic rippled. No matter what the request is, the api version changes for the whole session
ctx.session->apiSubVersion = ctx.apiVersion;
if (input.streams) {
auto const ledger = subscribeToStreams(ctx.yield, *(input.streams), ctx.session);
if (!ledger.empty())
output.ledger = ledger;
}
if (input.accounts)
subscribeToAccounts(*(input.accounts), ctx.session);
if (input.accountsProposed)
subscribeToAccountsProposed(*(input.accountsProposed), ctx.session);
if (input.books)
subscribeToBooks(*(input.books), ctx.session, ctx.yield, output);
return output;
}
boost::json::object
SubscribeHandler::subscribeToStreams(
boost::asio::yield_context yield,
std::vector<std::string> const& streams,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
auto response = boost::json::object{};
for (auto const& stream : streams) {
if (stream == "ledger") {
response = subscriptions_->subLedger(yield, session);
} else if (stream == "transactions") {
subscriptions_->subTransactions(session);
} else if (stream == "transactions_proposed") {
subscriptions_->subProposedTransactions(session);
} else if (stream == "validations") {
subscriptions_->subValidation(session);
} else if (stream == "manifests") {
subscriptions_->subManifest(session);
} else if (stream == "book_changes") {
subscriptions_->subBookChanges(session);
}
}
return response;
}
void
SubscribeHandler::subscribeToAccountsProposed(
std::vector<std::string> const& accounts,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->subProposedAccount(*accountID, session);
}
}
void
SubscribeHandler::subscribeToAccounts(
std::vector<std::string> const& accounts,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->subAccount(*accountID, session);
}
}
void
SubscribeHandler::subscribeToBooks(
std::vector<OrderBook> const& books,
std::shared_ptr<web::ConnectionBase> const& session,
boost::asio::yield_context yield,
Output& output
) const
{
static auto constexpr fetchLimit = 200;
std::optional<data::LedgerRange> rng;
for (auto const& internalBook : books) {
if (internalBook.snapshot) {
if (!rng)
rng = sharedPtrBackend_->fetchLedgerRange();
auto const getOrderBook = [&](auto const& book, auto& snapshots) {
auto const bookBase = getBookBase(book);
auto const [offers, _] =
sharedPtrBackend_->fetchBookOffers(bookBase, rng->maxSequence, fetchLimit, yield);
// the taker is not really uesed, same issue with
// https://github.com/XRPLF/xrpl-dev-portal/issues/1818
auto const takerID = internalBook.taker ? accountFromStringStrict(*(internalBook.taker)) : beast::zero;
auto const orderBook =
postProcessOrderBook(offers, book, *takerID, *sharedPtrBackend_, rng->maxSequence, yield);
std::copy(orderBook.begin(), orderBook.end(), std::back_inserter(snapshots));
};
if (internalBook.both) {
if (!output.bids)
output.bids = boost::json::array();
if (!output.asks)
output.asks = boost::json::array();
getOrderBook(internalBook.book, *(output.bids));
getOrderBook(ripple::reversed(internalBook.book), *(output.asks));
} else {
if (!output.offers)
output.offers = boost::json::array();
getOrderBook(internalBook.book, *(output.offers));
}
}
subscriptions_->subBook(internalBook.book, session);
if (internalBook.both)
subscriptions_->subBook(ripple::reversed(internalBook.book), session);
}
}
void
tag_invoke(boost::json::value_from_tag, boost::json::value& jv, SubscribeHandler::Output const& output)
{
jv = output.ledger ? *(output.ledger) : boost::json::object();
if (output.offers)
jv.as_object().emplace(JS(offers), *(output.offers));
if (output.asks)
jv.as_object().emplace(JS(asks), *(output.asks));
if (output.bids)
jv.as_object().emplace(JS(bids), *(output.bids));
}
SubscribeHandler::Input
tag_invoke(boost::json::value_to_tag<SubscribeHandler::Input>, boost::json::value const& jv)
{
auto input = SubscribeHandler::Input{};
auto const& jsonObject = jv.as_object();
if (auto const& streams = jsonObject.find(JS(streams)); streams != jsonObject.end()) {
input.streams = std::vector<std::string>();
for (auto const& stream : streams->value().as_array())
input.streams->push_back(boost::json::value_to<std::string>(stream));
}
if (auto const& accounts = jsonObject.find(JS(accounts)); accounts != jsonObject.end()) {
input.accounts = std::vector<std::string>();
for (auto const& account : accounts->value().as_array())
input.accounts->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& accountsProposed = jsonObject.find(JS(accounts_proposed)); accountsProposed != jsonObject.end()) {
input.accountsProposed = std::vector<std::string>();
for (auto const& account : accountsProposed->value().as_array())
input.accountsProposed->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& books = jsonObject.find(JS(books)); books != jsonObject.end()) {
input.books = std::vector<SubscribeHandler::OrderBook>();
for (auto const& book : books->value().as_array()) {
auto internalBook = SubscribeHandler::OrderBook{};
auto const& bookObject = book.as_object();
if (auto const& taker = bookObject.find(JS(taker)); taker != bookObject.end())
internalBook.taker = boost::json::value_to<std::string>(taker->value());
if (auto const& both = bookObject.find(JS(both)); both != bookObject.end())
internalBook.both = both->value().as_bool();
if (auto const& snapshot = bookObject.find(JS(snapshot)); snapshot != bookObject.end())
internalBook.snapshot = snapshot->value().as_bool();
auto const parsedBookMaybe = parseBook(book.as_object());
internalBook.book = std::get<ripple::Book>(parsedBookMaybe);
input.books->push_back(internalBook);
}
}
return input;
}
} // namespace rpc

273
src/rpc/handlers/Subscribe.hpp
View File

@@ -20,14 +20,9 @@
#pragma once
#include "data/BackendInterface.hpp"
#include "data/Types.hpp"
#include "rpc/Errors.hpp"
#include "rpc/JS.hpp"
#include "rpc/RPCHelpers.hpp"
#include "rpc/common/Checkers.hpp"
#include "rpc/common/MetaProcessors.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/common/Specs.hpp"
#include "rpc/common/Types.hpp"
#include "rpc/common/Validators.hpp"
#include <boost/asio/spawn.hpp>
#include <boost/json/array.hpp>
@@ -44,25 +39,20 @@
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <variant>
#include <vector>
namespace feed {
class SubscriptionManager;
} // namespace feed
namespace rpc {
/**
* @brief Contains functionality for handling the `subscribe` command
* @brief Contains functionality for handling the `subscribe` command.
* The subscribe method requests periodic notifications from the server when certain events happen.
*
* @tparam SubscriptionManagerType The type of the subscription manager to use
* For more details see: https://xrpl.org/subscribe.html
*/
template <typename SubscriptionManagerType>
class BaseSubscribeHandler {
class SubscribeHandler {
std::shared_ptr<BackendInterface> sharedPtrBackend_;
std::shared_ptr<SubscriptionManagerType> subscriptions_;
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
public:
/**
@@ -109,13 +99,10 @@ public:
* @param sharedPtrBackend The backend to use
* @param subscriptions The subscription manager to use
*/
BaseSubscribeHandler(
SubscribeHandler(
std::shared_ptr<BackendInterface> const& sharedPtrBackend,
std::shared_ptr<SubscriptionManagerType> const& subscriptions
)
: sharedPtrBackend_(sharedPtrBackend), subscriptions_(subscriptions)
{
}
std::shared_ptr<feed::SubscriptionManagerInterface> const& subscriptions
);
/**
* @brief Returns the API specification for the command
@@ -124,53 +111,7 @@ public:
* @return The spec for the given apiVersion
*/
static RpcSpecConstRef
spec([[maybe_unused]] uint32_t apiVersion)
{
static auto const booksValidator =
validation::CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
for (auto const& book : value.as_array()) {
if (!book.is_object())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "ItemNotObject"}};
if (book.as_object().contains("both") && !book.as_object().at("both").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "bothNotBool"}};
if (book.as_object().contains("snapshot") && !book.as_object().at("snapshot").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "snapshotNotBool"}};
if (book.as_object().contains("taker")) {
if (auto err = meta::WithCustomError(
validation::AccountValidator,
Status{RippledError::rpcBAD_ISSUER, "Issuer account malformed."}
)
.verify(book.as_object(), "taker");
!err)
return err;
}
auto const parsedBook = parseBook(book.as_object());
if (auto const status = std::get_if<Status>(&parsedBook))
return Error(*status);
}
return MaybeError{};
}};
static auto const rpcSpec = RpcSpec{
{JS(streams), validation::SubscribeStreamValidator},
{JS(accounts), validation::SubscribeAccountsValidator},
{JS(accounts_proposed), validation::SubscribeAccountsValidator},
{JS(books), booksValidator},
{"user", check::Deprecated{}},
{JS(password), check::Deprecated{}},
{JS(rt_accounts), check::Deprecated{}}
};
return rpcSpec;
}
spec([[maybe_unused]] uint32_t apiVersion);
/**
* @brief Process the Subscribe command
@@ -180,30 +121,7 @@ public:
* @return The result of the operation
*/
Result
process(Input input, Context const& ctx) const
{
auto output = Output{};
// Mimic rippled. No matter what the request is, the api version changes for the whole session
ctx.session->apiSubVersion = ctx.apiVersion;
if (input.streams) {
auto const ledger = subscribeToStreams(ctx.yield, *(input.streams), ctx.session);
if (!ledger.empty())
output.ledger = ledger;
}
if (input.accounts)
subscribeToAccounts(*(input.accounts), ctx.session);
if (input.accountsProposed)
subscribeToAccountsProposed(*(input.accountsProposed), ctx.session);
if (input.books)
subscribeToBooks(*(input.books), ctx.session, ctx.yield, output);
return output;
}
process(Input input, Context const& ctx) const;
private:
boost::json::object
@@ -211,50 +129,17 @@ private:
boost::asio::yield_context yield,
std::vector<std::string> const& streams,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
auto response = boost::json::object{};
for (auto const& stream : streams) {
if (stream == "ledger") {
response = subscriptions_->subLedger(yield, session);
} else if (stream == "transactions") {
subscriptions_->subTransactions(session);
} else if (stream == "transactions_proposed") {
subscriptions_->subProposedTransactions(session);
} else if (stream == "validations") {
subscriptions_->subValidation(session);
} else if (stream == "manifests") {
subscriptions_->subManifest(session);
} else if (stream == "book_changes") {
subscriptions_->subBookChanges(session);
}
}
return response;
}
) const;
void
subscribeToAccounts(std::vector<std::string> const& accounts, std::shared_ptr<web::ConnectionBase> const& session)
const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->subAccount(*accountID, session);
}
}
const;
void
subscribeToAccountsProposed(
std::vector<std::string> const& accounts,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->subProposedAccount(*accountID, session);
}
}
) const;
void
subscribeToBooks(
@@ -262,121 +147,25 @@ private:
std::shared_ptr<web::ConnectionBase> const& session,
boost::asio::yield_context yield,
Output& output
) const
{
static auto constexpr fetchLimit = 200;
std::optional<data::LedgerRange> rng;
for (auto const& internalBook : books) {
if (internalBook.snapshot) {
if (!rng)
rng = sharedPtrBackend_->fetchLedgerRange();
auto const getOrderBook = [&](auto const& book, auto& snapshots) {
auto const bookBase = getBookBase(book);
auto const [offers, _] =
sharedPtrBackend_->fetchBookOffers(bookBase, rng->maxSequence, fetchLimit, yield);
// the taker is not really uesed, same issue with
// https://github.com/XRPLF/xrpl-dev-portal/issues/1818
auto const takerID =
internalBook.taker ? accountFromStringStrict(*(internalBook.taker)) : beast::zero;
auto const orderBook =
postProcessOrderBook(offers, book, *takerID, *sharedPtrBackend_, rng->maxSequence, yield);
std::copy(orderBook.begin(), orderBook.end(), std::back_inserter(snapshots));
};
if (internalBook.both) {
if (!output.bids)
output.bids = boost::json::array();
if (!output.asks)
output.asks = boost::json::array();
getOrderBook(internalBook.book, *(output.bids));
getOrderBook(ripple::reversed(internalBook.book), *(output.asks));
} else {
if (!output.offers)
output.offers = boost::json::array();
getOrderBook(internalBook.book, *(output.offers));
}
}
subscriptions_->subBook(internalBook.book, session);
if (internalBook.both)
subscriptions_->subBook(ripple::reversed(internalBook.book), session);
}
}
) const;
/**
* @brief Convert output to json value
*
* @param jv The json value to convert to
* @param output The output to convert from
*/
friend void
tag_invoke(boost::json::value_from_tag, boost::json::value& jv, Output const& output)
{
jv = output.ledger ? *(output.ledger) : boost::json::object();
if (output.offers)
jv.as_object().emplace(JS(offers), *(output.offers));
if (output.asks)
jv.as_object().emplace(JS(asks), *(output.asks));
if (output.bids)
jv.as_object().emplace(JS(bids), *(output.bids));
}
tag_invoke(boost::json::value_from_tag, boost::json::value& jv, Output const& output);
/**
* @brief Convert json value to input
*
* @param jv The json value to convert from
* @return The input to convert to
*/
friend Input
tag_invoke(boost::json::value_to_tag<Input>, boost::json::value const& jv)
{
auto input = Input{};
auto const& jsonObject = jv.as_object();
if (auto const& streams = jsonObject.find(JS(streams)); streams != jsonObject.end()) {
input.streams = std::vector<std::string>();
for (auto const& stream : streams->value().as_array())
input.streams->push_back(boost::json::value_to<std::string>(stream));
}
if (auto const& accounts = jsonObject.find(JS(accounts)); accounts != jsonObject.end()) {
input.accounts = std::vector<std::string>();
for (auto const& account : accounts->value().as_array())
input.accounts->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& accountsProposed = jsonObject.find(JS(accounts_proposed));
accountsProposed != jsonObject.end()) {
input.accountsProposed = std::vector<std::string>();
for (auto const& account : accountsProposed->value().as_array())
input.accountsProposed->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& books = jsonObject.find(JS(books)); books != jsonObject.end()) {
input.books = std::vector<OrderBook>();
for (auto const& book : books->value().as_array()) {
auto internalBook = OrderBook{};
auto const& bookObject = book.as_object();
if (auto const& taker = bookObject.find(JS(taker)); taker != bookObject.end())
internalBook.taker = boost::json::value_to<std::string>(taker->value());
if (auto const& both = bookObject.find(JS(both)); both != bookObject.end())
internalBook.both = both->value().as_bool();
if (auto const& snapshot = bookObject.find(JS(snapshot)); snapshot != bookObject.end())
internalBook.snapshot = snapshot->value().as_bool();
auto const parsedBookMaybe = parseBook(book.as_object());
internalBook.book = std::get<ripple::Book>(parsedBookMaybe);
input.books->push_back(internalBook);
}
}
return input;
}
tag_invoke(boost::json::value_to_tag<Input>, boost::json::value const& jv);
};
/**
* @brief The subscribe method requests periodic notifications from the server when certain events happen.
*
* For more details see: https://xrpl.org/subscribe.html
*/
using SubscribeHandler = BaseSubscribeHandler<feed::SubscriptionManager>;
} // namespace rpc

208
src/rpc/handlers/Unsubscribe.cpp Normal file
View File

@@ -0,0 +1,208 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2024, the clio developers.
Permission to use, copy, modify, and distribute this software for any
purpose with or without fee is hereby granted, provided that the above
copyright notice and this permission notice appear in all copies.
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include "rpc/handlers/Unsubscribe.hpp"
#include "data/BackendInterface.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/Errors.hpp"
#include "rpc/JS.hpp"
#include "rpc/RPCHelpers.hpp"
#include "rpc/common/Checkers.hpp"
#include "rpc/common/Specs.hpp"
#include "rpc/common/Types.hpp"
#include "rpc/common/Validators.hpp"
#include "util/Assert.hpp"
#include <boost/json/conversion.hpp>
#include <boost/json/value.hpp>
#include <boost/json/value_to.hpp>
#include <ripple/protocol/Book.h>
#include <ripple/protocol/ErrorCodes.h>
#include <ripple/protocol/jss.h>
#include <cstdint>
#include <memory>
#include <string>
#include <string_view>
#include <variant>
#include <vector>
namespace rpc {
UnsubscribeHandler::UnsubscribeHandler(
std::shared_ptr<BackendInterface> const& sharedPtrBackend,
std::shared_ptr<feed::SubscriptionManagerInterface> const& subscriptions
)
: sharedPtrBackend_(sharedPtrBackend), subscriptions_(subscriptions)
{
}
RpcSpecConstRef
UnsubscribeHandler::spec([[maybe_unused]] uint32_t apiVersion)
{
static auto const booksValidator =
validation::CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
for (auto const& book : value.as_array()) {
if (!book.is_object())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "ItemNotObject"}};
if (book.as_object().contains("both") && !book.as_object().at("both").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "bothNotBool"}};
auto const parsedBook = parseBook(book.as_object());
if (auto const status = std::get_if<Status>(&parsedBook))
return Error(*status);
}
return MaybeError{};
}};
static auto const rpcSpec = RpcSpec{
{JS(streams), validation::SubscribeStreamValidator},
{JS(accounts), validation::SubscribeAccountsValidator},
{JS(accounts_proposed), validation::SubscribeAccountsValidator},
{JS(books), booksValidator},
{JS(url), check::Deprecated{}},
{JS(rt_accounts), check::Deprecated{}},
{"rt_transactions", check::Deprecated{}},
};
return rpcSpec;
}
UnsubscribeHandler::Result
UnsubscribeHandler::process(Input input, Context const& ctx) const
{
if (input.streams)
unsubscribeFromStreams(*(input.streams), ctx.session);
if (input.accounts)
unsubscribeFromAccounts(*(input.accounts), ctx.session);
if (input.accountsProposed)
unsubscribeFromProposedAccounts(*(input.accountsProposed), ctx.session);
if (input.books)
unsubscribeFromBooks(*(input.books), ctx.session);
return Output{};
}
void
UnsubscribeHandler::unsubscribeFromStreams(
std::vector<std::string> const& streams,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& stream : streams) {
if (stream == "ledger") {
subscriptions_->unsubLedger(session);
} else if (stream == "transactions") {
subscriptions_->unsubTransactions(session);
} else if (stream == "transactions_proposed") {
subscriptions_->unsubProposedTransactions(session);
} else if (stream == "validations") {
subscriptions_->unsubValidation(session);
} else if (stream == "manifests") {
subscriptions_->unsubManifest(session);
} else if (stream == "book_changes") {
subscriptions_->unsubBookChanges(session);
} else {
ASSERT(false, "Unknown stream: {}", stream);
}
}
}
void
UnsubscribeHandler::unsubscribeFromAccounts(
std::vector<std::string> accounts,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->unsubAccount(*accountID, session);
}
}
void
UnsubscribeHandler::unsubscribeFromProposedAccounts(
std::vector<std::string> accountsProposed,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accountsProposed) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->unsubProposedAccount(*accountID, session);
}
}
void
UnsubscribeHandler::unsubscribeFromBooks(
std::vector<OrderBook> const& books,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& orderBook : books) {
subscriptions_->unsubBook(orderBook.book, session);
if (orderBook.both)
subscriptions_->unsubBook(ripple::reversed(orderBook.book), session);
}
}
UnsubscribeHandler::Input
tag_invoke(boost::json::value_to_tag<UnsubscribeHandler::Input>, boost::json::value const& jv)
{
auto input = UnsubscribeHandler::Input{};
auto const& jsonObject = jv.as_object();
if (auto const& streams = jsonObject.find(JS(streams)); streams != jsonObject.end()) {
input.streams = std::vector<std::string>();
for (auto const& stream : streams->value().as_array())
input.streams->push_back(boost::json::value_to<std::string>(stream));
}
if (auto const& accounts = jsonObject.find(JS(accounts)); accounts != jsonObject.end()) {
input.accounts = std::vector<std::string>();
for (auto const& account : accounts->value().as_array())
input.accounts->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& accountsProposed = jsonObject.find(JS(accounts_proposed)); accountsProposed != jsonObject.end()) {
input.accountsProposed = std::vector<std::string>();
for (auto const& account : accountsProposed->value().as_array())
input.accountsProposed->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& books = jsonObject.find(JS(books)); books != jsonObject.end()) {
input.books = std::vector<UnsubscribeHandler::OrderBook>();
for (auto const& book : books->value().as_array()) {
auto internalBook = UnsubscribeHandler::OrderBook{};
auto const& bookObject = book.as_object();
if (auto const& both = bookObject.find(JS(both)); both != bookObject.end())
internalBook.both = both->value().as_bool();
auto const parsedBookMaybe = parseBook(book.as_object());
internalBook.book = std::get<ripple::Book>(parsedBookMaybe);
input.books->push_back(internalBook);
}
}
return input;
}
} // namespace rpc

184
src/rpc/handlers/Unsubscribe.hpp
View File

@@ -20,17 +20,13 @@
#pragma once
#include "data/BackendInterface.hpp"
#include "rpc/Errors.hpp"
#include "rpc/JS.hpp"
#include "rpc/RPCHelpers.hpp"
#include "rpc/common/Checkers.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/common/Specs.hpp"
#include "rpc/common/Types.hpp"
#include "rpc/common/Validators.hpp"
#include "util/Assert.hpp"
#include <boost/json/conversion.hpp>
#include <boost/json/value.hpp>
#include <boost/json/value_to.hpp>
#include <ripple/protocol/Book.h>
#include <ripple/protocol/ErrorCodes.h>
#include <ripple/protocol/jss.h>
@@ -39,8 +35,6 @@
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <variant>
#include <vector>
namespace feed {
@@ -50,12 +44,16 @@ class SubscriptionManager;
namespace rpc {
/**
* @brief Handles the `unsubscribe` command which is used to disconnect a subscriber from a feed
* @brief Handles the `unsubscribe` command which is used to disconnect a subscriber from a feed.
* The unsubscribe command tells the server to stop sending messages for a particular subscription or set of
* subscriptions.
*
* For more details see: https://xrpl.org/unsubscribe.html
*/
template <typename SubscriptionManagerType>
class BaseUnsubscribeHandler {
class UnsubscribeHandler {
std::shared_ptr<BackendInterface> sharedPtrBackend_;
std::shared_ptr<SubscriptionManagerType> subscriptions_;
std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions_;
public:
/**
@@ -85,13 +83,10 @@ public:
* @param sharedPtrBackend The backend to use
* @param subscriptions The subscription manager to use
*/
BaseUnsubscribeHandler(
UnsubscribeHandler(
std::shared_ptr<BackendInterface> const& sharedPtrBackend,
std::shared_ptr<SubscriptionManagerType> const& subscriptions
)
: sharedPtrBackend_(sharedPtrBackend), subscriptions_(subscriptions)
{
}
std::shared_ptr<feed::SubscriptionManagerInterface> const& subscriptions
);
/**
* @brief Returns the API specification for the command
@@ -100,40 +95,7 @@ public:
* @return The spec for the given apiVersion
*/
static RpcSpecConstRef
spec([[maybe_unused]] uint32_t apiVersion)
{
static auto const booksValidator =
validation::CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
for (auto const& book : value.as_array()) {
if (!book.is_object())
return Error{Status{RippledError::rpcINVALID_PARAMS, std::string(key) + "ItemNotObject"}};
if (book.as_object().contains("both") && !book.as_object().at("both").is_bool())
return Error{Status{RippledError::rpcINVALID_PARAMS, "bothNotBool"}};
auto const parsedBook = parseBook(book.as_object());
if (auto const status = std::get_if<Status>(&parsedBook))
return Error(*status);
}
return MaybeError{};
}};
static auto const rpcSpec = RpcSpec{
{JS(streams), validation::SubscribeStreamValidator},
{JS(accounts), validation::SubscribeAccountsValidator},
{JS(accounts_proposed), validation::SubscribeAccountsValidator},
{JS(books), booksValidator},
{JS(url), check::Deprecated{}},
{JS(rt_accounts), check::Deprecated{}},
{"rt_transactions", check::Deprecated{}},
};
return rpcSpec;
}
spec([[maybe_unused]] uint32_t apiVersion);
/**
* @brief Process the Unsubscribe command
@@ -143,127 +105,35 @@ public:
* @return The result of the operation
*/
Result
process(Input input, Context const& ctx) const
{
if (input.streams)
unsubscribeFromStreams(*(input.streams), ctx.session);
if (input.accounts)
unsubscribeFromAccounts(*(input.accounts), ctx.session);
if (input.accountsProposed)
unsubscribeFromProposedAccounts(*(input.accountsProposed), ctx.session);
if (input.books)
unsubscribeFromBooks(*(input.books), ctx.session);
return Output{};
}
process(Input input, Context const& ctx) const;
private:
void
unsubscribeFromStreams(std::vector<std::string> const& streams, std::shared_ptr<web::ConnectionBase> const& session)
const
{
for (auto const& stream : streams) {
if (stream == "ledger") {
subscriptions_->unsubLedger(session);
} else if (stream == "transactions") {
subscriptions_->unsubTransactions(session);
} else if (stream == "transactions_proposed") {
subscriptions_->unsubProposedTransactions(session);
} else if (stream == "validations") {
subscriptions_->unsubValidation(session);
} else if (stream == "manifests") {
subscriptions_->unsubManifest(session);
} else if (stream == "book_changes") {
subscriptions_->unsubBookChanges(session);
} else {
ASSERT(false, "Unknown stream: {}", stream);
}
}
}
const;
void
unsubscribeFromAccounts(std::vector<std::string> accounts, std::shared_ptr<web::ConnectionBase> const& session)
const
{
for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->unsubAccount(*accountID, session);
}
}
const;
void
unsubscribeFromProposedAccounts(
std::vector<std::string> accountsProposed,
std::shared_ptr<web::ConnectionBase> const& session
) const
{
for (auto const& account : accountsProposed) {
auto const accountID = accountFromStringStrict(account);
subscriptions_->unsubProposedAccount(*accountID, session);
}
}
) const;
void
unsubscribeFromBooks(std::vector<OrderBook> const& books, std::shared_ptr<web::ConnectionBase> const& session) const
{
for (auto const& orderBook : books) {
subscriptions_->unsubBook(orderBook.book, session);
if (orderBook.both)
subscriptions_->unsubBook(ripple::reversed(orderBook.book), session);
}
}
unsubscribeFromBooks(std::vector<OrderBook> const& books, std::shared_ptr<web::ConnectionBase> const& session)
const;
/**
* @brief Convert a JSON object to an Input
*
* @param jv The JSON object to convert
* @return The Input object
*/
friend Input
tag_invoke(boost::json::value_to_tag<Input>, boost::json::value const& jv)
{
auto input = Input{};
auto const& jsonObject = jv.as_object();
if (auto const& streams = jsonObject.find(JS(streams)); streams != jsonObject.end()) {
input.streams = std::vector<std::string>();
for (auto const& stream : streams->value().as_array())
input.streams->push_back(boost::json::value_to<std::string>(stream));
}
if (auto const& accounts = jsonObject.find(JS(accounts)); accounts != jsonObject.end()) {
input.accounts = std::vector<std::string>();
for (auto const& account : accounts->value().as_array())
input.accounts->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& accountsProposed = jsonObject.find(JS(accounts_proposed));
accountsProposed != jsonObject.end()) {
input.accountsProposed = std::vector<std::string>();
for (auto const& account : accountsProposed->value().as_array())
input.accountsProposed->push_back(boost::json::value_to<std::string>(account));
}
if (auto const& books = jsonObject.find(JS(books)); books != jsonObject.end()) {
input.books = std::vector<OrderBook>();
for (auto const& book : books->value().as_array()) {
auto internalBook = OrderBook{};
auto const& bookObject = book.as_object();
if (auto const& both = bookObject.find(JS(both)); both != bookObject.end())
internalBook.both = both->value().as_bool();
auto const parsedBookMaybe = parseBook(book.as_object());
internalBook.book = std::get<ripple::Book>(parsedBookMaybe);
input.books->push_back(internalBook);
}
}
return input;
}
tag_invoke(boost::json::value_to_tag<Input>, boost::json::value const& jv);
};
/**
* @brief The unsubscribe command tells the server to stop sending messages for a particular subscription or set of
* subscriptions.
*
* For more details see: https://xrpl.org/unsubscribe.html
*/
using UnsubscribeHandler = BaseUnsubscribeHandler<feed::SubscriptionManager>;
} // namespace rpc

8
src/util/Random.cpp
View File

@@ -16,13 +16,21 @@
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
#include "util/Random.hpp"
#include <chrono>
#include <cstddef>
#include <random>
namespace util {
void
Random::setSeed(size_t seed)
{
generator_.seed(seed);
}
std::mt19937_64 Random::generator_{std::chrono::system_clock::now().time_since_epoch().count()};
} // namespace util

9
src/util/Random.hpp
View File

@@ -20,6 +20,7 @@
#include "util/Assert.hpp"
#include <cstddef>
#include <random>
namespace util {
@@ -50,6 +51,14 @@ public:
return distribution(generator_);
}
/**
* @brief Set the seed for the random number generator
*
* @param seed Seed to set
*/
static void
setSeed(size_t seed);
private:
static std::mt19937_64 generator_;
};