ARCHIVE/clio
1
0
Fork 0
mirror of https://github.com/XRPLF/clio.git synced 2025-12-06 17:27:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Subscribe handler (#591)

Browse Source 
Fixes #593
This commit is contained in:
cyan317
2023-04-13 14:14:11 +01:00
committed by GitHub
parent 36bb20806e
commit 0bc84fefbf
17 changed files with 1365 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/rpc/common/Types.h
View File

@@ -26,6 +26,7 @@
#include <boost/json/value.hpp>
class WsBase;
class SubscriptionManager;
namespace RPCng {
/**
@@ -64,7 +65,7 @@ struct Context
// TODO: we shall change yield_context to const yield_context after we
// update backend interfaces to use const& yield
std::reference_wrapper<boost::asio::yield_context> yield;
std::shared_ptr<WsBase const> session;
std::shared_ptr<WsBase> session;
bool isAdmin = false;
std::string clientIp;
};

69
src/rpc/common/Validators.cpp
View File

@@ -134,7 +134,7 @@ CustomValidator AccountValidator =
// remove all old handler, this function can be moved to here
if (!RPC::accountFromStringStrict(value.as_string().c_str()))
{
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, std::string(key) + "Malformed"}};
return Error{RPC::Status{RPC::RippledError::rpcACT_MALFORMED, std::string(key) + "Malformed"}};
}
return MaybeError{};
}};
@@ -199,4 +199,71 @@ CustomValidator IssuerValidator =
return MaybeError{};
}};
CustomValidator SubscribeStreamValidator =
CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
static std::unordered_set<std::string> const validStreams = {
"ledger", "transactions", "transactions_proposed", "book_changes", "manifests", "validations"};
if (!value.is_array())
{
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
}
for (auto const& v : value.as_array())
{
if (!v.is_string())
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, "streamNotString"}};
if (not validStreams.contains(v.as_string().c_str()))
return Error{RPC::Status{RPC::RippledError::rpcSTREAM_MALFORMED}};
}
return MaybeError{};
}};
CustomValidator SubscribeAccountsValidator =
CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
if (value.as_array().size() == 0)
return Error{RPC::Status{RPC::RippledError::rpcACT_MALFORMED, std::string(key) + " malformed."}};
for (auto const& v : value.as_array())
{
auto obj = boost::json::object();
auto const keyItem = std::string(key) + "'sItem";
obj[keyItem] = v;
if (auto const err = AccountValidator.verify(obj, keyItem); !err)
return err;
}
return MaybeError{};
}};
CustomValidator SubscribeBooksValidator =
CustomValidator{[](boost::json::value const& value, std::string_view key) -> MaybeError {
if (!value.is_array())
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, std::string(key) + "NotArray"}};
for (auto const& book : value.as_array())
{
if (!book.is_object())
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, std::string(key) + "ItemNotObject"}};
if (book.as_object().contains("both") && !book.as_object().at("both").is_bool())
{
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, "bothNotBool"}};
}
if (book.as_object().contains("snapshot") && !book.as_object().at("snapshot").is_bool())
{
return Error{RPC::Status{RPC::RippledError::rpcINVALID_PARAMS, "snapshotNotBool"}};
}
if (book.as_object().contains("taker"))
{
if (auto const err = AccountValidator.verify(book.as_object(), "taker"); !err)
return err;
}
auto const parsedBook = RPC::parseBook(book.as_object());
if (auto const status = std::get_if<RPC::Status>(&parsedBook))
return Error(*status);
}
return MaybeError{};
}};
} // namespace RPCng::validation

18
src/rpc/common/Validators.h
View File

@@ -482,4 +482,22 @@ extern CustomValidator CurrencyValidator;
*/
extern CustomValidator IssuerValidator;
/**
* @brief Provide a validator for validating valid streams used in
* subscribe/unsubscribe
*/
extern CustomValidator SubscribeStreamValidator;
/**
* @brief Provide a validator for validating valid accounts used in
* subscribe/unsubscribe
*/
extern CustomValidator SubscribeAccountsValidator;
/**
* @brief Provide a validator for validating valid books used in
* subscribe/unsubscribe
*/
extern CustomValidator SubscribeBooksValidator;
} // namespace RPCng::validation

248
src/rpc/ngHandlers/Subscribe.h Normal file
View File

