Method UnblockOnStreams

src/server/server.cc:675–698

}

void Server::UnblockOnStreams(const std::vector<std::string> &keys, redis::Connection *conn) {
  std::lock_guard<std::mutex> guard(blocked_stream_consumers_mu_);

  DecrBlockedClientNum();

  for (const auto &key : keys) {
    auto iter = blocked_stream_consumers_.find(key);
    if (iter == blocked_stream_consumers_.end()) {
      continue;
    }

    for (auto it = iter->second.begin(); it != iter->second.end();) {
      const auto &consumer = *it;
      if (conn->GetFD() == consumer->fd && conn->Owner() == consumer->owner) {
        iter->second.erase(it);
        if (iter->second.empty()) {
          blocked_stream_consumers_.erase(iter);
        }
        break;
      }
      ++it;
    }
  }
}

void Server::WakeupBlockingConns(const std::string &key, size_t n_conns) {
  std::lock_guard<std::mutex> guard(blocking_keys_mu_);