merged a bunch of stuff

handlers/AccountObjects.cpp

@@ -1,5 +1,5 @@
-#include <ripple/app/paths/RippleState.h>
 #include <ripple/app/ledger/Ledger.h>
+#include <ripple/app/paths/RippleState.h>
 #include <ripple/basics/StringUtilities.h>
 #include <ripple/protocol/ErrorCodes.h>
 #include <ripple/protocol/Indexes.h>
@@ -104,19 +104,14 @@ doAccountObjects(
     auto const addToResponse = [&](ripple::SLE const& sle) {
         if (!objectType || objectType == sle.getType())
         {
-            jsonObjects.push_back(getJson(sle));
+            jsonObjects.push_back(toJson(sle));
         }
 
         return true;
     };
 
-    auto nextCursor = 
-        traverseOwnedNodes(
-            backend,
-            accountID,
-            *ledgerSequence,
-            cursor,
-            addToResponse);
+    auto nextCursor = traverseOwnedNodes(
+        backend, accountID, *ledgerSequence, cursor, addToResponse);
 
     if (nextCursor)
         response["next_cursor"] = ripple::strHex(*nextCursor);

handlers/RPCHelpers.cpp

@@ -104,6 +104,11 @@ toJson(ripple::SLE const& sle)
             .count();
     return value.as_object();
 }
+boost::json::value
+toBoostJson(RippledJson const& value)
+{
+    return boost::json::parse(value.toStyledString());
+}
 
 boost::json::object
 toJson(ripple::LedgerInfo const& lgrInfo)

handlers/RPCHelpers.h

@@ -32,10 +32,11 @@ boost::json::object
 toJson(ripple::LedgerInfo const& info);
 
 boost::json::object
-getJson(ripple::TxMeta const& meta);
+toJson(ripple::TxMeta const& meta);
 
+using RippledJson = Json::Value;
 boost::json::value
-getJson(Json::Value const& value);
+toBoostJson(RippledJson const& value);
 
 std::optional<uint32_t>
 ledgerSequenceFromRequest(

reporting/ReportingETL.cpp

@@ -241,8 +241,7 @@ ReportingETL::publishLedger(uint32_t ledgerSequence, uint32_t maxAttempts)
             }
             else
             {
-                auto lgr =
-                    flatMapBackend_->fetchLedgerBySequence(ledgerSequence);
+                auto lgr = backend_->fetchLedgerBySequence(ledgerSequence);
                 assert(lgr);
                 publishLedger(*lgr);
             }
@@ -574,7 +573,7 @@ ReportingETL::runETLPipeline(uint32_t startSequence, int numExtractors)
                     BOOST_LOG_TRIVIAL(info) << "Running online delete";
                     backend_->doOnlineDelete(*onlineDeleteInterval_);
                     BOOST_LOG_TRIVIAL(info) << "Finished online delete";
-                    auto rng = flatMapBackend_->fetchLedgerRangeNoThrow();
+                    auto rng = backend_->fetchLedgerRangeNoThrow();
                     minSequence = rng->minSequence;
                     deleting_ = false;
                 });

reporting/ReportingETL.h