@@ -0,0 +1,248 @@
//------------------------------------------------------------------------------
/*
This file is part of clio: https://github.com/XRPLF/clio
Copyright (c) 2023, 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 <backend/BackendInterface.h>
#include <rpc/RPCHelpers.h>
#include <rpc/common/Types.h>
#include <rpc/common/Validators.h>
namespace RPCng {
template <typename SubscriptionManagerType>
class BaseSubscribeHandler
{
std::shared_ptr<BackendInterface> sharedPtrBackend_;
std::shared_ptr<SubscriptionManagerType> subscriptions_;
public:
struct Output
{
// response of stream "ledger"
// TODO: use better type than json, this type will be used in the stream as well
std::optional<boost::json::object> ledger;
// books returns nothing by default, if snapshot is true, it returns offers
// TODO: use better type than json
std::optional<boost::json::array> offers;
bool validated = true;
};
struct OrderBook
{
ripple::Book book;
std::optional<std::string> taker;
bool snapshot = false;
bool both = false;
};
struct Input
{
std::optional<std::vector<std::string>> accounts;
std::optional<std::vector<std::string>> streams;
std::optional<std::vector<std::string>> accountsProposed;
std::optional<std::vector<OrderBook>> books;
};
using Result = RPCng::HandlerReturnType<Output>;
BaseSubscribeHandler(
std::shared_ptr<BackendInterface> const& sharedPtrBackend,
std::shared_ptr<SubscriptionManagerType> const& subscriptions)
: sharedPtrBackend_(sharedPtrBackend), subscriptions_(subscriptions)
{
}
RpcSpecConstRef
spec() const
{
static auto const rpcSpec = RpcSpec{
{JS(streams), validation::SubscribeStreamValidator},
{JS(accounts), validation::SubscribeAccountsValidator},
{JS(accounts_proposed), validation::SubscribeAccountsValidator},
{JS(books), validation::SubscribeBooksValidator}};
return rpcSpec;
}
Result
process(Input input, Context const& ctx) const
{
Output output;
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)
{
auto const offers = subscribeToBooks(*(input.books), ctx.session, ctx.yield);
if (!offers.empty())
output.offers = offers;
};
return output;
}
private:
boost::json::object
subscribeToStreams(
boost::asio::yield_context& yield,
std::vector<std::string> const& streams,
std::shared_ptr<WsBase> const& session) const
{
boost::json::object response;
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
subscribeToAccounts(std::vector<std::string> const& accounts, std::shared_ptr<WsBase> const& session) const
{
for (auto const& account : accounts)
{
auto const accountID = RPC::accountFromStringStrict(account);
subscriptions_->subAccount(*accountID, session);
}
}
void
subscribeToAccountsProposed(std::vector<std::string> const& accounts, std::shared_ptr<WsBase> const& session) const
{
for (auto const& account : accounts)
{
auto const accountID = RPC::accountFromStringStrict(account);
subscriptions_->subProposedAccount(*accountID, session);
}
}
boost::json::array
subscribeToBooks(
std::vector<OrderBook> const& books,
std::shared_ptr<WsBase> const& session,
boost::asio::yield_context& yield) const
{
boost::json::array snapshots;
std::optional<Backend::LedgerRange> rng;
static auto constexpr fetchLimit = 200;
for (auto const& internalBook : books)
{
if (internalBook.snapshot)
{
if (!rng)
rng = sharedPtrBackend_->fetchLedgerRange();
auto const getOrderBook = [&](auto const& book) {
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 ? RPC::accountFromStringStrict(*(internalBook.taker)) : beast::zero;
auto const orderBook =
RPC::postProcessOrderBook(offers, book, *takerID, *sharedPtrBackend_, rng->maxSequence, yield);
std::copy(orderBook.begin(), orderBook.end(), std::back_inserter(snapshots));
};
getOrderBook(internalBook.book);
if (internalBook.both)
getOrderBook(ripple::reversed(internalBook.book));
}
subscriptions_->subBook(internalBook.book, session);
if (internalBook.both)
subscriptions_->subBook(ripple::reversed(internalBook.book), session);
}
return snapshots;
}
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));
}
friend Input
tag_invoke(boost::json::value_to_tag<Input>, boost::json::value const& jv)
{
auto const& jsonObject = jv.as_object();
Input input;
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(stream.as_string().c_str());
}
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(account.as_string().c_str());
}
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(account.as_string().c_str());
}
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 const& bookObject = book.as_object();
OrderBook internalBook;
if (auto const& taker = bookObject.find(JS(taker)); taker != bookObject.end())
internalBook.taker = taker->value().as_string().c_str();
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 = RPC::parseBook(book.as_object());
internalBook.book = std::get<ripple::Book>(parsedBookMaybe);
input.books->push_back(internalBook);
}
}
return input;
}
};
using SubscribeHandler = BaseSubscribeHandler<SubscriptionManager>;
} // namespace RPCng

29
src/subscriptions/SubscriptionManager.cpp
View File

@@ -66,7 +66,7 @@ getLedgerPubMessage(
boost::json::object
SubscriptionManager::subLedger(boost::asio::yield_context& yield, std::shared_ptr<WsBase> session)
{
subscribeHelper(session, ledgerSubscribers_, [this](session_ptr session) { unsubLedger(session); });
subscribeHelper(session, ledgerSubscribers_, [this](SessionPtrType session) { unsubLedger(session); });
auto ledgerRange = backend_->fetchLedgerRange();
assert(ledgerRange);
@@ -94,7 +94,7 @@ SubscriptionManager::unsubLedger(std::shared_ptr<WsBase> session)
void
SubscriptionManager::subTransactions(std::shared_ptr<WsBase> session)
{
subscribeHelper(session, txSubscribers_, [this](session_ptr session) { unsubTransactions(session); });
subscribeHelper(session, txSubscribers_, [this](SessionPtrType session) { unsubTransactions(session); });
}
void
@@ -104,15 +104,15 @@ SubscriptionManager::unsubTransactions(std::shared_ptr<WsBase> session)
}
void
SubscriptionManager::subAccount(ripple::AccountID const& account, std::shared_ptr<WsBase>& session)
SubscriptionManager::subAccount(ripple::AccountID const& account, std::shared_ptr<WsBase> const& session)
{
subscribeHelper(session, account, accountSubscribers_, [this, account](session_ptr session) {
subscribeHelper(session, account, accountSubscribers_, [this, account](SessionPtrType session) {
unsubAccount(account, session);
});
}
void
SubscriptionManager::unsubAccount(ripple::AccountID const& account, std::shared_ptr<WsBase>& session)
SubscriptionManager::unsubAccount(ripple::AccountID const& account, std::shared_ptr<WsBase> const& session)
{
accountSubscribers_.unsubscribe(session, account);
}
@@ -120,7 +120,8 @@ SubscriptionManager::unsubAccount(ripple::AccountID const& account, std::shared_
void
SubscriptionManager::subBook(ripple::Book const& book, std::shared_ptr<WsBase> session)
{
subscribeHelper(session, book, bookSubscribers_, [this, book](session_ptr session) { unsubBook(book, session); });
subscribeHelper(
session, book, bookSubscribers_, [this, book](SessionPtrType session) { unsubBook(book, session); });
}
void
@@ -132,7 +133,7 @@ SubscriptionManager::unsubBook(ripple::Book const& book, std::shared_ptr<WsBase>
void
SubscriptionManager::subBookChanges(std::shared_ptr<WsBase> session)
{
subscribeHelper(session, bookChangesSubscribers_, [this](session_ptr session) { unsubBookChanges(session); });
subscribeHelper(session, bookChangesSubscribers_, [this](SessionPtrType session) { unsubBookChanges(session); });
}
void
@@ -281,7 +282,7 @@ SubscriptionManager::forwardValidation(boost::json::object const& response)
void
SubscriptionManager::subProposedAccount(ripple::AccountID const& account, std::shared_ptr<WsBase> session)
{
subscribeHelper(session, account, accountProposedSubscribers_, [this, account](session_ptr session) {
subscribeHelper(session, account, accountProposedSubscribers_, [this, account](SessionPtrType session) {
unsubProposedAccount(account, session);
});
}
@@ -289,7 +290,7 @@ SubscriptionManager::subProposedAccount(ripple::AccountID const& account, std::s
void
SubscriptionManager::subManifest(std::shared_ptr<WsBase> session)
{
subscribeHelper(session, manifestSubscribers_, [this](session_ptr session) { unsubManifest(session); });
subscribeHelper(session, manifestSubscribers_, [this](SessionPtrType session) { unsubManifest(session); });
}
void
@@ -301,7 +302,7 @@ SubscriptionManager::unsubManifest(std::shared_ptr<WsBase> session)
void
SubscriptionManager::subValidation(std::shared_ptr<WsBase> session)
{
subscribeHelper(session, validationsSubscribers_, [this](session_ptr session) { unsubValidation(session); });
subscribeHelper(session, validationsSubscribers_, [this](SessionPtrType session) { unsubValidation(session); });
}
void
@@ -320,7 +321,7 @@ void
SubscriptionManager::subProposedTransactions(std::shared_ptr<WsBase> session)
{
subscribeHelper(
session, txProposedSubscribers_, [this](session_ptr session) { unsubProposedTransactions(session); });
session, txProposedSubscribers_, [this](SessionPtrType session) { unsubProposedTransactions(session); });
}
void
@@ -328,17 +329,19 @@ SubscriptionManager::unsubProposedTransactions(std::shared_ptr<WsBase> session)
{
txProposedSubscribers_.unsubscribe(session);
}
void
SubscriptionManager::subscribeHelper(std::shared_ptr<WsBase>& session, Subscription& subs, CleanupFunction&& func)
SubscriptionManager::subscribeHelper(std::shared_ptr<WsBase> const& session, Subscription& subs, CleanupFunction&& func)
{
subs.subscribe(session);
std::scoped_lock lk(cleanupMtx_);
cleanupFuncs_[session].push_back(std::move(func));
}
template <typename Key>
void
SubscriptionManager::subscribeHelper(
std::shared_ptr<WsBase>& session,
std::shared_ptr<WsBase> const& session,
Key const& k,
SubscriptionMap<Key>& subs,
CleanupFunction&& func)

50
src/subscriptions/SubscriptionManager.h
View File

@@ -189,7 +189,7 @@ SubscriptionMap<Key>::publish(std::shared_ptr<Message> const& message, Key const
class SubscriptionManager
{
using session_ptr = std::shared_ptr<WsBase>;
using SessionPtrType = std::shared_ptr<WsBase>;
clio::Logger log_{"Subscriptions"};
std::vector<std::thread> workers_;
@@ -251,7 +251,7 @@ public:
}
boost::json::object
subLedger(boost::asio::yield_context& yield, session_ptr session);
subLedger(boost::asio::yield_context& yield, SessionPtrType session);
void
pubLedger(
@@ -264,28 +264,28 @@ public:
pubBookChanges(ripple::LedgerInfo const& lgrInfo, std::vector<Backend::TransactionAndMetadata> const& transactions);
void
unsubLedger(session_ptr session);
unsubLedger(SessionPtrType session);
void
subTransactions(session_ptr session);
subTransactions(SessionPtrType session);
void
unsubTransactions(session_ptr session);
unsubTransactions(SessionPtrType session);
void
pubTransaction(Backend::TransactionAndMetadata const& blobs, ripple::LedgerInfo const& lgrInfo);
void
subAccount(ripple::AccountID const& account, session_ptr& session);
subAccount(ripple::AccountID const& account, SessionPtrType const& session);
void
unsubAccount(ripple::AccountID const& account, session_ptr& session);
unsubAccount(ripple::AccountID const& account, SessionPtrType const& session);
void
subBook(ripple::Book const& book, session_ptr session);
subBook(ripple::Book const& book, SessionPtrType session);
void
unsubBook(ripple::Book const& book, session_ptr session);
unsubBook(ripple::Book const& book, SessionPtrType session);
void
subBookChanges(std::shared_ptr<WsBase> session);
@@ -294,16 +294,16 @@ public:
unsubBookChanges(std::shared_ptr<WsBase> session);
void
subManifest(session_ptr session);
subManifest(SessionPtrType session);
void
unsubManifest(session_ptr session);
unsubManifest(SessionPtrType session);
void
subValidation(session_ptr session);
subValidation(SessionPtrType session);
void
unsubValidation(session_ptr session);
unsubValidation(SessionPtrType session);
void
forwardProposedTransaction(boost::json::object const& response);
@@ -315,19 +315,19 @@ public:
forwardValidation(boost::json::object const& response);
void
subProposedAccount(ripple::AccountID const& account, session_ptr session);
subProposedAccount(ripple::AccountID const& account, SessionPtrType session);
void
unsubProposedAccount(ripple::AccountID const& account, session_ptr session);
unsubProposedAccount(ripple::AccountID const& account, SessionPtrType session);
void
subProposedTransactions(session_ptr session);
subProposedTransactions(SessionPtrType session);
void
unsubProposedTransactions(session_ptr session);
unsubProposedTransactions(SessionPtrType session);
void
cleanup(session_ptr session);
cleanup(SessionPtrType session);
boost::json::object
report() const
@@ -349,16 +349,20 @@ public:
private:
void
sendAll(std::string const& pubMsg, std::unordered_set<session_ptr>& subs);
sendAll(std::string const& pubMsg, std::unordered_set<SessionPtrType>& subs);
using CleanupFunction = std::function<void(session_ptr)>;
using CleanupFunction = std::function<void(SessionPtrType const)>;
void
subscribeHelper(std::shared_ptr<WsBase>& session, Subscription& subs, CleanupFunction&& func);
subscribeHelper(std::shared_ptr<WsBase> const& session, Subscription& subs, CleanupFunction&& func);
template <typename Key>
void
subscribeHelper(std::shared_ptr<WsBase>& session, Key const& k, SubscriptionMap<Key>& subs, CleanupFunction&& func);
subscribeHelper(
std::shared_ptr<WsBase> const& session,
Key const& k,
SubscriptionMap<Key>& subs,
CleanupFunction&& func);
/**
* This is how we chose to cleanup subscriptions that have been closed.
@@ -367,5 +371,5 @@ private:
* closed.
*/
std::mutex cleanupMtx_;
std::unordered_map<session_ptr, std::vector<CleanupFunction>> cleanupFuncs_ = {};
std::unordered_map<SessionPtrType, std::vector<CleanupFunction>> cleanupFuncs_ = {};
};