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

Fix bug: api version change does not affect the subscribed session (#1404)

Browse Source 
Fix #1133
This commit is contained in:
cyan317
2024-05-13 09:12:35 +01:00
committed by GitHub
parent a8c90a31d9
commit 3a6390caf5
10 changed files with 105 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/feed/SubscriptionManager.cpp
View File

@@ -145,9 +145,9 @@ SubscriptionManager::forwardValidation(boost::json::object const& validationJson
} }
void void
SubscriptionManager::subTransactions(SubscriberSharedPtr const& subscriber, std::uint32_t const apiVersion) SubscriptionManager::subTransactions(SubscriberSharedPtr const& subscriber)
{ {
transactionFeed_.sub(subscriber, apiVersion); transactionFeed_.sub(subscriber);
} }
void void
@@ -157,13 +157,9 @@ SubscriptionManager::unsubTransactions(SubscriberSharedPtr const& subscriber)
} }
void void
SubscriptionManager::subAccount( SubscriptionManager::subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
ripple::AccountID const& account,
SubscriberSharedPtr const& subscriber,
std::uint32_t const apiVersion
)
{ {
transactionFeed_.sub(account, subscriber, apiVersion); transactionFeed_.sub(account, subscriber);
} }
void void
@@ -173,13 +169,9 @@ SubscriptionManager::unsubAccount(ripple::AccountID const& account, SubscriberSh
} }
void void
SubscriptionManager::subBook( SubscriptionManager::subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber)
ripple::Book const& book,
SubscriberSharedPtr const& subscriber,
std::uint32_t const apiVersion
)
{ {
transactionFeed_.sub(book, subscriber, apiVersion); transactionFeed_.sub(book, subscriber);
} }
void void

9
src/feed/SubscriptionManager.hpp
View File