@@ -276,7 +276,7 @@ private:
     {
         boost::json::object result;
 
-        result["etl_sources"] = loadBalancer_.toJson();
+        result["etl_sources"] = loadBalancer_->toJson();
         result["is_writer"] = writing_.load();
         result["read_only"] = readOnly_;
         auto last = getLastPublish();

reporting/server/SubscriptionManager.cpp

@@ -1,5 +1,5 @@
-#include<reporting/server/SubscriptionManager.h>
 #include <handlers/RPCHelpers.h>
+#include <reporting/server/SubscriptionManager.h>
 
 void
 SubscriptionManager::subLedger(std::shared_ptr<session>& session)
@@ -27,10 +27,10 @@ SubscriptionManager::pubLedger(
     pubMsg["ledger_hash"] = to_string(lgrInfo.hash);
     pubMsg["ledger_time"] = lgrInfo.closeTime.time_since_epoch().count();
 
-    pubMsg["fee_ref"] = getJson(fees.units.jsonClipped());
-    pubMsg["fee_base"] = getJson(fees.base.jsonClipped());
-    pubMsg["reserve_base"] = getJson(fees.accountReserve(0).jsonClipped());
-    pubMsg["reserve_inc"] = getJson(fees.increment.jsonClipped());
+    pubMsg["fee_ref"] = toBoostJson(fees.units.jsonClipped());
+    pubMsg["fee_base"] = toBoostJson(fees.base.jsonClipped());
+    pubMsg["reserve_base"] = toBoostJson(fees.accountReserve(0).jsonClipped());
+    pubMsg["reserve_inc"] = toBoostJson(fees.increment.jsonClipped());
 
     pubMsg["validated_ledgers"] = ledgerRange;
     pubMsg["txn_count"] = txnCount;
@@ -75,8 +75,8 @@ SubscriptionManager::pubTransaction(
     auto [tx, meta] = deserializeTxPlusMeta(blob, seq);
 
     boost::json::object pubMsg;
-    pubMsg["transaction"] = getJson(*tx);
-    pubMsg["meta"] = getJson(*meta);
+    pubMsg["transaction"] = toJson(*tx);
+    pubMsg["meta"] = toJson(*meta);
 
     for (auto const& session : streamSubscribers_[Transactions])
         session->send(boost::json::serialize(pubMsg));
@@ -89,9 +89,9 @@ SubscriptionManager::pubTransaction(
             session->send(boost::json::serialize(pubMsg));
 }
 
-
 void
-SubscriptionManager::forwardProposedTransaction(boost::json::object const& response)
+SubscriptionManager::forwardProposedTransaction(
+    boost::json::object const& response)
 {
     for (auto const& session : streamSubscribers_[TransactionsProposed])
         session->send(boost::json::serialize(response));
@@ -127,7 +127,8 @@ SubscriptionManager::subProposedTransactions(std::shared_ptr<session>& session)
 }
 
 void
-SubscriptionManager::unsubProposedTransactions(std::shared_ptr<session>& session)
+SubscriptionManager::unsubProposedTransactions(
+    std::shared_ptr<session>& session)
 {
     streamSubscribers_[TransactionsProposed].erase(session);
 }

reporting/server/listener.h

@@ -48,10 +48,11 @@ public:
         boost::asio::ip::tcp::endpoint endpoint,
         std::shared_ptr<BackendInterface> backend,
         std::shared_ptr<SubscriptionManager> subscriptions,
-        std::shared_ptr<ETLLoadBalancer> balancer)
+        std::shared_ptr<ETLLoadBalancer> balancer,
+        DOSGuard& dosGuard)
     {
         std::make_shared<listener>(
-            ioc, endpoint, backend, subscriptions, balancer)
+            ioc, endpoint, backend, subscriptions, balancer, dosGuard)
             ->run();
     }
 
@@ -67,6 +68,7 @@ public:
         , backend_(backend)
         , subscriptions_(subscriptions)
         , balancer_(balancer)
