Add test

Author: godexsoft

src/etlng/impl/GrpcSource.cpp

@@ -65,9 +65,10 @@ resolve(std::string const& ip, std::string const& port)
 
 namespace etlng::impl {
 
-GrpcSource::GrpcSource(std::string const& ip, std::string const& grpcPort)
+GrpcSource::GrpcSource(std::string const& ip, std::string const& grpcPort, std::chrono::system_clock::duration deadline)
     : log_(fmt::format("ETL_Grpc[{}:{}]", ip, grpcPort))
     , initialLoadShouldStop_(std::make_unique<std::atomic_bool>(false))
+    , deadline_{deadline}
 {
     try {
         grpc::ChannelArguments chArgs;
@@ -97,7 +98,7 @@ GrpcSource::fetchLedger(uint32_t sequence, bool getObjects, bool getObjectNeighb
     org::xrpl::rpc::v1::GetLedgerRequest request;
     grpc::ClientContext context;
 
-    context.set_deadline(std::chrono::system_clock::now() + kDEADLINE);  // Prevent indefinite blocking
+    context.set_deadline(std::chrono::system_clock::now() + deadline_);  // Prevent indefinite blocking
 
     request.mutable_ledger()->set_sequence(sequence);
     request.set_transactions(true);

src/etlng/impl/GrpcSource.hpp

@@ -41,6 +41,7 @@ class GrpcSource {
     util::Logger log_;
     std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub> stub_;
     std::unique_ptr<std::atomic_bool> initialLoadShouldStop_;
+    std::chrono::system_clock::duration deadline_;
 
     static constexpr auto kKEEPALIVE_PING_INTERVAL_MS = 10000;
     static constexpr auto kKEEPALIVE_TIMEOUT_MS = 5000;
@@ -49,7 +50,11 @@ class GrpcSource {
     static constexpr auto kDEADLINE = std::chrono::seconds(30);
 
 public:
-    GrpcSource(std::string const& ip, std::string const& grpcPort);
+    GrpcSource(
+        std::string const& ip,
+        std::string const& grpcPort,
+        std::chrono::system_clock::duration deadline = kDEADLINE
+    );
 
     /**
      * @brief Fetch data for a specific ledger.

tests/common/util/MockXrpLedgerAPIService.hpp

@@ -32,7 +32,9 @@
 #include <org/xrpl/rpc/v1/get_ledger_entry.pb.h>
 #include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
+#include <chrono>
 #include <memory>
+#include <optional>
 #include <string>
 #include <thread>
 
@@ -90,15 +92,27 @@ struct WithMockXrpLedgerAPIService : virtual ::testing::Test {
 
     ~WithMockXrpLedgerAPIService() override
     {
-        server_->Shutdown();
-        serverThread_.join();
+        shutdown();
     }
 
     int
     getXRPLMockPort() const
     {
         return port_;
     }
+
+    void
+    shutdown(std::optional<std::chrono::system_clock::duration> deadline = std::nullopt)
+    {
+        if (deadline.has_value()) {
+            server_->Shutdown(std::chrono::system_clock::now() + *deadline);
+        } else {
+            server_->Shutdown();
+        }
+        if (serverThread_.joinable())
+            serverThread_.join();
+    }
+
     MockXrpLedgerAPIService mockXrpLedgerAPIService;
 
 private:

tests/unit/etlng/GrpcSourceTests.cpp

@@ -41,16 +41,20 @@
 #include <xrpl/basics/strHex.h>
 
 #include <atomic>
+#include <chrono>
 #include <condition_variable>
 #include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <future>
 #include <map>
+#include <memory>
 #include <mutex>
 #include <optional>
 #include <queue>
+#include <semaphore>
 #include <string>
+#include <thread>
 #include <vector>
 
 using namespace etlng::model;
@@ -357,3 +361,40 @@ TEST_F(GrpcSourceStopTests, LoadInitialLedgerStopsWhenRequested)
     ASSERT_FALSE(res.has_value());
     EXPECT_EQ(res.error(), etlng::InitialLedgerLoadError::Cancelled);
 }
+
+TEST_F(GrpcSourceNgTests, DeadlineIsHandledCorrectly)
+{
+    uint32_t const sequence = 123u;
+    bool const getObjects = true;
+    bool const getObjectNeighbors = false;
+
+    std::mutex mtx;
+    std::condition_variable cv;
+    bool finished = false;
+
+    auto grpcSource = std::make_unique<etlng::impl::GrpcSource>(
+        "localhost", std::to_string(getXRPLMockPort()), std::chrono::milliseconds{1}
+    );
+
+    EXPECT_CALL(mockXrpLedgerAPIService, GetLedger)
+        .WillOnce([&](grpc::ServerContext*,
+                      org::xrpl::rpc::v1::GetLedgerRequest const*,
+                      org::xrpl::rpc::v1::GetLedgerResponse*) {
+            std::unique_lock lk(mtx);
+            cv.wait(lk, [&] { return finished; });
+
+            return grpc::Status{};
+        });
+
+    auto const [status, response] = grpcSource->fetchLedger(sequence, getObjects, getObjectNeighbors);
+
+    {
+        std::unique_lock lk(mtx);
+        finished = true;
+    }
+    cv.notify_all();
+    grpcSource.reset();
+
+    ASSERT_FALSE(status.ok());
+    shutdown(std::chrono::milliseconds{10});
+}