@@ -224,10 +224,9 @@ public:
/** /**
* @brief Subscribe to the transactions feed. * @brief Subscribe to the transactions feed.
* @param subscriber * @param subscriber
* @param apiVersion The api version of feed to subscribe.
*/ */
void void
subTransactions(SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); subTransactions(SubscriberSharedPtr const& subscriber);
/** /**
* @brief Unsubscribe to the transactions feed. * @brief Unsubscribe to the transactions feed.
@@ -240,10 +239,9 @@ public:
* @brief Subscribe to the transactions feed, only receive the feed when particular account is affected. * @brief Subscribe to the transactions feed, only receive the feed when particular account is affected.
* @param account The account to watch. * @param account The account to watch.
* @param subscriber * @param subscriber
* @param apiVersion The api version of feed to subscribe.
*/ */
void void
subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); subAccount(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
/** /**
* @brief Unsubscribe to the transactions feed for particular account. * @brief Unsubscribe to the transactions feed for particular account.
@@ -257,10 +255,9 @@ public:
* @brief Subscribe to the transactions feed, only receive feed when particular order book is affected. * @brief Subscribe to the transactions feed, only receive feed when particular order book is affected.
* @param book The book to watch. * @param book The book to watch.
* @param subscriber * @param subscriber
* @param apiVersion The api version of feed to subscribe.
*/ */
void void
subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); subBook(ripple::Book const& book, SubscriberSharedPtr const& subscriber);
/** /**
* @brief Unsubscribe to the transactions feed for particular order book. * @brief Unsubscribe to the transactions feed for particular order book.

13
src/feed/impl/TransactionFeed.cpp
View File

@@ -70,29 +70,23 @@ TransactionFeed::TransactionSlot::operator()(AllVersionTransactionsType const& a
} }
void void
TransactionFeed::sub(SubscriberSharedPtr const& subscriber, std::uint32_t const apiVersion) TransactionFeed::sub(SubscriberSharedPtr const& subscriber)
{ {
auto const added = signal_.connectTrackableSlot(subscriber, TransactionSlot(*this, subscriber)); auto const added = signal_.connectTrackableSlot(subscriber, TransactionSlot(*this, subscriber));
if (added) { if (added) {
LOG(logger_.info()) << subscriber->tag() << "Subscribed transactions"; LOG(logger_.info()) << subscriber->tag() << "Subscribed transactions";
++subAllCount_.get(); ++subAllCount_.get();
subscriber->apiSubVersion = apiVersion;
subscriber->onDisconnect.connect([this](SubscriberPtr connection) { unsubInternal(connection); }); subscriber->onDisconnect.connect([this](SubscriberPtr connection) { unsubInternal(connection); });
} }
} }
void void
TransactionFeed::sub( TransactionFeed::sub(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber)
ripple::AccountID const& account,
SubscriberSharedPtr const& subscriber,
std::uint32_t const apiVersion
)
{ {
auto const added = accountSignal_.connectTrackableSlot(subscriber, account, TransactionSlot(*this, subscriber)); auto const added = accountSignal_.connectTrackableSlot(subscriber, account, TransactionSlot(*this, subscriber));
if (added) { if (added) {
LOG(logger_.info()) << subscriber->tag() << "Subscribed account " << account; LOG(logger_.info()) << subscriber->tag() << "Subscribed account " << account;
++subAccountCount_.get(); ++subAccountCount_.get();
subscriber->apiSubVersion = apiVersion;
subscriber->onDisconnect.connect([this, account](SubscriberPtr connection) { subscriber->onDisconnect.connect([this, account](SubscriberPtr connection) {
unsubInternal(account, connection); unsubInternal(account, connection);
}); });
@@ -100,13 +94,12 @@ TransactionFeed::sub(
} }
void void
TransactionFeed::sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber, std::uint32_t const apiVersion) TransactionFeed::sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber)
{ {
auto const added = bookSignal_.connectTrackableSlot(subscriber, book, TransactionSlot(*this, subscriber)); auto const added = bookSignal_.connectTrackableSlot(subscriber, book, TransactionSlot(*this, subscriber));
if (added) { if (added) {
LOG(logger_.info()) << subscriber->tag() << "Subscribed book " << book; LOG(logger_.info()) << subscriber->tag() << "Subscribed book " << book;
++subBookCount_.get(); ++subBookCount_.get();
subscriber->apiSubVersion = apiVersion;
subscriber->onDisconnect.connect([this, book](SubscriberPtr connection) { unsubInternal(book, connection); }); subscriber->onDisconnect.connect([this, book](SubscriberPtr connection) { unsubInternal(book, connection); });
} }
} }

9
src/feed/impl/TransactionFeed.hpp
View File

@@ -91,28 +91,25 @@ public:
/** /**
* @brief Subscribe to the transaction feed. * @brief Subscribe to the transaction feed.
* @param subscriber * @param subscriber
* @param apiVersion The api version of feed.
*/ */
void void
sub(SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); sub(SubscriberSharedPtr const& subscriber);
/** /**
* @brief Subscribe to the transaction feed, only receive the feed when particular account is affected. * @brief Subscribe to the transaction feed, only receive the feed when particular account is affected.
* @param subscriber * @param subscriber
* @param account The account to watch. * @param account The account to watch.
* @param apiVersion The api version of feed.
*/ */
void void
sub(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); sub(ripple::AccountID const& account, SubscriberSharedPtr const& subscriber);
/** /**
* @brief Subscribe to the transaction feed, only receive the feed when particular order book is affected. * @brief Subscribe to the transaction feed, only receive the feed when particular order book is affected.
* @param subscriber * @param subscriber
* @param book The order book to watch. * @param book The order book to watch.
* @param apiVersion The api version of feed.
*/ */
void void
sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber, std::uint32_t apiVersion); sub(ripple::Book const& book, SubscriberSharedPtr const& subscriber);
/** /**
* @brief Unsubscribe to the transaction feed. * @brief Unsubscribe to the transaction feed.

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

@@ -184,20 +184,23 @@ public:
{ {
auto output = Output{}; 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) { if (input.streams) {
auto const ledger = subscribeToStreams(ctx.yield, *(input.streams), ctx.session, ctx.apiVersion); auto const ledger = subscribeToStreams(ctx.yield, *(input.streams), ctx.session);
if (!ledger.empty()) if (!ledger.empty())
output.ledger = ledger; output.ledger = ledger;
} }
if (input.accounts) if (input.accounts)
subscribeToAccounts(*(input.accounts), ctx.session, ctx.apiVersion); subscribeToAccounts(*(input.accounts), ctx.session);
if (input.accountsProposed) if (input.accountsProposed)
subscribeToAccountsProposed(*(input.accountsProposed), ctx.session); subscribeToAccountsProposed(*(input.accountsProposed), ctx.session);
if (input.books) if (input.books)
subscribeToBooks(*(input.books), ctx.session, ctx.yield, ctx.apiVersion, output); subscribeToBooks(*(input.books), ctx.session, ctx.yield, output);
return output; return output;
} }
@@ -207,8 +210,7 @@ private:
subscribeToStreams( subscribeToStreams(
boost::asio::yield_context yield, boost::asio::yield_context yield,
std::vector<std::string> const& streams, std::vector<std::string> const& streams,
std::shared_ptr<web::ConnectionBase> const& session, std::shared_ptr<web::ConnectionBase> const& session
std::uint32_t apiVersion
) const ) const
{ {
auto response = boost::json::object{}; auto response = boost::json::object{};
@@ -217,7 +219,7 @@ private:
if (stream == "ledger") { if (stream == "ledger") {
response = subscriptions_->subLedger(yield, session); response = subscriptions_->subLedger(yield, session);
} else if (stream == "transactions") { } else if (stream == "transactions") {
subscriptions_->subTransactions(session, apiVersion); subscriptions_->subTransactions(session);
} else if (stream == "transactions_proposed") { } else if (stream == "transactions_proposed") {
subscriptions_->subProposedTransactions(session); subscriptions_->subProposedTransactions(session);
} else if (stream == "validations") { } else if (stream == "validations") {
@@ -233,15 +235,12 @@ private:
} }
void void
subscribeToAccounts( subscribeToAccounts(std::vector<std::string> const& accounts, std::shared_ptr<web::ConnectionBase> const& session)
std::vector<std::string> const& accounts, const
std::shared_ptr<web::ConnectionBase> const& session,
std::uint32_t apiVersion
) const
{ {
for (auto const& account : accounts) { for (auto const& account : accounts) {
auto const accountID = accountFromStringStrict(account); auto const accountID = accountFromStringStrict(account);
subscriptions_->subAccount(*accountID, session, apiVersion); subscriptions_->subAccount(*accountID, session);
} }
} }
@@ -262,7 +261,6 @@ private:
std::vector<OrderBook> const& books, std::vector<OrderBook> const& books,
std::shared_ptr<web::ConnectionBase> const& session, std::shared_ptr<web::ConnectionBase> const& session,
boost::asio::yield_context yield, boost::asio::yield_context yield,
uint32_t apiVersion,
Output& output Output& output
) const ) const
{ {
@@ -304,10 +302,10 @@ private:
} }
} }
subscriptions_->subBook(internalBook.book, session, apiVersion); subscriptions_->subBook(internalBook.book, session);
if (internalBook.both) if (internalBook.both)
subscriptions_->subBook(ripple::reversed(internalBook.book), session, apiVersion); subscriptions_->subBook(ripple::reversed(internalBook.book), session);
} }
} }

5
src/web/interface/ConnectionBase.hpp
View File

@@ -50,6 +50,11 @@ public:
std::string const clientIp; std::string const clientIp;
bool upgraded = false; bool upgraded = false;
boost::signals2::signal<void(ConnectionBase*)> onDisconnect; boost::signals2::signal<void(ConnectionBase*)> onDisconnect;
/**
* @brief The API version of the web stream client.
* This is used to track the api version of this connection, which mainly is used by subscription. It is different
* from the api version in Context, which is only used for the current request.
*/
std::uint32_t apiSubVersion = 0; std::uint32_t apiSubVersion = 0;
/** /**

1
tests/common/feed/FeedTestUtil.hpp
View File

@@ -47,6 +47,7 @@ protected:
MockBackendTest::SetUp(); MockBackendTest::SetUp();
testFeedPtr = std::make_shared<TestedFeed>(ctx); testFeedPtr = std::make_shared<TestedFeed>(ctx);
sessionPtr = std::make_shared<MockSession>(); sessionPtr = std::make_shared<MockSession>();
sessionPtr->apiSubVersion = 1;
mockSessionPtr = dynamic_cast<MockSession*>(sessionPtr.get()); mockSessionPtr = dynamic_cast<MockSession*>(sessionPtr.get());
} }

20
tests/unit/feed/SubscriptionManagerTests.cpp
View File

@@ -68,6 +68,7 @@ protected:
SyncAsioContextTest::SetUp(); SyncAsioContextTest::SetUp();
SubscriptionManagerPtr = std::make_shared<SubscriptionManager>(ctx, backend); SubscriptionManagerPtr = std::make_shared<SubscriptionManager>(ctx, backend);
session = std::make_shared<MockSession>(); session = std::make_shared<MockSession>();
session->apiSubVersion = 1;
sessionPtr = dynamic_cast<MockSession*>(session.get()); sessionPtr = dynamic_cast<MockSession*>(session.get());
} }
@@ -159,19 +160,20 @@ TEST_F(SubscriptionManagerTest, ReportCurrentSubscriber)
SubscriptionManagerPtr->subManifest(session2); SubscriptionManagerPtr->subManifest(session2);
SubscriptionManagerPtr->subProposedTransactions(session1); SubscriptionManagerPtr->subProposedTransactions(session1);
SubscriptionManagerPtr->subProposedTransactions(session2); SubscriptionManagerPtr->subProposedTransactions(session2);
SubscriptionManagerPtr->subTransactions(session1, 1); SubscriptionManagerPtr->subTransactions(session1);
SubscriptionManagerPtr->subTransactions(session2, 2); session2->apiSubVersion = 2;
SubscriptionManagerPtr->subTransactions(session2);
SubscriptionManagerPtr->subValidation(session1); SubscriptionManagerPtr->subValidation(session1);
SubscriptionManagerPtr->subValidation(session2); SubscriptionManagerPtr->subValidation(session2);
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
SubscriptionManagerPtr->subAccount(account, session1, 1); SubscriptionManagerPtr->subAccount(account, session1);
SubscriptionManagerPtr->subAccount(account, session2, 2); SubscriptionManagerPtr->subAccount(account, session2);
SubscriptionManagerPtr->subProposedAccount(account, session1); SubscriptionManagerPtr->subProposedAccount(account, session1);
SubscriptionManagerPtr->subProposedAccount(account, session2); SubscriptionManagerPtr->subProposedAccount(account, session2);
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
SubscriptionManagerPtr->subBook(book, session1, 1); SubscriptionManagerPtr->subBook(book, session1);
SubscriptionManagerPtr->subBook(book, session2, 2); SubscriptionManagerPtr->subBook(book, session2);
EXPECT_EQ(SubscriptionManagerPtr->report(), json::parse(ReportReturn)); EXPECT_EQ(SubscriptionManagerPtr->report(), json::parse(ReportReturn));
// count down when unsub manually // count down when unsub manually
@@ -338,9 +340,9 @@ TEST_F(SubscriptionManagerTest, TransactionTest)
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
auto const account = GetAccountIDWithString(ISSUER); auto const account = GetAccountIDWithString(ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
SubscriptionManagerPtr->subBook(book, session, 1); SubscriptionManagerPtr->subBook(book, session);
SubscriptionManagerPtr->subTransactions(session, 1); SubscriptionManagerPtr->subTransactions(session);
SubscriptionManagerPtr->subAccount(account, session, 1); SubscriptionManagerPtr->subAccount(account, session);
EXPECT_EQ(SubscriptionManagerPtr->report()["account"], 1); EXPECT_EQ(SubscriptionManagerPtr->report()["account"], 1);
EXPECT_EQ(SubscriptionManagerPtr->report()["transactions"], 1); EXPECT_EQ(SubscriptionManagerPtr->report()["transactions"], 1);
EXPECT_EQ(SubscriptionManagerPtr->report()["books"], 1); EXPECT_EQ(SubscriptionManagerPtr->report()["books"], 1);

77
tests/unit/feed/TransactionFeedTests.cpp
View File

@@ -165,7 +165,7 @@ using FeedTransactionTest = FeedBaseTest<TransactionFeed>;
TEST_F(FeedTransactionTest, SubTransactionV1) TEST_F(FeedTransactionTest, SubTransactionV1)
{ {
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 1); EXPECT_EQ(testFeedPtr->transactionSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -189,7 +189,8 @@ TEST_F(FeedTransactionTest, SubTransactionV1)
TEST_F(FeedTransactionTest, SubTransactionV2) TEST_F(FeedTransactionTest, SubTransactionV2)
{ {
testFeedPtr->sub(sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(sessionPtr);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 1); EXPECT_EQ(testFeedPtr->transactionSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -214,7 +215,7 @@ TEST_F(FeedTransactionTest, SubTransactionV2)
TEST_F(FeedTransactionTest, SubAccountV1) TEST_F(FeedTransactionTest, SubAccountV1)
{ {
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
EXPECT_EQ(testFeedPtr->accountSubCount(), 1); EXPECT_EQ(testFeedPtr->accountSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -239,7 +240,8 @@ TEST_F(FeedTransactionTest, SubAccountV1)
TEST_F(FeedTransactionTest, SubAccountV2) TEST_F(FeedTransactionTest, SubAccountV2)
{ {
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(account, sessionPtr);
EXPECT_EQ(testFeedPtr->accountSubCount(), 1); EXPECT_EQ(testFeedPtr->accountSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -265,8 +267,9 @@ TEST_F(FeedTransactionTest, SubAccountV2)
TEST_F(FeedTransactionTest, SubBothTransactionAndAccount) TEST_F(FeedTransactionTest, SubBothTransactionAndAccount)
{ {
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(sessionPtr, 2); testFeedPtr->sub(account, sessionPtr);
testFeedPtr->sub(sessionPtr);
EXPECT_EQ(testFeedPtr->accountSubCount(), 1); EXPECT_EQ(testFeedPtr->accountSubCount(), 1);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 1); EXPECT_EQ(testFeedPtr->transactionSubCount(), 1);
@@ -297,7 +300,7 @@ TEST_F(FeedTransactionTest, SubBookV1)
{ {
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
testFeedPtr->sub(book, sessionPtr, 1); testFeedPtr->sub(book, sessionPtr);
EXPECT_EQ(testFeedPtr->bookSubCount(), 1); EXPECT_EQ(testFeedPtr->bookSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -498,7 +501,8 @@ TEST_F(FeedTransactionTest, SubBookV2)
{ {
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
testFeedPtr->sub(book, sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(book, sessionPtr);
EXPECT_EQ(testFeedPtr->bookSubCount(), 1); EXPECT_EQ(testFeedPtr->bookSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -584,11 +588,12 @@ TEST_F(FeedTransactionTest, SubBookV2)
TEST_F(FeedTransactionTest, TransactionContainsBothAccountsSubed) TEST_F(FeedTransactionTest, TransactionContainsBothAccountsSubed)
{ {
sessionPtr->apiSubVersion = 2;
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 2); testFeedPtr->sub(account, sessionPtr);
auto const account2 = GetAccountIDWithString(ACCOUNT2); auto const account2 = GetAccountIDWithString(ACCOUNT2);
testFeedPtr->sub(account2, sessionPtr, 2); testFeedPtr->sub(account2, sessionPtr);
EXPECT_EQ(testFeedPtr->accountSubCount(), 2); EXPECT_EQ(testFeedPtr->accountSubCount(), 2);
@@ -620,10 +625,11 @@ TEST_F(FeedTransactionTest, TransactionContainsBothAccountsSubed)
TEST_F(FeedTransactionTest, SubAccountRepeatWithDifferentVersion) TEST_F(FeedTransactionTest, SubAccountRepeatWithDifferentVersion)
{ {
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
auto const account2 = GetAccountIDWithString(ACCOUNT2); auto const account2 = GetAccountIDWithString(ACCOUNT2);
testFeedPtr->sub(account2, sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(account2, sessionPtr);
EXPECT_EQ(testFeedPtr->accountSubCount(), 2); EXPECT_EQ(testFeedPtr->accountSubCount(), 2);
@@ -656,9 +662,11 @@ TEST_F(FeedTransactionTest, SubAccountRepeatWithDifferentVersion)
TEST_F(FeedTransactionTest, SubTransactionRepeatWithDifferentVersion) TEST_F(FeedTransactionTest, SubTransactionRepeatWithDifferentVersion)
{ {
// sub version 1 first // sub version 1 first
testFeedPtr->sub(sessionPtr, 1); sessionPtr->apiSubVersion = 1;
testFeedPtr->sub(sessionPtr);
// sub version 2 later // sub version 2 later
testFeedPtr->sub(sessionPtr, 2); sessionPtr->apiSubVersion = 2;
testFeedPtr->sub(sessionPtr);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 1); EXPECT_EQ(testFeedPtr->transactionSubCount(), 1);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
@@ -669,7 +677,7 @@ TEST_F(FeedTransactionTest, SubTransactionRepeatWithDifferentVersion)
trans1.metadata = CreatePaymentTransactionMetaObject(ACCOUNT1, ACCOUNT2, 110, 30, 22).getSerializer().peekData(); trans1.metadata = CreatePaymentTransactionMetaObject(ACCOUNT1, ACCOUNT2, 110, 30, 22).getSerializer().peekData();
testFeedPtr->pub(trans1, ledgerinfo, backend); testFeedPtr->pub(trans1, ledgerinfo, backend);
EXPECT_CALL(*mockSessionPtr, send(SharedStringJsonEq(TRAN_V1))).Times(1); EXPECT_CALL(*mockSessionPtr, send(SharedStringJsonEq(TRAN_V2))).Times(1);
ctx.run(); ctx.run();
testFeedPtr->unsub(sessionPtr); testFeedPtr->unsub(sessionPtr);
@@ -683,13 +691,14 @@ TEST_F(FeedTransactionTest, SubTransactionRepeatWithDifferentVersion)
TEST_F(FeedTransactionTest, SubRepeat) TEST_F(FeedTransactionTest, SubRepeat)
{ {
auto const session2 = std::make_shared<MockSession>(); auto const session2 = std::make_shared<MockSession>();
session2->apiSubVersion = 1;
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
testFeedPtr->sub(session2, 1); testFeedPtr->sub(session2);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 2); EXPECT_EQ(testFeedPtr->transactionSubCount(), 2);
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
testFeedPtr->sub(session2, 1); testFeedPtr->sub(session2);
EXPECT_EQ(testFeedPtr->transactionSubCount(), 2); EXPECT_EQ(testFeedPtr->transactionSubCount(), 2);
testFeedPtr->unsub(sessionPtr); testFeedPtr->unsub(sessionPtr);
@@ -701,12 +710,12 @@ TEST_F(FeedTransactionTest, SubRepeat)
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
auto const account2 = GetAccountIDWithString(ACCOUNT2); auto const account2 = GetAccountIDWithString(ACCOUNT2);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
testFeedPtr->sub(account2, session2, 1); testFeedPtr->sub(account2, session2);
EXPECT_EQ(testFeedPtr->accountSubCount(), 2); EXPECT_EQ(testFeedPtr->accountSubCount(), 2);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
testFeedPtr->sub(account2, session2, 1); testFeedPtr->sub(account2, session2);
EXPECT_EQ(testFeedPtr->accountSubCount(), 2); EXPECT_EQ(testFeedPtr->accountSubCount(), 2);
testFeedPtr->unsub(account, sessionPtr); testFeedPtr->unsub(account, sessionPtr);
@@ -718,9 +727,9 @@ TEST_F(FeedTransactionTest, SubRepeat)
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
testFeedPtr->sub(book, sessionPtr, 1); testFeedPtr->sub(book, sessionPtr);
EXPECT_EQ(testFeedPtr->bookSubCount(), 1); EXPECT_EQ(testFeedPtr->bookSubCount(), 1);
testFeedPtr->sub(book, session2, 1); testFeedPtr->sub(book, session2);
EXPECT_EQ(testFeedPtr->bookSubCount(), 2); EXPECT_EQ(testFeedPtr->bookSubCount(), 2);
testFeedPtr->unsub(book, sessionPtr); testFeedPtr->unsub(book, sessionPtr);
@@ -733,7 +742,7 @@ TEST_F(FeedTransactionTest, SubRepeat)
TEST_F(FeedTransactionTest, PubTransactionWithOwnerFund) TEST_F(FeedTransactionTest, PubTransactionWithOwnerFund)
{ {
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
auto trans1 = TransactionAndMetadata(); auto trans1 = TransactionAndMetadata();
@@ -846,7 +855,7 @@ constexpr static auto TRAN_FROZEN =
TEST_F(FeedTransactionTest, PubTransactionOfferCreationFrozenLine) TEST_F(FeedTransactionTest, PubTransactionOfferCreationFrozenLine)
{ {
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
auto trans1 = TransactionAndMetadata(); auto trans1 = TransactionAndMetadata();
@@ -884,7 +893,7 @@ TEST_F(FeedTransactionTest, PubTransactionOfferCreationFrozenLine)
TEST_F(FeedTransactionTest, SubTransactionOfferCreationGlobalFrozen) TEST_F(FeedTransactionTest, SubTransactionOfferCreationGlobalFrozen)
{ {
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33); auto const ledgerinfo = CreateLedgerInfo(LEDGERHASH, 33);
auto trans1 = TransactionAndMetadata(); auto trans1 = TransactionAndMetadata();
@@ -955,16 +964,16 @@ TEST_F(TransactionFeedMockPrometheusTest, subUnsub)
EXPECT_CALL(counterBook, add(1)); EXPECT_CALL(counterBook, add(1));
EXPECT_CALL(counterBook, add(-1)); EXPECT_CALL(counterBook, add(-1));
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
testFeedPtr->unsub(sessionPtr); testFeedPtr->unsub(sessionPtr);
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
testFeedPtr->unsub(account, sessionPtr); testFeedPtr->unsub(account, sessionPtr);
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
testFeedPtr->sub(book, sessionPtr, 1); testFeedPtr->sub(book, sessionPtr);
testFeedPtr->unsub(book, sessionPtr); testFeedPtr->unsub(book, sessionPtr);
} }
@@ -981,14 +990,14 @@ TEST_F(TransactionFeedMockPrometheusTest, AutoDisconnect)
EXPECT_CALL(counterBook, add(1)); EXPECT_CALL(counterBook, add(1));
EXPECT_CALL(counterBook, add(-1)); EXPECT_CALL(counterBook, add(-1));
testFeedPtr->sub(sessionPtr, 1); testFeedPtr->sub(sessionPtr);
auto const account = GetAccountIDWithString(ACCOUNT1); auto const account = GetAccountIDWithString(ACCOUNT1);
testFeedPtr->sub(account, sessionPtr, 1); testFeedPtr->sub(account, sessionPtr);
auto const issue1 = GetIssue(CURRENCY, ISSUER); auto const issue1 = GetIssue(CURRENCY, ISSUER);
ripple::Book const book{ripple::xrpIssue(), issue1}; ripple::Book const book{ripple::xrpIssue(), issue1};
testFeedPtr->sub(book, sessionPtr, 1); testFeedPtr->sub(book, sessionPtr);
sessionPtr.reset(); sessionPtr.reset();
} }

25
tests/unit/rpc/handlers/SubscribeTests.cpp
View File

@@ -618,7 +618,6 @@ TEST_F(RPCSubscribeHandlerTest, StreamsWithoutLedger)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_TRUE(output.result->as_object().empty()); EXPECT_TRUE(output.result->as_object().empty());
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
EXPECT_EQ(report.at("transactions_proposed").as_uint64(), 1); EXPECT_EQ(report.at("transactions_proposed").as_uint64(), 1);
EXPECT_EQ(report.at("transactions").as_uint64(), 1); EXPECT_EQ(report.at("transactions").as_uint64(), 1);
@@ -663,7 +662,6 @@ TEST_F(RPCSubscribeHandlerTest, StreamsLedger)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_EQ(output.result->as_object(), json::parse(expectedOutput)); EXPECT_EQ(output.result->as_object(), json::parse(expectedOutput));
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
EXPECT_EQ(report.at("ledger").as_uint64(), 1); EXPECT_EQ(report.at("ledger").as_uint64(), 1);
}); });
@@ -684,7 +682,6 @@ TEST_F(RPCSubscribeHandlerTest, Accounts)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_TRUE(output.result->as_object().empty()); EXPECT_TRUE(output.result->as_object().empty());
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
// filter the duplicates // filter the duplicates
EXPECT_EQ(report.at("account").as_uint64(), 2); EXPECT_EQ(report.at("account").as_uint64(), 2);
@@ -706,7 +703,6 @@ TEST_F(RPCSubscribeHandlerTest, AccountsProposed)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_TRUE(output.result->as_object().empty()); EXPECT_TRUE(output.result->as_object().empty());
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
// filter the duplicates // filter the duplicates
EXPECT_EQ(report.at("accounts_proposed").as_uint64(), 2); EXPECT_EQ(report.at("accounts_proposed").as_uint64(), 2);
@@ -739,7 +735,6 @@ TEST_F(RPCSubscribeHandlerTest, JustBooks)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_TRUE(output.result->as_object().empty()); EXPECT_TRUE(output.result->as_object().empty());
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
EXPECT_EQ(report.at("books").as_uint64(), 1); EXPECT_EQ(report.at("books").as_uint64(), 1);
}); });
@@ -772,7 +767,6 @@ TEST_F(RPCSubscribeHandlerTest, BooksBothSet)
auto const output = handler.process(input, Context{yield, session_}); auto const output = handler.process(input, Context{yield, session_});
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_TRUE(output.result->as_object().empty()); EXPECT_TRUE(output.result->as_object().empty());
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
// original book + reverse book // original book + reverse book
EXPECT_EQ(report.at("books").as_uint64(), 2); EXPECT_EQ(report.at("books").as_uint64(), 2);
@@ -942,7 +936,6 @@ TEST_F(RPCSubscribeHandlerTest, BooksBothSnapshotSet)
EXPECT_EQ(output.result->as_object().at("asks").as_array().size(), 10); EXPECT_EQ(output.result->as_object().at("asks").as_array().size(), 10);
EXPECT_EQ(output.result->as_object().at("bids").as_array()[0].as_object(), json::parse(expectedOffer)); EXPECT_EQ(output.result->as_object().at("bids").as_array()[0].as_object(), json::parse(expectedOffer));
EXPECT_EQ(output.result->as_object().at("asks").as_array()[0].as_object(), json::parse(expectedReversedOffer)); EXPECT_EQ(output.result->as_object().at("asks").as_array()[0].as_object(), json::parse(expectedReversedOffer));
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
// original book + reverse book // original book + reverse book
EXPECT_EQ(report.at("books").as_uint64(), 2); EXPECT_EQ(report.at("books").as_uint64(), 2);
@@ -1083,13 +1076,29 @@ TEST_F(RPCSubscribeHandlerTest, BooksBothUnsetSnapshotSet)
ASSERT_TRUE(output); ASSERT_TRUE(output);
EXPECT_EQ(output.result->as_object().at("offers").as_array().size(), 10); EXPECT_EQ(output.result->as_object().at("offers").as_array().size(), 10);
EXPECT_EQ(output.result->as_object().at("offers").as_array()[0].as_object(), json::parse(expectedOffer)); EXPECT_EQ(output.result->as_object().at("offers").as_array()[0].as_object(), json::parse(expectedOffer));
std::this_thread::sleep_for(milliseconds(20));
auto const report = subManager_->report(); auto const report = subManager_->report();
// original book + reverse book // original book + reverse book
EXPECT_EQ(report.at("books").as_uint64(), 1); EXPECT_EQ(report.at("books").as_uint64(), 1);
}); });
} }
TEST_F(RPCSubscribeHandlerTest, APIVersion)
{
auto const input = json::parse(
R"({
"streams": ["transactions_proposed"]
})"
);
auto const apiVersion = 2;
runSpawn([&, this](auto yield) {
auto const handler = AnyHandler{SubscribeHandler{backend, subManager_}};
auto const output =
handler.process(input, Context{.yield = yield, .session = session_, .apiVersion = apiVersion});
ASSERT_TRUE(output);
EXPECT_EQ(session_->apiSubVersion, apiVersion);
});
}
TEST(RPCSubscribeHandlerSpecTest, DeprecatedFields) TEST(RPCSubscribeHandlerSpecTest, DeprecatedFields)
{ {
boost::json::value const json{ boost::json::value const json{