chore: pass SinkReplyBuilder and Transaction explicitly. Part10 (#3998)

dragonflydb · Oct 28, 2024 · 1bdd56c · 1bdd56c
1 parent 41d8c66
commit 1bdd56c
Show file tree

Hide file tree

Showing 8 changed files with 342 additions and 294 deletions.
diff --git a/src/server/dflycmd.cc b/src/server/dflycmd.cc
@@ -489,7 +489,7 @@ void DflyCmd::TakeOver(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext
   VLOG(1) << "Takeover accepted, shutting down.";
   std::string save_arg = "NOSAVE";
   MutableSlice sargs(save_arg);
-  return sf_->ShutdownCmd(CmdArgList(&sargs, 1), cntx);
+  return sf_->ShutdownCmd(CmdArgList(&sargs, 1), nullptr, rb, cntx);
 }
 
 void DflyCmd::Expire(CmdArgList args, RedisReplyBuilder* rb, ConnectionContext* cntx) {

diff --git a/src/server/main_service.cc b/src/server/main_service.cc
@@ -1619,7 +1619,7 @@ void Service::DispatchMC(const MemcacheParser::Command& cmd, std::string_view va
       strcpy(cmd_name, "QUIT");
       break;
     case MemcacheParser::STATS:
-      server_family_.StatsMC(cmd.key, cntx);
+      server_family_.StatsMC(cmd.key, mc_builder);
       return;
     case MemcacheParser::VERSION:
       mc_builder->SendSimpleString("VERSION 1.5.0 DF");

diff --git a/src/server/replica.cc b/src/server/replica.cc
@@ -90,18 +90,18 @@ Replica::~Replica() {
 
 static const char kConnErr[] = "could not connect to master: ";
 
-error_code Replica::Start(ConnectionContext* cntx) {
+error_code Replica::Start(facade::SinkReplyBuilder* builder) {
   VLOG(1) << "Starting replication";
   ProactorBase* mythread = ProactorBase::me();
   CHECK(mythread);
 
-  auto check_connection_error = [this, &cntx](error_code ec, const char* msg) -> error_code {
+  auto check_connection_error = [this, builder](error_code ec, const char* msg) -> error_code {
     if (cntx_.IsCancelled()) {
-      cntx->SendError("replication cancelled");
+      builder->SendError("replication cancelled");
       return std::make_error_code(errc::operation_canceled);
     }
     if (ec) {
-      cntx->SendError(absl::StrCat(msg, ec.message()));
+      builder->SendError(absl::StrCat(msg, ec.message()));
       cntx_.Cancel();
     }
     return ec;
@@ -131,17 +131,17 @@ error_code Replica::Start(ConnectionContext* cntx) {
   // 4. Spawn main coordination fiber.
   sync_fb_ = fb2::Fiber("main_replication", &Replica::MainReplicationFb, this);
 
-  cntx->SendOk();
+  builder->SendOk();
   return {};
 }
 
-void Replica::EnableReplication(ConnectionContext* cntx) {
+void Replica::EnableReplication(facade::SinkReplyBuilder* builder) {
   VLOG(1) << "Enabling replication";
 
   state_mask_.store(R_ENABLED);                             // set replica state to enabled
   sync_fb_ = MakeFiber(&Replica::MainReplicationFb, this);  // call replication fiber
 
-  cntx->SendOk();
+  builder->SendOk();
 }
 
 void Replica::Stop() {

diff --git a/src/server/replica.h b/src/server/replica.h
@@ -58,12 +58,12 @@ class Replica : ProtocolClient {
   // Spawns a fiber that runs until link with master is broken or the replication is stopped.
   // Returns true if initial link with master has been established or
   // false if it has failed.
-  std::error_code Start(ConnectionContext* cntx);
+  std::error_code Start(facade::SinkReplyBuilder* builder);
 
   // Sets the server state to have replication enabled.
   // It is like Start(), but does not attempt to establish
   // a connection right-away, but instead lets MainReplicationFb do the work.
-  void EnableReplication(ConnectionContext* cntx);
+  void EnableReplication(facade::SinkReplyBuilder* builder);
 
   void Stop();  // thread-safe