refactor: Remove SubscriptionManagerRunner (#1623)

src/app/ClioApplication.cpp

@@ -101,9 +101,7 @@ ClioApplication::run()
     auto backend = data::make_Backend(config_);
 
     // Manages clients subscribed to streams
-    auto subscriptionsRunner = feed::SubscriptionManagerRunner(config_, backend);
-
-    auto const subscriptions = subscriptionsRunner.getManager();
+    auto subscriptions = feed::SubscriptionManager::make_SubscriptionManager(config_, backend);
 
     // Tracks which ledgers have been validated by the network
     auto ledgers = etl::NetworkValidatedLedgers::make_ValidatedLedgers();

src/feed/SubscriptionManager.hpp

@@ -30,6 +30,7 @@
 #include "feed/impl/TransactionFeed.hpp"
 #include "util/async/AnyExecutionContext.hpp"
 #include "util/async/context/BasicExecutionContext.hpp"
+#include "util/config/Config.hpp"
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/executor_work_guard.hpp>
@@ -44,6 +45,7 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <utility>
 #include <vector>
 
 /**
@@ -67,16 +69,36 @@ class SubscriptionManager : public SubscriptionManagerInterface {
     impl::ProposedTransactionFeed proposedTransactionFeed_;
 
 public:
+    /**
+     * @brief Factory function to create a new SubscriptionManager with a PoolExecutionContext.
+     *
+     * @param config The configuration to use
+     * @param backend The backend to use
+     * @return A shared pointer to a new instance of SubscriptionManager
+     */
+    static std::shared_ptr<SubscriptionManager>
+    make_SubscriptionManager(util::Config const& config, std::shared_ptr<data::BackendInterface const> const& backend)
+    {
+        auto const workersNum = config.valueOr<std::uint64_t>("subscription_workers", 1);
+
+        util::Logger logger{"Subscriptions"};
+        LOG(logger.info()) << "Starting subscription manager with " << workersNum << " workers";
+
+        return std::make_shared<feed::SubscriptionManager>(util::async::PoolExecutionContext(workersNum), backend);
+    }
+
     /**
      * @brief Construct a new Subscription Manager object
      *
      * @param executor The executor to use to publish the feeds
      * @param backend The backend to use
      */
-    template <class ExecutorCtx>
-    SubscriptionManager(ExecutorCtx& executor, std::shared_ptr<data::BackendInterface const> const& backend)
+    SubscriptionManager(
+        util::async::AnyExecutionContext&& executor,
+        std::shared_ptr<data::BackendInterface const> const& backend
+    )
         : backend_(backend)
-        , ctx_(executor)
+        , ctx_(std::move(executor))
         , manifestFeed_(ctx_, "manifest")
         , validationsFeed_(ctx_, "validations")
         , ledgerFeed_(ctx_)
@@ -291,41 +313,4 @@ public:
     report() const final;
 };
 
-/**
- * @brief The help class to run the subscription manager. The container of PoolExecutionContext which is used to publish
- * the feeds.
- */
-class SubscriptionManagerRunner {
-    std::uint64_t workersNum_;
-    using ActualExecutionCtx = util::async::PoolExecutionContext;
-    ActualExecutionCtx ctx_;
-    std::shared_ptr<SubscriptionManager> subscriptionManager_;
-    util::Logger logger_{"Subscriptions"};
-
-public:
-    /**
-     * @brief Construct a new Subscription Manager Runner object
-     *
-     * @param config The configuration
-     * @param backend The backend to use
-     */
-    SubscriptionManagerRunner(util::Config const& config, std::shared_ptr<data::BackendInterface> const& backend)
-        : workersNum_(config.valueOr<std::uint64_t>("subscription_workers", 1))
-        , ctx_(workersNum_)
-        , subscriptionManager_(std::make_shared<SubscriptionManager>(ctx_, backend))
-    {
-        LOG(logger_.info()) << "Starting subscription manager with " << workersNum_ << " workers";
-    }
-
-    /**
-     * @brief Get the subscription manager
-     *
-     * @return The subscription manager
-     */
-    std::shared_ptr<SubscriptionManager>
-    getManager()
-    {
-        return subscriptionManager_;
-    }
-};
 }  // namespace feed

src/util/async/context/BasicExecutionContext.hpp

@@ -369,7 +369,7 @@ public:
      * @brief Block until all operations are completed
      */
     void
-    join() noexcept
+    join() const noexcept
     {
         context_.join();
     }

tests/unit/feed/LedgerFeedTests.cpp

@@ -21,6 +21,7 @@
 #include "feed/impl/LedgerFeed.hpp"
 #include "util/TestObject.hpp"
 
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/parse.hpp>
 #include <gmock/gmock.h>
@@ -57,11 +58,13 @@ TEST_F(FeedLedgerTest, SubPub)
             "reserve_base":3,
             "reserve_inc":2
         })";
-    boost::asio::spawn(ctx, [this](boost::asio::yield_context yield) {
+    boost::asio::io_context ioContext;
+    boost::asio::spawn(ioContext, [this](boost::asio::yield_context yield) {
         auto res = testFeedPtr->sub(yield, backend, sessionPtr);
         // check the response
         EXPECT_EQ(res, json::parse(LedgerResponse));
     });
+    ioContext.run();
     EXPECT_EQ(testFeedPtr->count(), 1);
 
     constexpr static auto ledgerPub =

tests/unit/feed/SubscriptionManagerTests.cpp

@@ -28,6 +28,7 @@
 #include "util/async/context/SyncExecutionContext.hpp"
 #include "web/interface/ConnectionBase.hpp"
 
+#include <boost/asio/io_context.hpp>
 #include <boost/asio/spawn.hpp>
 #include <boost/json/object.hpp>
 #include <boost/json/parse.hpp>
@@ -57,13 +58,12 @@ class SubscriptionManagerBaseTest : public util::prometheus::WithPrometheus, pub
 protected:
     std::shared_ptr<SubscriptionManager> subscriptionManagerPtr;
     std::shared_ptr<web::ConnectionBase> session;
-    Execution ctx{2};
     MockSession* sessionPtr = nullptr;
 
     void
     SetUp() override
     {
-        subscriptionManagerPtr = std::make_shared<SubscriptionManager>(ctx, backend);
+        subscriptionManagerPtr = std::make_shared<SubscriptionManager>(Execution(2), backend);
         session = std::make_shared<MockSession>();
         session->apiSubVersion = 1;
         sessionPtr = dynamic_cast<MockSession*>(session.get());
@@ -264,11 +264,13 @@ TEST_F(SubscriptionManagerTest, LedgerTest)
             "reserve_base":3,
             "reserve_inc":2
         })";
+    boost::asio::io_context ctx;
     boost::asio::spawn(ctx, [this](boost::asio::yield_context yield) {
         auto const res = subscriptionManagerPtr->subLedger(yield, session);
         // check the response
         EXPECT_EQ(res, json::parse(LedgerResponse));
     });
+    ctx.run();
     EXPECT_EQ(subscriptionManagerPtr->report()["ledger"], 1);
 
     // test publish