Client callback streaming

9 files changed, 860 insertions, 13 deletions

diff --git a/include/grpcpp/generic/generic_stub.h b/include/grpcpp/generic/generic_stub.h
index d509d9a520..ccbf8a0e55 100644
--- a/include/grpcpp/generic/generic_stub.h
+++ b/include/grpcpp/generic/generic_stub.h
@@ -24,6 +24,7 @@
 #include <grpcpp/support/async_stream.h>
 #include <grpcpp/support/async_unary_call.h>
 #include <grpcpp/support/byte_buffer.h>
+#include <grpcpp/support/client_callback.h>
 #include <grpcpp/support/status.h>
 
 namespace grpc {
@@ -76,6 +77,10 @@ class GenericStub final {
                    const ByteBuffer* request, ByteBuffer* response,
                    std::function<void(Status)> on_completion);
 
+    experimental::ClientCallbackReaderWriter<ByteBuffer, ByteBuffer>*
+    PrepareBidiStreamingCall(ClientContext* context, const grpc::string& method,
+                             experimental::ClientBidiReactor* reactor);
+
    private:
     GenericStub* stub_;
   };
diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h
index 51367cf550..f7a24204dc 100644
--- a/include/grpcpp/impl/codegen/callback_common.h
+++ b/include/grpcpp/impl/codegen/callback_common.h
@@ -145,18 +145,19 @@ class CallbackWithSuccessTag
   // or on a tag that has been Set before unless the tag has been cleared.
   void Set(grpc_call* call, std::function<void(bool)> f,
            CompletionQueueTag* ops) {
+    GPR_CODEGEN_ASSERT(call_ == nullptr);
+    g_core_codegen_interface->grpc_call_ref(call);
     call_ = call;
     func_ = std::move(f);
     ops_ = ops;
-    g_core_codegen_interface->grpc_call_ref(call);
     functor_run = &CallbackWithSuccessTag::StaticRun;
   }
 
   void Clear() {
     if (call_ != nullptr) {
-      func_ = nullptr;
       grpc_call* call = call_;
       call_ = nullptr;
+      func_ = nullptr;
       g_core_codegen_interface->grpc_call_unref(call);
     }
   }
