Merge pull request #131 from crystal-xu/ISSUE-130

ISSUE-130: add switch for streaming svc
eBay · May 29, 2024 · 7add958 · 7add958
2 parents b92542a + fd34e0c
commit 7add958
Show file tree

Hide file tree

Showing 6 changed files with 16 additions and 3 deletions.
diff --git a/config/app_demo/raft_0.ini b/config/app_demo/raft_0.ini
@@ -10,3 +10,7 @@ storage.type = file
 storage.dir = ./node_0
 segment.data.size.limit = 67108864     ; 64MB
 segment.meta.size.limit = 4194304      ; 4MB
+
+[streaming]
+enable                  = true
+max.concurrency         = 4
diff --git a/config/app_demo/raft_1.ini b/config/app_demo/raft_1.ini
@@ -12,4 +12,5 @@ segment.data.size.limit = 67108864     ; 64MB
 segment.meta.size.limit = 4194304      ; 4MB
 
 [streaming]
+enable                  = true
 max.concurrency         = 4
diff --git a/config/app_demo/raft_2.ini b/config/app_demo/raft_2.ini
@@ -12,4 +12,5 @@ segment.data.size.limit = 67108864     ; 64MB
 segment.meta.size.limit = 4194304      ; 4MB
 
 [streaming]
+enable                  = true
 max.concurrency         = 4
diff --git a/config/app_demo/raft_3.ini b/config/app_demo/raft_3.ini
@@ -12,4 +12,5 @@ segment.data.size.limit = 67108864     ; 64MB
 segment.meta.size.limit = 4194304      ; 4MB
 
 [streaming]
+enable                  = true
 max.concurrency         = 4
diff --git a/src/infra/raft/v2/RaftCore.cpp b/src/infra/raft/v2/RaftCore.cpp
@@ -79,6 +79,7 @@ void RaftCore::initConfigurableVars(const INIReader &iniReader) {
   mMaxDecrStep = iniReader.GetInteger("raft.default", "max.decr.step", 0);
   mMaxTailedEntryNum = iniReader.GetInteger("raft.default", "max.tailed.entry.num", 0);
   mPreVoteEnabled = iniReader.GetBoolean("raft.protocol", "enable.prevote", true);
+  mStreamingSvcEnabled = iniReader.GetBoolean("streaming", "enable", true);
   mHeartBeatIntervalInMillis = iniReader.GetInteger("raft.default", "heartbeat.interval.millis",
                                                     RaftConstants::kHeartBeatIntervalInMillis);
   mMinElectionTimeoutInMillis = iniReader.GetInteger("raft.default", "min.election.timeout.millis",
@@ -121,12 +122,14 @@ void RaftCore::initConfigurableVars(const INIReader &iniReader) {
               "max.decr.step={}, "
               "max.tailed.entry.num={}, "
               "enable.provote={}, "
+              "streaming.enable={}, "
               "heartbeat.interval.millis={}, "
               "min.election.timeout.millis={}, "
               "max.election.timeout.millis={}, "
               "rpc.append.entries.timeout.millis={}, "
               "rpc.request.vote.timeout.millis={}",
-              mMaxBatchSize, mMaxLenInBytes, mMaxDecrStep, mMaxTailedEntryNum, mPreVoteEnabled,
+              mMaxBatchSize, mMaxLenInBytes, mMaxDecrStep, mMaxTailedEntryNum,
+              mPreVoteEnabled, mStreamingSvcEnabled,
               mHeartBeatIntervalInMillis, mMinElectionTimeoutInMillis, mMaxElectionTimeoutInMillis,
               mRpcAppendEntriesTimeoutInMillis, mRpcRequestVoteTimeoutInMillis);
 }
@@ -222,8 +225,10 @@ void RaftCore::initService(const INIReader &iniReader, std::shared_ptr<DNSResolv
   /// init raftMainLoop
   mRaftLoop = std::thread(&RaftCore::raftLoopMain, this);
 
-  /// init StreamingService
-  mStreamingService = std::make_unique<StreamingService>(mStreamingPort, iniReader, *this);
+  /// init StreamingService if enabled
+  if (mStreamingSvcEnabled) {
+    mStreamingService = std::make_unique<StreamingService>(mStreamingPort, iniReader, *this);
+  }
 }
 
 RaftCore::~RaftCore() {

diff --git a/src/infra/raft/v2/RaftCore.h b/src/infra/raft/v2/RaftCore.h
@@ -345,6 +345,7 @@ class RaftCore : public RaftInterface {
   std::map<uint64_t, std::unique_ptr<RaftClient>> mClients;
 
   /// streaming service
+  bool mStreamingSvcEnabled = false;
   uint64_t mStreamingPort;
   std::unique_ptr<StreamingService> mStreamingService;
   std::optional<TlsConf> mTlsConfOpt;