+        , dosGuard_(dosGuard)
     {
         boost::beast::error_code ec;
 

reporting/server/session.cpp

@@ -20,20 +20,20 @@ buildResponse(
     boost::json::object response;
 
     if (shouldForwardToP2p(request))
-        return balancer->forwardToP2p(request);
+        return {balancer->forwardToP2p(request), 10};
 
     switch (commandMap[command])
     {
         case tx:
             return {doTx(request, *backend), 1};
         case account_tx: {
-            auto res = doAccountTx(request, backend);
+            auto res = doAccountTx(request, *backend);
             if (res.contains("transactions"))
                 return {res, res["transactions"].as_array().size()};
             return {res, 1};
         }
         case ledger: {
-            auto res = doLedger(request, backend);
+            auto res = doLedger(request, *backend);
             if (res.contains("transactions"))
                 return {res, res["transactions"].as_array().size()};
             return {res, 1};
@@ -43,7 +43,7 @@ buildResponse(
         case ledger_range:
             return {doLedgerRange(request, *backend), 1};
         case ledger_data: {
-            auto res = doLedgerData(request, backend);
+            auto res = doLedgerData(request, *backend);
             if (res.contains("objects"))
                 return {res, res["objects"].as_array().size() * 4};
             return {res, 1};
@@ -51,7 +51,7 @@ buildResponse(
         case account_info:
             return {doAccountInfo(request, *backend), 1};
         case book_offers: {
-            auto res = doBookOffers(request, backend);
+            auto res = doBookOffers(request, *backend);
             if (res.contains("offers"))
                 return {res, res["offers"].as_array().size() * 4};
             return {res, 1};
@@ -73,7 +73,7 @@ buildResponse(
             size_t count = 1;
             if (res.contains("send_currencies"))
                 count = res["send_currencies"].as_array().size();
-            if(res.contains("receive_currencies"]))
+            if (res.contains("receive_currencies"))
                 count += res["receive_currencies"].as_array().size();
             return {res, count};
         }
@@ -100,7 +100,7 @@ buildResponse(
         case unsubscribe:
             return {doUnsubscribe(request, session, *manager), 1};
         case server_info: {
-            return {doServerInfo(request, backend), 1};
+            return {doServerInfo(request, *backend), 1};
             break;
         }
         default:

reporting/server/session.h

@@ -27,6 +27,7 @@
 #include <reporting/BackendInterface.h>
 #include <reporting/ETLSource.h>
 #include <reporting/server/SubscriptionManager.h>
+#include <server/DOSGuard.h>
 
 class session;
 class SubscriptionManager;
@@ -134,6 +135,10 @@ boost::json::object
 doAccountObjects(
     boost::json::object const& request,
     BackendInterface const& backend);
+boost::json::object
+doServerInfo(
+    boost::json::object const& request,
+    BackendInterface const& backend);
 
 boost::json::object
 doChannelAuthorize(boost::json::object const& request);
@@ -151,7 +156,7 @@ doUnsubscribe(
     std::shared_ptr<session>& session,
     SubscriptionManager& manager);
 
-boost::json::object
+std::pair<boost::json::object, uint32_t>
 buildResponse(
     boost::json::object const& request,
     std::shared_ptr<BackendInterface> backend,

server/websocket_server_async.cpp

@@ -32,386 +32,11 @@
 #include <reporting/ReportingETL.h>
 #include <reporting/server/listener.h>
 #include <reporting/server/session.h>
-#include <server/DOSGuard.h>
 #include <sstream>
 #include <string>
 #include <thread>
 #include <vector>
 
-<<<<<<< HEAD:websocket_server_async.cpp
-=======
-//------------------------------------------------------------------------------
-enum RPCCommand {
-    tx,
-    account_tx,
-    ledger,
-    account_info,
-    ledger_data,
-    book_offers,
-    ledger_range,
-    ledger_entry,
-    server_info
-};
-std::unordered_map<std::string, RPCCommand> commandMap{
-    {"tx", tx},
-    {"account_tx", account_tx},
-    {"ledger", ledger},
-    {"ledger_range", ledger_range},
-    {"ledger_entry", ledger_entry},
-    {"account_info", account_info},
-    {"ledger_data", ledger_data},
-    {"book_offers", book_offers},
-    {"server_info", server_info}};
-
-boost::json::object
-doAccountInfo(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doTx(boost::json::object const& request, BackendInterface const& backend);
-boost::json::object
-doAccountTx(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doLedgerData(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doLedgerEntry(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doBookOffers(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doLedger(boost::json::object const& request, BackendInterface const& backend);
-boost::json::object
-doLedgerRange(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-boost::json::object
-doServerInfo(
-    boost::json::object const& request,
-    BackendInterface const& backend);
-
-std::pair<boost::json::object, uint32_t>
-buildResponse(
-    boost::json::object const& request,
-    BackendInterface const& backend)
-{
-    std::string command = request.at("command").as_string().c_str();
-    BOOST_LOG_TRIVIAL(info) << "Received rpc command : " << request;
-    boost::json::object response;
-    switch (commandMap[command])
-    {
-        case tx:
-            return {doTx(request, backend), 1};
-            break;
-        case account_tx: {
-            auto res = doAccountTx(request, backend);
-            if (res.contains("transactions"))
-                return {res, res["transactions"].as_array().size()};
-            return {res, 1};
-        }
-        break;
-        case ledger: {
-            auto res = doLedger(request, backend);
-            if (res.contains("transactions"))
-                return {res, res["transactions"].as_array().size()};
-            return {res, 1};
-        }
-        break;
-        case ledger_entry:
-            return {doLedgerEntry(request, backend), 1};
-            break;
-        case ledger_range:
-            return {doLedgerRange(request, backend), 1};
-            break;
-        case ledger_data: {
-            auto res = doLedgerData(request, backend);
-            if (res.contains("objects"))
-                return {res, res["objects"].as_array().size()};
-            return {res, 1};
-        }
-        break;
-        case server_info: {
-            return {doServerInfo(request, backend), 1};
-            break;
-        }
-        case account_info:
-            return {doAccountInfo(request, backend), 1};
-            break;
-        case book_offers: {
-            auto res = doBookOffers(request, backend);
-            if (res.contains("offers"))
-                return {res, res["offers"].as_array().size()};
-            return {res, 1};
-        }
-        break;
-        default:
-            BOOST_LOG_TRIVIAL(error) << "Unknown command: " << command;
-    }
-    return {response, 1};
-}
-// Report a failure
-void
-fail(boost::beast::error_code ec, char const* what)
-{
-    std::cerr << what << ": " << ec.message() << "\n";
-}
-
-// Echoes back all received WebSocket messages
-class session : public std::enable_shared_from_this<session>
-{
-    boost::beast::websocket::stream<boost::beast::tcp_stream> ws_;
-    boost::beast::flat_buffer buffer_;
-    std::string response_;
-    BackendInterface const& backend_;
-    DOSGuard& dosGuard_;
-
-public:
-    // Take ownership of the socket
-    explicit session(
-        boost::asio::ip::tcp::socket&& socket,
-        BackendInterface const& backend,
-        DOSGuard& dosGuard)
-        : ws_(std::move(socket)), backend_(backend), dosGuard_(dosGuard)
-    {
-    }
-
-    // Get on the correct executor
-    void
-    run()
-    {
-        // We need to be executing within a strand to perform async
-        // operations on the I/O objects in this session. Although not
-        // strictly necessary for single-threaded contexts, this example
-        // code is written to be thread-safe by default.
-        boost::asio::dispatch(
-            ws_.get_executor(),
-            boost::beast::bind_front_handler(
-                &session::on_run, shared_from_this()));
-    }
-
-    // Start the asynchronous operation
-    void
-    on_run()
-    {
-        // Set suggested timeout settings for the websocket
-        ws_.set_option(boost::beast::websocket::stream_base::timeout::suggested(
-            boost::beast::role_type::server));
-
-        // Set a decorator to change the Server of the handshake
-        ws_.set_option(boost::beast::websocket::stream_base::decorator(
-            [](boost::beast::websocket::response_type& res) {
-                res.set(
-                    boost::beast::http::field::server,
-                    std::string(BOOST_BEAST_VERSION_STRING) +
-                        " websocket-server-async");
-            }));
-        // Accept the websocket handshake
-        ws_.async_accept(boost::beast::bind_front_handler(
-            &session::on_accept, shared_from_this()));
-    }
-
-    void
-    on_accept(boost::beast::error_code ec)
-    {
-        if (ec)
-            return fail(ec, "accept");
-
-        // Read a message
-        do_read();
-    }
-
-    void
-    do_read()
-    {
-        // Read a message into our buffer
-        ws_.async_read(
-            buffer_,
-            boost::beast::bind_front_handler(
-                &session::on_read, shared_from_this()));
-    }
-
-    void
-    on_read(boost::beast::error_code ec, std::size_t bytes_transferred)
-    {
-        boost::ignore_unused(bytes_transferred);
-
-        // This indicates that the session was closed
-        if (ec == boost::beast::websocket::error::closed)
-            return;
-
-        if (ec)
-            fail(ec, "read");
-        std::string msg{
-            static_cast<char const*>(buffer_.data().data()), buffer_.size()};
-        // BOOST_LOG_TRIVIAL(debug) << __func__ << msg;
-        boost::json::object response;
-        auto ip =
-            ws_.next_layer().socket().remote_endpoint().address().to_string();
-        BOOST_LOG_TRIVIAL(debug)
-            << __func__ << " received request from ip = " << ip;
-        if (!dosGuard_.isOk(ip))
-            response["error"] = "Too many requests. Slow down";
-        else
-        {
-            try
-            {
-                boost::json::value raw = boost::json::parse(msg);
-                boost::json::object request = raw.as_object();
-                BOOST_LOG_TRIVIAL(debug) << " received request : " << request;
-                try
-                {
-                    auto start = std::chrono::system_clock::now();
-                    auto [res, cost] = buildResponse(request, backend_);
-                    response = std::move(res);
-                    if (!dosGuard_.add(ip, cost))
-                    {
-                        response["warning"] = "Too many requests";
-                    }
-
-                    auto end = std::chrono::system_clock::now();
-                    BOOST_LOG_TRIVIAL(info)
-                        << __func__ << " RPC call took "
-                        << ((end - start).count() / 1000000000.0)
-                        << " . request = " << request;
-                }
-                catch (Backend::DatabaseTimeout const& t)
-                {
-                    BOOST_LOG_TRIVIAL(error) << __func__ << " Database timeout";
-                    response["error"] =
-                        "Database read timeout. Please retry the request";
-                }
-            }
-            catch (std::exception const& e)
-            {
-                BOOST_LOG_TRIVIAL(error)
-                    << __func__ << "caught exception : " << e.what();
-                response["error"] = "Unknown exception";
-            }
-        }
-        BOOST_LOG_TRIVIAL(trace) << __func__ << response;
-        response_ = boost::json::serialize(response);
-
-        // Echo the message
-        ws_.text(ws_.got_text());
-        ws_.async_write(
-            boost::asio::buffer(response_),
-            boost::beast::bind_front_handler(
-                &session::on_write, shared_from_this()));
-    }
-
-    void
-    on_write(boost::beast::error_code ec, std::size_t bytes_transferred)
-    {
-        boost::ignore_unused(bytes_transferred);
-
-        if (ec)
-            return fail(ec, "write");
-
-        // Clear the buffer
-        buffer_.consume(buffer_.size());
-
-        // Do another read
-        do_read();
-    }
-};
-
-//------------------------------------------------------------------------------
-
-// Accepts incoming connections and launches the sessions
-class listener : public std::enable_shared_from_this<listener>
-{
-    boost::asio::io_context& ioc_;
-    boost::asio::ip::tcp::acceptor acceptor_;
-    BackendInterface const& backend_;
-    DOSGuard& dosGuard_;
-
-public:
-    listener(
-        boost::asio::io_context& ioc,
-        boost::asio::ip::tcp::endpoint endpoint,
-        BackendInterface const& backend,
-        DOSGuard& dosGuard)
-        : ioc_(ioc), acceptor_(ioc), backend_(backend), dosGuard_(dosGuard)
-    {
-        boost::beast::error_code ec;
-
-        // Open the acceptor
-        acceptor_.open(endpoint.protocol(), ec);
-        if (ec)
-        {
-            fail(ec, "open");
-            return;
-        }
-
-        // Allow address reuse
-        acceptor_.set_option(boost::asio::socket_base::reuse_address(true), ec);
-        if (ec)
-        {
-            fail(ec, "set_option");
-            return;
-        }
-
-        // Bind to the server address
-        acceptor_.bind(endpoint, ec);
-        if (ec)
-        {
-            fail(ec, "bind");
-            return;
-        }
-
-        // Start listening for connections
-        acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
-        if (ec)
-        {
-            fail(ec, "listen");
-            return;
-        }
-    }
-
-    // Start accepting incoming connections
-    void
-    run()
-    {
-        do_accept();
-    }
-
-private:
-    void
-    do_accept()
-    {
-        // The new connection gets its own strand
-        acceptor_.async_accept(
-            boost::asio::make_strand(ioc_),
-            boost::beast::bind_front_handler(
-                &listener::on_accept, shared_from_this()));
-    }
-
-    void
-    on_accept(boost::beast::error_code ec, boost::asio::ip::tcp::socket socket)
-    {
-        if (ec)
-        {
-            fail(ec, "accept");
-        }
-        else
-        {
-            // Create the session and run it
-            std::make_shared<session>(std::move(socket), backend_, dosGuard_)
-                ->run();
-        }
-
-        // Accept another connection
-        do_accept();
-    }
-};
-
->>>>>>> dev:server/websocket_server_async.cpp
 std::optional<boost::json::object>
 parse_config(const char* filename)
 {
@@ -515,34 +140,20 @@ main(int argc, char* argv[])
 
     // The io_context is required for all I/O
     boost::asio::io_context ioc{threads};
-<<<<<<< HEAD:websocket_server_async.cpp
+
+    DOSGuard dosGuard{config.value(), ioc};
 
     std::shared_ptr<BackendInterface> backend{Backend::make_Backend(*config)};
 
     std::shared_ptr<SubscriptionManager> subscriptions{
         SubscriptionManager::make_SubscriptionManager()};
-=======
-    ReportingETL etl{config.value(), ioc};
-    DOSGuard dosGuard{config.value(), ioc};
->>>>>>> dev:server/websocket_server_async.cpp
 
     std::shared_ptr<NetworkValidatedLedgers> ledgers{
         NetworkValidatedLedgers::make_ValidatedLedgers()};
 
     std::shared_ptr<ETLLoadBalancer> balancer{
         ETLLoadBalancer::make_ETLLoadBalancer(
-            *config,
-            ioc,
-<<<<<<< HEAD:websocket_server_async.cpp
-            backend,
-            subscriptions,
-            ledgers)};
-=======
-            boost::asio::ip::tcp::endpoint{address, port},
-            etl.getFlatMapBackend(),
-            dosGuard)
-            ->run();
->>>>>>> dev:server/websocket_server_async.cpp
+            *config, ioc, backend, subscriptions, ledgers)};
 
     std::shared_ptr<ReportingETL> etl{ReportingETL::make_ReportingETL(
         *config, ioc, backend, subscriptions, balancer, ledgers)};
@@ -552,7 +163,8 @@ main(int argc, char* argv[])
         boost::asio::ip::tcp::endpoint{address, port},
         backend,
         subscriptions,
-        balancer);
+        balancer,
+        dosGuard);
 
     // Blocks until stopped.
     // When stopped, shared_ptrs fall out of scope

unittests/main.cpp

@@ -45,7 +45,7 @@ TEST(BackendTest, Basic)
     for (auto& config : configs)
     {
         std::cout << keyspace << std::endl;
-        auto backend = Backend::makeBackend(config);
+        auto backend = Backend::make_Backend(config);
         backend->open(false);
 
         std::string rawHeader =