@@ -182,10 +183,9 @@ class CallbackWithSuccessTag
   }
   void Run(bool ok) {
     void* ignored = ops_;
-    bool new_ok = ok;
     // Allow a "false" return value from FinalizeResult to silence the
     // callback, just as it silences a CQ tag in the async cases
-    bool do_callback = ops_->FinalizeResult(&ignored, &new_ok);
+    bool do_callback = ops_->FinalizeResult(&ignored, &ok);
     GPR_CODEGEN_ASSERT(ignored == ops_);
 
     if (do_callback) {
diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h
index 6ec1ffb8c7..728a7b9049 100644
--- a/include/grpcpp/impl/codegen/channel_interface.h
+++ b/include/grpcpp/impl/codegen/channel_interface.h
@@ -53,6 +53,12 @@ template <class W, class R>
 class ClientAsyncReaderWriterFactory;
 template <class R>
 class ClientAsyncResponseReaderFactory;
+template <class W, class R>
+class ClientCallbackReaderWriterFactory;
+template <class R>
+class ClientCallbackReaderFactory;
+template <class W>
+class ClientCallbackWriterFactory;
 class InterceptedChannel;
 }  // namespace internal
 
@@ -106,6 +112,12 @@ class ChannelInterface {
   friend class ::grpc::internal::ClientAsyncReaderWriterFactory;
   template <class R>
   friend class ::grpc::internal::ClientAsyncResponseReaderFactory;
+  template <class W, class R>
+  friend class ::grpc::internal::ClientCallbackReaderWriterFactory;
+  template <class R>
+  friend class ::grpc::internal::ClientCallbackReaderFactory;
+  template <class W>
+  friend class ::grpc::internal::ClientCallbackWriterFactory;
   template <class InputMessage, class OutputMessage>
   friend class ::grpc::internal::BlockingUnaryCallImpl;
   template <class InputMessage, class OutputMessage>
diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h
index 4baa819091..01a28a9e6f 100644
--- a/include/grpcpp/impl/codegen/client_callback.h
+++ b/include/grpcpp/impl/codegen/client_callback.h
@@ -22,6 +22,7 @@
 #include <functional>
 
 #include <grpcpp/impl/codegen/call.h>
+#include <grpcpp/impl/codegen/call_op_set.h>
 #include <grpcpp/impl/codegen/callback_common.h>
 #include <grpcpp/impl/codegen/channel_interface.h>
 #include <grpcpp/impl/codegen/config.h>
@@ -88,6 +89,510 @@ class CallbackUnaryCallImpl {
     call.PerformOps(ops);
   }
 };
+}  // namespace internal
+
+namespace experimental {
+
+// The user must implement this reactor interface with reactions to each event
+// type that gets called by the library. An empty reaction is provided by
+// default
+
+class ClientBidiReactor {
+ public:
+  virtual ~ClientBidiReactor() {}
+  virtual void OnDone(Status s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+  virtual void OnWritesDoneDone(bool ok) {}
+};
+
+class ClientReadReactor {
+ public:
+  virtual ~ClientReadReactor() {}
+  virtual void OnDone(Status s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+};
+
+class ClientWriteReactor {
+ public:
+  virtual ~ClientWriteReactor() {}
+  virtual void OnDone(Status s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+  virtual void OnWritesDoneDone(bool ok) {}
+};
+
+template <class Request, class Response>
+class ClientCallbackReaderWriter {
+ public:
+  virtual ~ClientCallbackReaderWriter() {}
+  virtual void StartCall() = 0;
+  void Write(const Request* req) { Write(req, WriteOptions()); }
+  virtual void Write(const Request* req, WriteOptions options) = 0;
+  void WriteLast(const Request* req, WriteOptions options) {
+    Write(req, options.set_last_message());
+  }
+  virtual void WritesDone() = 0;
+  virtual void Read(Response* resp) = 0;
+};
+
+template <class Response>
+class ClientCallbackReader {
+ public:
+  virtual ~ClientCallbackReader() {}
+  virtual void StartCall() = 0;
+  virtual void Read(Response* resp) = 0;
+};
+
+template <class Request>
+class ClientCallbackWriter {
+ public:
+  virtual ~ClientCallbackWriter() {}
+  virtual void StartCall() = 0;
+  void Write(const Request* req) { Write(req, WriteOptions()); }
+  virtual void Write(const Request* req, WriteOptions options) = 0;
+  void WriteLast(const Request* req, WriteOptions options) {
+    Write(req, options.set_last_message());
+  }
+  virtual void WritesDone() = 0;
+};
+
+}  // namespace experimental
+
+namespace internal {
+
+// Forward declare factory classes for friendship
+template <class Request, class Response>
+class ClientCallbackReaderWriterFactory;
+template <class Response>
+class ClientCallbackReaderFactory;
+template <class Request>
+class ClientCallbackWriterFactory;
+
+template <class Request, class Response>
+class ClientCallbackReaderWriterImpl
+    : public ::grpc::experimental::ClientCallbackReaderWriter<Request,
+                                                              Response> {
+ public:
+  // always allocated against a call arena, no memory free required
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(ClientCallbackReaderWriterImpl));
+  }
+
+  // This operator should never be called as the memory should be freed as part
+  // of the arena destruction. It only exists to provide a matching operator
+  // delete to the operator new so that some compilers will not complain (see
+  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
+  // there are no tests catching the compiler warning.
+  static void operator delete(void*, void*) { assert(0); }
+
+  void MaybeFinish() {
+    if (--callbacks_outstanding_ == 0) {
+      reactor_->OnDone(std::move(finish_status_));
+      auto* call = call_.call();
+      this->~ClientCallbackReaderWriterImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches
+    // 1. Send initial metadata (unless corked)/recv initial metadata
+    // 2. Recv trailing metadata, on_completion callback
+    callbacks_outstanding_ = 2;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_);
+    start_corked_ = context_->initial_metadata_corked_;
+    if (!start_corked_) {
+      start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+    }
+    start_ops_.RecvInitialMetadata(context_);
+    start_ops_.set_core_cq_tag(&start_tag_);
+    call_.PerformOps(&start_ops_);
+
+    finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
+                    &finish_ops_);
+    finish_ops_.ClientRecvStatus(context_, &finish_status_);
+    finish_ops_.set_core_cq_tag(&finish_tag_);
+    call_.PerformOps(&finish_ops_);
+
+    // Also set up the read and write tags so that they don't have to be set up
+    // each time
+    write_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnWriteDone(ok);
+                     MaybeFinish();
+                   },
+                   &write_ops_);
+    write_ops_.set_core_cq_tag(&write_tag_);
+
+    read_tag_.Set(call_.call(),
+                  [this](bool ok) {
+                    reactor_->OnReadDone(ok);
+                    MaybeFinish();
+                  },
+                  &read_ops_);
+    read_ops_.set_core_cq_tag(&read_tag_);
+  }
+
+  void Read(Response* msg) override {
+    read_ops_.RecvMessage(msg);
+    callbacks_outstanding_++;
+    call_.PerformOps(&read_ops_);
+  }
+
+  void Write(const Request* msg, WriteOptions options) override {
+    if (start_corked_) {
+      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+    // TODO(vjpai): don't assert
+    GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*msg).ok());
+
+    if (options.is_last_message()) {
+      options.set_buffer_hint();
+      write_ops_.ClientSendClose();
+    }
+    callbacks_outstanding_++;
+    call_.PerformOps(&write_ops_);
+  }
+  void WritesDone() override {
+    if (start_corked_) {
+      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                           context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+    writes_done_ops_.ClientSendClose();
+    writes_done_tag_.Set(call_.call(),
+                         [this](bool ok) {
+                           reactor_->OnWritesDoneDone(ok);
+                           MaybeFinish();
+                         },
+                         &writes_done_ops_);
+    writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
+    callbacks_outstanding_++;
+    call_.PerformOps(&writes_done_ops_);
+  }
+
+ private:
+  friend class ClientCallbackReaderWriterFactory<Request, Response>;
+
+  ClientCallbackReaderWriterImpl(
+      Call call, ClientContext* context,
+      ::grpc::experimental::ClientBidiReactor* reactor)
+      : context_(context), call_(call), reactor_(reactor) {}
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientBidiReactor* reactor_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpRecvInitialMetadata> start_ops_;
+  CallbackWithSuccessTag start_tag_;
+  bool start_corked_;
+
+  CallOpSet<CallOpClientRecvStatus> finish_ops_;
+  CallbackWithSuccessTag finish_tag_;
+  Status finish_status_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
+      write_ops_;
+  CallbackWithSuccessTag write_tag_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
+  CallbackWithSuccessTag writes_done_tag_;
+
+  CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<Response>> read_ops_;
+  CallbackWithSuccessTag read_tag_;
+
+  std::atomic_int callbacks_outstanding_;
+};
+
+template <class Request, class Response>
+class ClientCallbackReaderWriterFactory {
+ public:
+  static experimental::ClientCallbackReaderWriter<Request, Response>* Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context,
+      ::grpc::experimental::ClientBidiReactor* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    return new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call.call(), sizeof(ClientCallbackReaderWriterImpl<Request, Response>)))
+        ClientCallbackReaderWriterImpl<Request, Response>(call, context,
+                                                          reactor);
+  }
+};
+
+template <class Response>
+class ClientCallbackReaderImpl
+    : public ::grpc::experimental::ClientCallbackReader<Response> {
+ public:
+  // always allocated against a call arena, no memory free required
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(ClientCallbackReaderImpl));
+  }
+
+  // This operator should never be called as the memory should be freed as part
+  // of the arena destruction. It only exists to provide a matching operator
+  // delete to the operator new so that some compilers will not complain (see
+  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
+  // there are no tests catching the compiler warning.
+  static void operator delete(void*, void*) { assert(0); }
+
+  void MaybeFinish() {
+    if (--callbacks_outstanding_ == 0) {
+      reactor_->OnDone(std::move(finish_status_));
+      auto* call = call_.call();
+      this->~ClientCallbackReaderImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches
+    // 1. Send initial metadata (unless corked)/recv initial metadata
+    // 2. Recv trailing metadata, on_completion callback
+    callbacks_outstanding_ = 2;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_);
+    start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                   context_->initial_metadata_flags());
+    start_ops_.RecvInitialMetadata(context_);
+    start_ops_.set_core_cq_tag(&start_tag_);
+    call_.PerformOps(&start_ops_);
+
+    finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
+                    &finish_ops_);
+    finish_ops_.ClientRecvStatus(context_, &finish_status_);
+    finish_ops_.set_core_cq_tag(&finish_tag_);
+    call_.PerformOps(&finish_ops_);
+
+    // Also set up the read tag so it doesn't have to be set up each time
+    read_tag_.Set(call_.call(),
+                  [this](bool ok) {
+                    reactor_->OnReadDone(ok);
+                    MaybeFinish();
+                  },
+                  &read_ops_);
+    read_ops_.set_core_cq_tag(&read_tag_);
+  }
+
+  void Read(Response* msg) override {
+    read_ops_.RecvMessage(msg);
+    callbacks_outstanding_++;
+    call_.PerformOps(&read_ops_);
+  }
+
+ private:
+  friend class ClientCallbackReaderFactory<Response>;
+
+  template <class Request>
+  ClientCallbackReaderImpl(Call call, ClientContext* context, Request* request,
+                           ::grpc::experimental::ClientReadReactor* reactor)
+      : context_(context), call_(call), reactor_(reactor) {
+    // TODO(vjpai): don't assert
+    GPR_CODEGEN_ASSERT(start_ops_.SendMessage(*request).ok());
+    start_ops_.ClientSendClose();
+  }
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientReadReactor* reactor_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose,
+            CallOpRecvInitialMetadata>
+      start_ops_;
+  CallbackWithSuccessTag start_tag_;
+
+  CallOpSet<CallOpClientRecvStatus> finish_ops_;
+  CallbackWithSuccessTag finish_tag_;
+  Status finish_status_;
+
+  CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<Response>> read_ops_;
+  CallbackWithSuccessTag read_tag_;
+
+  std::atomic_int callbacks_outstanding_;
+};
+
+template <class Response>
+class ClientCallbackReaderFactory {
+ public:
+  template <class Request>
+  static experimental::ClientCallbackReader<Response>* Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context, const Request* request,
+      ::grpc::experimental::ClientReadReactor* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    return new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call.call(), sizeof(ClientCallbackReaderImpl<Response>)))
+        ClientCallbackReaderImpl<Response>(call, context, request, reactor);
+  }
+};
+
+template <class Request>
+class ClientCallbackWriterImpl
+    : public ::grpc::experimental::ClientCallbackWriter<Request> {
+ public:
+  // always allocated against a call arena, no memory free required
+  static void operator delete(void* ptr, std::size_t size) {
+    assert(size == sizeof(ClientCallbackWriterImpl));
+  }
+
+  // This operator should never be called as the memory should be freed as part
+  // of the arena destruction. It only exists to provide a matching operator
+  // delete to the operator new so that some compilers will not complain (see
+  // https://github.com/grpc/grpc/issues/11301) Note at the time of adding this
+  // there are no tests catching the compiler warning.
+  static void operator delete(void*, void*) { assert(0); }
+
+  void MaybeFinish() {
+    if (--callbacks_outstanding_ == 0) {
+      reactor_->OnDone(std::move(finish_status_));
+      auto* call = call_.call();
+      this->~ClientCallbackWriterImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches
+    // 1. Send initial metadata (unless corked)/recv initial metadata
+    // 2. Recv message + trailing metadata, on_completion callback
+    callbacks_outstanding_ = 2;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_);
+    start_corked_ = context_->initial_metadata_corked_;
+    if (!start_corked_) {
+      start_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+    }
+    start_ops_.RecvInitialMetadata(context_);
+    start_ops_.set_core_cq_tag(&start_tag_);
+    call_.PerformOps(&start_ops_);
+
+    finish_tag_.Set(call_.call(), [this](bool ok) { MaybeFinish(); },
+                    &finish_ops_);
+    finish_ops_.ClientRecvStatus(context_, &finish_status_);
+    finish_ops_.set_core_cq_tag(&finish_tag_);
+    call_.PerformOps(&finish_ops_);
+
+    // Also set up the read and write tags so that they don't have to be set up
+    // each time
+    write_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnWriteDone(ok);
+                     MaybeFinish();
+                   },
+                   &write_ops_);
+    write_ops_.set_core_cq_tag(&write_tag_);
+  }
+
+  void Write(const Request* msg, WriteOptions options) override {
+    if (start_corked_) {
+      write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                     context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+    // TODO(vjpai): don't assert
+    GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*msg).ok());
+
+    if (options.is_last_message()) {
+      options.set_buffer_hint();
+      write_ops_.ClientSendClose();
+    }
+    callbacks_outstanding_++;
+    call_.PerformOps(&write_ops_);
+  }
+  void WritesDone() override {
+    if (start_corked_) {
+      writes_done_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
+                                           context_->initial_metadata_flags());
+      start_corked_ = false;
+    }
+    writes_done_ops_.ClientSendClose();
+    writes_done_tag_.Set(call_.call(),
+                         [this](bool ok) {
+                           reactor_->OnWritesDoneDone(ok);
+                           MaybeFinish();
+                         },
+                         &writes_done_ops_);
+    writes_done_ops_.set_core_cq_tag(&writes_done_tag_);
+    callbacks_outstanding_++;
+    call_.PerformOps(&writes_done_ops_);
+  }
+
+ private:
+  friend class ClientCallbackWriterFactory<Request>;
+
+  template <class Response>
+  ClientCallbackWriterImpl(Call call, ClientContext* context,
+                           Response* response,
+                           ::grpc::experimental::ClientWriteReactor* reactor)
+      : context_(context), call_(call), reactor_(reactor) {
+    finish_ops_.RecvMessage(response);
+    finish_ops_.AllowNoMessage();
+  }
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientWriteReactor* reactor_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpRecvInitialMetadata> start_ops_;
+  CallbackWithSuccessTag start_tag_;
+  bool start_corked_;
+
+  CallOpSet<CallOpGenericRecvMessage, CallOpClientRecvStatus> finish_ops_;
+  CallbackWithSuccessTag finish_tag_;
+  Status finish_status_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose>
+      write_ops_;
+  CallbackWithSuccessTag write_tag_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
+  CallbackWithSuccessTag writes_done_tag_;
+
+  std::atomic_int callbacks_outstanding_;
+};
+
+template <class Request>
+class ClientCallbackWriterFactory {
+ public:
+  template <class Response>
+  static experimental::ClientCallbackWriter<Request>* Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context, Response* response,
+      ::grpc::experimental::ClientWriteReactor* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    return new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call.call(), sizeof(ClientCallbackWriterImpl<Request>)))
+        ClientCallbackWriterImpl<Request>(call, context, response, reactor);
+  }
+};
 
 }  // namespace internal
 }  // namespace grpc
