Method OnEntryAddedToStream

src/server/server.cc:718–738

}

void Server::OnEntryAddedToStream(const std::string &ns, const std::string &key, const redis::StreamEntryID &entry_id) {
  std::lock_guard<std::mutex> guard(blocked_stream_consumers_mu_);

  auto iter = blocked_stream_consumers_.find(key);
  if (iter == blocked_stream_consumers_.end() || iter->second.empty()) {
    return;
  }

  for (auto it = iter->second.begin(); it != iter->second.end();) {
    auto consumer = *it;
    if (consumer->ns == ns && entry_id > consumer->last_consumed_id) {
      auto s = consumer->owner->EnableWriteEvent(consumer->fd);
      if (!s.IsOK()) {
        ERROR("[server] Failed to enable write event on blocked stream consumer {}: {}", consumer->fd, s.Msg());
      }
      it = iter->second.erase(it);
    } else {
      ++it;
    }
  }
}

void Server::BlockOnWait(redis::Connection *conn, rocksdb::SequenceNumber target_seq, uint64_t num_replicas) {
  std::unique_lock<std::shared_mutex> guard(wait_contexts_mu_);