Method parseWriteBatch

src/cluster/replication.cc:1157–1197

}

Status ReplicationThread::parseWriteBatch(const rocksdb::WriteBatch &write_batch) {
  WriteBatchHandler write_batch_handler;

  auto db_status = write_batch.Iterate(&write_batch_handler);
  if (!db_status.ok()) return {Status::NotOK, "failed to iterate over write batch: " + db_status.ToString()};

  switch (write_batch_handler.Type()) {
    case kBatchTypePublish:
      srv_->PublishMessage(write_batch_handler.Key(), write_batch_handler.Value());
      break;
    case kBatchTypePropagate:
      if (write_batch_handler.Key() == engine::kPropagateScriptCommand) {
        std::vector<std::string> tokens = util::TokenizeRedisProtocol(write_batch_handler.Value());
        if (!tokens.empty()) {
          auto s = srv_->ExecPropagatedCommand(tokens);
          if (!s.IsOK()) {
            return s.Prefixed("failed to execute propagate command");
          }
        }
      } else if (write_batch_handler.Key() == kNamespaceDBKey) {
        auto s = srv_->GetNamespace()->LoadAndRewrite();
        if (!s.IsOK()) {
          return s.Prefixed("failed to load namespaces");
        }
      }
      break;
    case kBatchTypeStream: {
      auto key = write_batch_handler.Key();
      InternalKey ikey(key, storage_->IsSlotIdEncoded());
      Slice entry_id = ikey.GetSubKey();
      redis::StreamEntryID id;
      GetFixed64(&entry_id, &id.ms);
      GetFixed64(&entry_id, &id.seq);
      srv_->OnEntryAddedToStream(ikey.GetNamespace().ToString(), ikey.GetKey().ToString(), id);
      break;
    }
    case kBatchTypeNone:
      break;
  }
  return Status::OK();
}

bool ReplicationThread::isRestoringError(std::string_view err) {
  // err doesn't contain the CRLF, so cannot use redis::Error here.