diff --git a/include/grpcpp/impl/codegen/client_context.h b/include/grpcpp/impl/codegen/client_context.h
index 75b955e760..6059c3c58a 100644
--- a/include/grpcpp/impl/codegen/client_context.h
+++ b/include/grpcpp/impl/codegen/client_context.h
@@ -71,6 +71,12 @@ template <class InputMessage, class OutputMessage>
 class BlockingUnaryCallImpl;
 template <class InputMessage, class OutputMessage>
 class CallbackUnaryCallImpl;
+template <class Request, class Response>
+class ClientCallbackReaderWriterImpl;
+template <class Response>
+class ClientCallbackReaderImpl;
+template <class Request>
+class ClientCallbackWriterImpl;
 }  // namespace internal
 
 template <class R>
@@ -394,6 +400,12 @@ class ClientContext {
   friend class ::grpc::internal::BlockingUnaryCallImpl;
   template <class InputMessage, class OutputMessage>
   friend class ::grpc::internal::CallbackUnaryCallImpl;
+  template <class Request, class Response>
+  friend class ::grpc::internal::ClientCallbackReaderWriterImpl;
+  template <class Response>
+  friend class ::grpc::internal::ClientCallbackReaderImpl;
+  template <class Request>
+  friend class ::grpc::internal::ClientCallbackWriterImpl;
 
   // Used by friend class CallOpClientRecvStatus
   void set_debug_error_string(const grpc::string& debug_error_string) {
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index 7986aca696..473e57166f 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -132,6 +132,7 @@ grpc::string GetHeaderIncludes(grpc_generator::File* file,
         "grpcpp/impl/codegen/async_generic_service.h",
         "grpcpp/impl/codegen/async_stream.h",
         "grpcpp/impl/codegen/async_unary_call.h",
+        "grpcpp/impl/codegen/client_callback.h",
         "grpcpp/impl/codegen/method_handler_impl.h",
         "grpcpp/impl/codegen/proto_utils.h",
         "grpcpp/impl/codegen/rpc_method.h",
@@ -580,11 +581,23 @@ void PrintHeaderClientMethodCallbackInterfaces(
                    "const $Request$* request, $Response$* response, "
                    "std::function<void(::grpc::Status)>) = 0;\n");
   } else if (ClientOnlyStreaming(method)) {
-    // TODO(vjpai): Add support for client-side streaming
+    printer->Print(*vars,
+                   "virtual ::grpc::experimental::ClientCallbackWriter< "
+                   "$Request$>* $Method$(::grpc::ClientContext* context, "
+                   "$Response$* response, "
+                   "::grpc::experimental::ClientWriteReactor* reactor) = 0;\n");
   } else if (ServerOnlyStreaming(method)) {
-    // TODO(vjpai): Add support for server-side streaming
+    printer->Print(*vars,
+                   "virtual ::grpc::experimental::ClientCallbackReader< "
+                   "$Response$>* $Method$(::grpc::ClientContext* context, "
+                   "$Request$* request, "
+                   "::grpc::experimental::ClientReadReactor* reactor) = 0;\n");
   } else if (method->BidiStreaming()) {
-    // TODO(vjpai): Add support for bidi streaming
+    printer->Print(
+        *vars,
+        "virtual ::grpc::experimental::ClientCallbackReaderWriter< $Request$, "
+        "$Response$>* $Method$(::grpc::ClientContext* context, "
+        "::grpc::experimental::ClientBidiReactor* reactor) = 0;\n");
   }
 }
 
@@ -631,11 +644,26 @@ void PrintHeaderClientMethodCallback(grpc_generator::Printer* printer,
                    "const $Request$* request, $Response$* response, "
                    "std::function<void(::grpc::Status)>) override;\n");
   } else if (ClientOnlyStreaming(method)) {
-    // TODO(vjpai): Add support for client-side streaming
+    printer->Print(
+        *vars,
+        "::grpc::experimental::ClientCallbackWriter< $Request$>* "
+        "$Method$(::grpc::ClientContext* context, "
+        "$Response$* response, "
+        "::grpc::experimental::ClientWriteReactor* reactor) override;\n");
   } else if (ServerOnlyStreaming(method)) {
-    // TODO(vjpai): Add support for server-side streaming
+    printer->Print(
+        *vars,
+        "::grpc::experimental::ClientCallbackReader< $Response$>* "
+        "$Method$(::grpc::ClientContext* context, "
+        "$Request$* request, "
+        "::grpc::experimental::ClientReadReactor* reactor) override;\n");
+
   } else if (method->BidiStreaming()) {
-    // TODO(vjpai): Add support for bidi streaming
+    printer->Print(
+        *vars,
+        "::grpc::experimental::ClientCallbackReaderWriter< $Request$, "
+        "$Response$>* $Method$(::grpc::ClientContext* context, "
+        "::grpc::experimental::ClientBidiReactor* reactor) override;\n");
   }
 }
 
