(
    absl::string_view rpc_name,
    const Token& token,
    GrpcStreamObserver* observer) {
  EnsureActiveStub();

  auto context = CreateContext(token);
  auto call =
      grpc_stub_->PrepareCall(context.get(), MakeString(rpc_name), grpc_queue_);
  return absl::make_unique<GrpcStream>(std::move(context), std::move(call),
                                       worker_queue_, this, observer);
}

std::unique_ptr<GrpcUnaryCall> GrpcConnection::CreateUnaryCall(
    absl::string_view rpc_name,
    const Token& token,
    const grpc::ByteBuffer& message) {
  EnsureActiveStub();

  auto context = CreateContext(token);
  auto call = grpc_stub_->PrepareUnaryCall(context.get(), MakeString(rpc_name),
                                           message, grpc_queue_);
  return absl::make_unique<GrpcUnaryCall>(std::move(context), std::move(call),
                                          worker_queue_, this, message);
}

std::unique_ptr<GrpcStreamingReader> GrpcConnection::CreateStreamingReader(
    absl::string_view rpc_name,
    const Token& token,
    const grpc::ByteBuffer& message) {
  EnsureActiveStub();

  auto context = CreateContext(token);
  auto call =
      grpc_stub_->PrepareCall(context.get(), MakeString(rpc_name), grpc_queue_);
  return absl::make_unique<GrpcStreamingReader>(
      std::move(context), std::move(call), worker_queue_, this, message);
}

void GrpcConnection::RegisterConnectivityMonitor() {
  connectivity_monitor_->AddCallback(
      [this](ConnectivityMonitor::NetworkStatus /*ignored*/) {
        // Calls may unregister themselves on finish, so make a protective copy.
        auto calls = active_calls_;
        for (GrpcCall* call : calls) {
          // This will trigger the observers.
          call->FinishAndNotify(Status{FirestoreErrorCode::Unavailable,
                                       "Network connectivity changed"});
        }
        // The old channel may hang for a long time trying to reestablish
        // connection before eventually failing. Note that gRPC Objective-C
        //