@@ -1607,7 +1635,20 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
         "context, response);\n"
         "}\n\n");
 
-    // TODO(vjpai): Add callback version
+    printer->Print(
+        *vars,
+        "::grpc::experimental::ClientCallbackWriter< $Request$>* "
+        "$ns$$Service$::"
+        "Stub::experimental_async::$Method$(::grpc::ClientContext* context, "
+        "$Response$* response, "
+        "::grpc::experimental::ClientWriteReactor* reactor) {\n");
+    printer->Print(*vars,
+                   "  return ::grpc::internal::ClientCallbackWriterFactory< "
+                   "$Request$>::Create("
+                   "stub_->channel_.get(), "
+                   "stub_->rpcmethod_$Method$_, "
+                   "context, response, reactor);\n"
+                   "}\n\n");
 
     for (auto async_prefix : async_prefixes) {
       (*vars)["AsyncPrefix"] = async_prefix.prefix;
@@ -1641,7 +1682,19 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
         "context, request);\n"
         "}\n\n");
 
-    // TODO(vjpai): Add callback version
+    printer->Print(*vars,
+                   "::grpc::experimental::ClientCallbackReader< $Response$>* "
+                   "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::"
+                   "ClientContext* context, "
+                   "$Request$* request, "
+                   "::grpc::experimental::ClientReadReactor* reactor) {\n");
+    printer->Print(*vars,
+                   "  return ::grpc::internal::ClientCallbackReaderFactory< "
+                   "$Response$>::Create("
+                   "stub_->channel_.get(), "
+                   "stub_->rpcmethod_$Method$_, "
+                   "context, request, reactor);\n"
+                   "}\n\n");
 
     for (auto async_prefix : async_prefixes) {
       (*vars)["AsyncPrefix"] = async_prefix.prefix;
@@ -1675,7 +1728,20 @@ void PrintSourceClientMethod(grpc_generator::Printer* printer,
                    "context);\n"
                    "}\n\n");
 
-    // TODO(vjpai): Add callback version
+    printer->Print(*vars,
+                   "::grpc::experimental::ClientCallbackReaderWriter< "
+                   "$Request$,$Response$>* "
+                   "$ns$$Service$::Stub::experimental_async::$Method$(::grpc::"
+                   "ClientContext* context, "
+                   "::grpc::experimental::ClientBidiReactor* reactor) {\n");
+    printer->Print(
+        *vars,
+        "  return ::grpc::internal::ClientCallbackReaderWriterFactory< "
+        "$Request$,$Response$>::Create("
+        "stub_->channel_.get(), "
+        "stub_->rpcmethod_$Method$_, "
+        "context, reactor);\n"
+        "}\n\n");
 
     for (auto async_prefix : async_prefixes) {
       (*vars)["AsyncPrefix"] = async_prefix.prefix;
diff --git a/src/cpp/client/generic_stub.cc b/src/cpp/client/generic_stub.cc
index 87902b26f0..f029daec65 100644
--- a/src/cpp/client/generic_stub.cc
+++ b/src/cpp/client/generic_stub.cc
@@ -72,4 +72,16 @@ void GenericStub::experimental_type::UnaryCall(
       context, request, response, std::move(on_completion));
 }
 
+experimental::ClientCallbackReaderWriter<ByteBuffer, ByteBuffer>*
+GenericStub::experimental_type::PrepareBidiStreamingCall(
+    ClientContext* context, const grpc::string& method,
+    experimental::ClientBidiReactor* reactor) {
+  return internal::ClientCallbackReaderWriterFactory<
+      ByteBuffer, ByteBuffer>::Create(stub_->channel_.get(),
+                                      internal::RpcMethod(
+                                          method.c_str(),
+                                          internal::RpcMethod::BIDI_STREAMING),
+                                      context, reactor);
+}
+
 }  // namespace grpc
diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden
index fdc67969d9..7a25f51d10 100644
--- a/test/cpp/codegen/compiler_test_golden
+++ b/test/cpp/codegen/compiler_test_golden
@@ -30,6 +30,7 @@
 #include <grpcpp/impl/codegen/async_generic_service.h>
 #include <grpcpp/impl/codegen/async_stream.h>
 #include <grpcpp/impl/codegen/async_unary_call.h>
+#include <grpcpp/impl/codegen/client_callback.h>
 #include <grpcpp/impl/codegen/method_handler_impl.h>
 #include <grpcpp/impl/codegen/proto_utils.h>
 #include <grpcpp/impl/codegen/rpc_method.h>
@@ -117,10 +118,13 @@ class ServiceA final {
       //
       // Method A2 leading comment 1
       // Method A2 leading comment 2
+      virtual ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) = 0;
       // MethodA2 trailing comment 1
       // Method A3 leading comment 1
+      virtual ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) = 0;
       // Method A3 trailing comment 1
       // Method A4 leading comment 1
+      virtual ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) = 0;
       // Method A4 trailing comment 1
     };
     virtual class experimental_async_interface* experimental_async() { return nullptr; }
@@ -178,6 +182,9 @@ class ServiceA final {
       public StubInterface::experimental_async_interface {
      public:
       void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override;
+      ::grpc::experimental::ClientCallbackWriter< ::grpc::testing::Request>* MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor* reactor) override;
+      ::grpc::experimental::ClientCallbackReader< ::grpc::testing::Response>* MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor* reactor) override;
+      ::grpc::experimental::ClientCallbackReaderWriter< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor* reactor) override;
      private:
       friend class Stub;
       explicit experimental_async(Stub* stub): stub_(stub) { }
diff --git a/test/cpp/end2end/client_callback_end2end_test.cc b/test/cpp/end2end/client_callback_end2end_test.cc
index a35991396a..e39fc5aab2 100644
--- a/test/cpp/end2end/client_callback_end2end_test.cc
+++ b/test/cpp/end2end/client_callback_end2end_test.cc
@@ -182,6 +182,59 @@ class ClientCallbackEnd2endTest
     }
   }
 
+  void SendGenericEchoAsBidi(int num_rpcs) {
+    const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo");
+    grpc::string test_string("");
+    for (int i = 0; i < num_rpcs; i++) {
+      test_string += "Hello world. ";
+      class Client : public grpc::experimental::ClientBidiReactor {
+       public:
+        Client(ClientCallbackEnd2endTest* test, const grpc::string& method_name,
+               const grpc::string& test_str) {
+          stream_ =
+              test->generic_stub_->experimental().PrepareBidiStreamingCall(
+                  &cli_ctx_, method_name, this);
+          stream_->StartCall();
+          request_.set_message(test_str);
+          send_buf_ = SerializeToByteBuffer(&request_);
+          stream_->Read(&recv_buf_);
+          stream_->Write(send_buf_.get());
+        }
+        void OnWriteDone(bool ok) override { stream_->WritesDone(); }
+        void OnReadDone(bool ok) override {
+          EchoResponse response;
+          EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response));
+          EXPECT_EQ(request_.message(), response.message());
+        };
+        void OnDone(Status s) override {
+          // The stream is invalid once OnDone is called
+          stream_ = nullptr;
+          EXPECT_TRUE(s.ok());
+          std::unique_lock<std::mutex> l(mu_);
+          done_ = true;
+          cv_.notify_one();
+        }
+        void Await() {
+          std::unique_lock<std::mutex> l(mu_);
+          while (!done_) {
+            cv_.wait(l);
+          }
+        }
+
+        EchoRequest request_;
+        std::unique_ptr<ByteBuffer> send_buf_;
+        ByteBuffer recv_buf_;
+        ClientContext cli_ctx_;
+        experimental::ClientCallbackReaderWriter<ByteBuffer, ByteBuffer>*
+            stream_;
+        std::mutex mu_;
+        std::condition_variable cv_;
+        bool done_ = false;
+      } rpc{this, kMethodName, test_string};
+
+      rpc.Await();
+    }
+  }
   bool is_server_started_;
   std::shared_ptr<Channel> channel_;
   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
@@ -211,6 +264,11 @@ TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcs) {
   SendRpcsGeneric(10, false);
 }
 
+TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidi) {
+  ResetStub();
+  SendGenericEchoAsBidi(10);
+}
+
 #if GRPC_ALLOW_EXCEPTIONS
 TEST_P(ClientCallbackEnd2endTest, ExceptingRpc) {
   ResetStub();
@@ -267,6 +325,176 @@ TEST_P(ClientCallbackEnd2endTest, CancelRpcBeforeStart) {
   }
 }
 
+TEST_P(ClientCallbackEnd2endTest, RequestStream) {
+  // TODO(vjpai): test with callback server once supported
+  if (GetParam().callback_server) {
+    return;
+  }
+
+  ResetStub();
+  class Client : public grpc::experimental::ClientWriteReactor {
+   public:
+    explicit Client(grpc::testing::EchoTestService::Stub* stub) {
+      context_.set_initial_metadata_corked(true);
+      stream_ = stub->experimental_async()->RequestStream(&context_, &response_,
+                                                          this);
+      stream_->StartCall();
+      request_.set_message("Hello server.");
+      stream_->Write(&request_);
+    }
+    void OnWriteDone(bool ok) override {
+      writes_left_--;
+      if (writes_left_ > 1) {
+        stream_->Write(&request_);
+      } else if (writes_left_ == 1) {
+        stream_->WriteLast(&request_, WriteOptions());
+      }
+    }
+    void OnDone(Status s) override {
+      stream_ = nullptr;
+      EXPECT_TRUE(s.ok());
+      EXPECT_EQ(response_.message(), "Hello server.Hello server.Hello server.");
+      std::unique_lock<std::mutex> l(mu_);
+      done_ = true;
+      cv_.notify_one();
+    }
+    void Await() {
+      std::unique_lock<std::mutex> l(mu_);
+      while (!done_) {
+        cv_.wait(l);
+      }
+    }
+
+   private:
+    ::grpc::experimental::ClientCallbackWriter<EchoRequest>* stream_;
+    EchoRequest request_;
+    EchoResponse response_;
+    ClientContext context_;
+    int writes_left_{3};
+    std::mutex mu_;
+    std::condition_variable cv_;
+    bool done_ = false;
+  } test{stub_.get()};
+
+  test.Await();
+}
+
+TEST_P(ClientCallbackEnd2endTest, ResponseStream) {
+  // TODO(vjpai): test with callback server once supported
+  if (GetParam().callback_server) {
+    return;
+  }
+
+  ResetStub();
+  class Client : public grpc::experimental::ClientReadReactor {
+   public:
+    explicit Client(grpc::testing::EchoTestService::Stub* stub) {
+      request_.set_message("Hello client ");
+      stream_ = stub->experimental_async()->ResponseStream(&context_, &request_,
+                                                           this);
+      stream_->StartCall();
+      stream_->Read(&response_);
+    }
+    void OnReadDone(bool ok) override {
+      // Note that != is the boolean XOR operator
+      EXPECT_NE(ok, reads_complete_ == kServerDefaultResponseStreamsToSend);
+      if (ok) {
+        EXPECT_EQ(response_.message(),
+                  request_.message() + grpc::to_string(reads_complete_));
+        reads_complete_++;
+        stream_->Read(&response_);
+      }
+    }
+    void OnDone(Status s) override {
+      stream_ = nullptr;
+      EXPECT_TRUE(s.ok());
+      std::unique_lock<std::mutex> l(mu_);
+      done_ = true;
+      cv_.notify_one();
+    }
+    void Await() {
+      std::unique_lock<std::mutex> l(mu_);
+      while (!done_) {
+        cv_.wait(l);
+      }
+    }
+
+   private:
+    ::grpc::experimental::ClientCallbackReader<EchoResponse>* stream_;
+    EchoRequest request_;
+    EchoResponse response_;
+    ClientContext context_;
+    int reads_complete_{0};
+    std::mutex mu_;
+    std::condition_variable cv_;
+    bool done_ = false;
+  } test{stub_.get()};
+
+  test.Await();
+}
+
+TEST_P(ClientCallbackEnd2endTest, BidiStream) {
+  // TODO(vjpai): test with callback server once supported
+  if (GetParam().callback_server) {
+    return;
+  }
+  ResetStub();
+  class Client : public grpc::experimental::ClientBidiReactor {
+   public:
+    explicit Client(grpc::testing::EchoTestService::Stub* stub) {
+      request_.set_message("Hello fren ");
+      stream_ = stub->experimental_async()->BidiStream(&context_, this);
+      stream_->StartCall();
+      stream_->Read(&response_);
+      stream_->Write(&request_);
+    }
+    void OnReadDone(bool ok) override {
+      // Note that != is the boolean XOR operator
+      EXPECT_NE(ok, reads_complete_ == kServerDefaultResponseStreamsToSend);
+      if (ok) {
+        EXPECT_EQ(response_.message(), request_.message());
+        reads_complete_++;
+        stream_->Read(&response_);
+      }
+    }
+    void OnWriteDone(bool ok) override {
+      EXPECT_TRUE(ok);
+      if (++writes_complete_ == kServerDefaultResponseStreamsToSend) {
+        stream_->WritesDone();
+      } else {
+        stream_->Write(&request_);
+      }
+    }
+    void OnDone(Status s) override {
+      stream_ = nullptr;
+      EXPECT_TRUE(s.ok());
+      std::unique_lock<std::mutex> l(mu_);
+      done_ = true;
+      cv_.notify_one();
+    }
+    void Await() {
+      std::unique_lock<std::mutex> l(mu_);
+      while (!done_) {
+        cv_.wait(l);
+      }
+    }
+
+   private:
+    ::grpc::experimental::ClientCallbackReaderWriter<EchoRequest, EchoResponse>*
+        stream_;
+    EchoRequest request_;
+    EchoResponse response_;
+    ClientContext context_;
+    int reads_complete_{0};
+    int writes_complete_{0};
+    std::mutex mu_;
+    std::condition_variable cv_;
+    bool done_ = false;
+  } test{stub_.get()};
+
+  test.Await();
+}
+
 TestScenario scenarios[] = {TestScenario{false}, TestScenario{true}};
 
 INSTANTIATE_TEST_CASE_P(ClientCallbackEnd2endTest, ClientCallbackEnd2endTest,