Merge branch 'master' into config-isolation

40 files changed, 4092 insertions, 846 deletions

diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 514d9ddc4b..d3b74cabab 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -506,6 +506,9 @@ GRPCAPI char* grpc_channelz_get_top_channels(intptr_t start_channel_id);
 /* Gets all servers that exist in the process. */
 GRPCAPI char* grpc_channelz_get_servers(intptr_t start_server_id);
 
+/* Returns a single Server, or else a NOT_FOUND code. */
+GRPCAPI char* grpc_channelz_get_server(intptr_t server_id);
+
 /* Gets all server sockets that exist in the server. */
 GRPCAPI char* grpc_channelz_get_server_sockets(intptr_t server_id,
                                                intptr_t start_socket_id);
diff --git a/include/grpc/impl/codegen/atm_gcc_sync.h b/include/grpc/impl/codegen/atm_gcc_sync.h
index c0010a3469..728c3d5412 100644
--- a/include/grpc/impl/codegen/atm_gcc_sync.h
+++ b/include/grpc/impl/codegen/atm_gcc_sync.h
@@ -26,6 +26,8 @@
 typedef intptr_t gpr_atm;
 #define GPR_ATM_MAX INTPTR_MAX
 #define GPR_ATM_MIN INTPTR_MIN
+#define GPR_ATM_INC_CAS_THEN(blah) blah
+#define GPR_ATM_INC_ADD_THEN(blah) blah
 
 #define GPR_ATM_COMPILE_BARRIER_() __asm__ __volatile__("" : : : "memory")
 
diff --git a/include/grpc/impl/codegen/atm_windows.h b/include/grpc/impl/codegen/atm_windows.h
index f6b27e5df7..c016b90095 100644
--- a/include/grpc/impl/codegen/atm_windows.h
+++ b/include/grpc/impl/codegen/atm_windows.h
@@ -25,6 +25,8 @@
 typedef intptr_t gpr_atm;
 #define GPR_ATM_MAX INTPTR_MAX
 #define GPR_ATM_MIN INTPTR_MIN
+#define GPR_ATM_INC_CAS_THEN(blah) blah
+#define GPR_ATM_INC_ADD_THEN(blah) blah
 
 #define gpr_atm_full_barrier MemoryBarrier
 
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index 402e490e34..610548b5f3 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -293,7 +293,7 @@ typedef struct {
   "grpc.max_channel_trace_event_memory_per_node"
 /** If non-zero, gRPC library will track stats and information at at per channel
  * level. Disabling channelz naturally disables channel tracing. The default
- * is for channelz to be disabled. */
+ * is for channelz to be enabled. */
 #define GRPC_ARG_ENABLE_CHANNELZ "grpc.enable_channelz"
 /** If non-zero, Cronet transport will coalesce packets to fewer frames
  * when possible. */
@@ -350,6 +350,11 @@ typedef struct {
 /** If set, inhibits health checking (which may be enabled via the
  *  service config.) */
 #define GRPC_ARG_INHIBIT_HEALTH_CHECKING "grpc.inhibit_health_checking"
+/** If set, determines the number of milliseconds that the c-ares based
+ * DNS resolver will wait on queries before cancelling them. The default value
+ * is 10000. Setting this to "0" will disable c-ares query timeouts
+ * entirely. */
+#define GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS "grpc.dns_ares_query_timeout"
 /** gRPC Objective-C channel pooling domain string. */
 #define GRPC_ARG_CHANNEL_POOL_DOMAIN "grpc.channel_pooling_domain"
 /** gRPC Objective-C channel pooling id. */
diff --git a/include/grpc/impl/codegen/port_platform.h b/include/grpc/impl/codegen/port_platform.h
index b2028a6305..031c0c36ae 100644
--- a/include/grpc/impl/codegen/port_platform.h
+++ b/include/grpc/impl/codegen/port_platform.h
@@ -526,6 +526,15 @@ typedef unsigned __int64 uint64_t;
 #endif /* GPR_ATTRIBUTE_NO_TSAN (2) */
 #endif /* GPR_ATTRIBUTE_NO_TSAN (1) */
 
+/* GRPC_TSAN_ENABLED will be defined, when compiled with thread sanitizer. */
+#if defined(__SANITIZE_THREAD__)
+#define GRPC_TSAN_ENABLED
+#elif defined(__has_feature)
+#if __has_feature(thread_sanitizer)
+#define GRPC_TSAN_ENABLED
+#endif
+#endif
+
 /* GRPC_ALLOW_EXCEPTIONS should be 0 or 1 if exceptions are allowed or not */
 #ifndef GRPC_ALLOW_EXCEPTIONS
 /* If not already set, set to 1 on Windows (style guide standard) but to
diff --git a/include/grpcpp/channel.h b/include/grpcpp/channel.h
index b7c9e354de..ee83396069 100644
--- a/include/grpcpp/channel.h
+++ b/include/grpcpp/channel.h
@@ -20,6 +20,7 @@
 #define GRPCPP_CHANNEL_H
 
 #include <memory>
+#include <mutex>
 
 #include <grpc/grpc.h>
 #include <grpcpp/impl/call.h>
@@ -64,12 +65,13 @@ class Channel final : public ChannelInterface,
   friend void experimental::ChannelResetConnectionBackoff(Channel* channel);
   friend std::shared_ptr<Channel> CreateChannelInternal(
       const grpc::string& host, grpc_channel* c_channel,
-      std::unique_ptr<std::vector<
-          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+      std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
           interceptor_creators);
+  friend class internal::InterceptedChannel;
   Channel(const grpc::string& host, grpc_channel* c_channel,
-          std::unique_ptr<std::vector<
-              std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+          std::vector<
+              std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
               interceptor_creators);
 
   internal::Call CreateCall(const internal::RpcMethod& method,
@@ -87,6 +89,10 @@ class Channel final : public ChannelInterface,
 
   CompletionQueue* CallbackCQ() override;
 
+  internal::Call CreateCallInternal(const internal::RpcMethod& method,
+                                    ClientContext* context, CompletionQueue* cq,
+                                    size_t interceptor_pos) override;
+
   const grpc::string host_;
   grpc_channel* const c_channel_;  // owned
 
diff --git a/include/grpcpp/create_channel.h b/include/grpcpp/create_channel.h
index 43188d09e7..e8a2a70581 100644
--- a/include/grpcpp/create_channel.h
+++ b/include/grpcpp/create_channel.h
@@ -70,8 +70,8 @@ std::shared_ptr<Channel> CreateCustomChannelWithInterceptors(
     const grpc::string& target,
     const std::shared_ptr<ChannelCredentials>& creds,
     const ChannelArguments& args,
-    std::unique_ptr<std::vector<
-        std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+    std::vector<
+        std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
         interceptor_creators);
 }  // namespace experimental
 }  // namespace grpc
diff --git a/include/grpcpp/generic/generic_stub.h b/include/grpcpp/generic/generic_stub.h
index d509d9a520..eb014184e4 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);
 
+    void PrepareBidiStreamingCall(
+        ClientContext* context, const grpc::string& method,
+        experimental::ClientBidiReactor<ByteBuffer, ByteBuffer>* reactor);
+
    private:
     GenericStub* stub_;
   };
diff --git a/include/grpcpp/health_check_service_interface.h b/include/grpcpp/health_check_service_interface.h
index b45a699bda..dfd4c3983a 100644
--- a/include/grpcpp/health_check_service_interface.h
+++ b/include/grpcpp/health_check_service_interface.h
@@ -37,6 +37,10 @@ class HealthCheckServiceInterface {
                                 bool serving) = 0;
   /// Apply to all registered service names.
   virtual void SetServingStatus(bool serving) = 0;
+
+  /// Set all registered service names to not serving and prevent future
+  /// state changes.
+  virtual void Shutdown() {}
 };
 
 /// Enable/disable the default health checking service. This applies to all C++
diff --git a/include/grpcpp/impl/codegen/async_stream.h b/include/grpcpp/impl/codegen/async_stream.h
index 6e58fd0eef..bfb2df4f23 100644
--- a/include/grpcpp/impl/codegen/async_stream.h
+++ b/include/grpcpp/impl/codegen/async_stream.h
@@ -276,7 +276,7 @@ class ClientAsyncReader final : public ClientAsyncReaderInterface<R> {
   }
 
   void StartCallInternal(void* tag) {
-    init_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+    init_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
                                   context_->initial_metadata_flags());
     init_ops_.set_output_tag(tag);
     call_.PerformOps(&init_ops_);
@@ -441,7 +441,7 @@ class ClientAsyncWriter final : public ClientAsyncWriterInterface<W> {
   }
 
   void StartCallInternal(void* tag) {
-    write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+    write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
                                    context_->initial_metadata_flags());
     // if corked bit is set in context, we just keep the initial metadata
     // buffered up to coalesce with later message send. No op is performed.
@@ -612,7 +612,7 @@ class ClientAsyncReaderWriter final
   }
 
   void StartCallInternal(void* tag) {
-    write_ops_.SendInitialMetadata(context_->send_initial_metadata_,
+    write_ops_.SendInitialMetadata(&context_->send_initial_metadata_,
                                    context_->initial_metadata_flags());
     // if corked bit is set in context, we just keep the initial metadata
     // buffered up to coalesce with later message send. No op is performed.
@@ -710,7 +710,7 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     meta_ops_.set_output_tag(tag);
-    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+    meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                   ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       meta_ops_.set_compression_level(ctx_->compression_level());
@@ -739,7 +739,7 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
   void Finish(const W& msg, const Status& status, void* tag) override {
     finish_ops_.set_output_tag(tag);
     if (!ctx_->sent_initial_metadata_) {
-      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+      finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                       ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         finish_ops_.set_compression_level(ctx_->compression_level());
@@ -748,10 +748,10 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
     }
     // The response is dropped if the status is not OK.
     if (status.ok()) {
-      finish_ops_.ServerSendStatus(ctx_->trailing_metadata_,
+      finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_,
                                    finish_ops_.SendMessage(msg));
     } else {
-      finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+      finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     }
     call_.PerformOps(&finish_ops_);
   }
@@ -769,14 +769,14 @@ class ServerAsyncReader final : public ServerAsyncReaderInterface<W, R> {
     GPR_CODEGEN_ASSERT(!status.ok());
     finish_ops_.set_output_tag(tag);
     if (!ctx_->sent_initial_metadata_) {
-      finish_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+      finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                       ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         finish_ops_.set_compression_level(ctx_->compression_level());
       }
       ctx_->sent_initial_metadata_ = true;
     }
-    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_ops_);
   }
 
@@ -859,7 +859,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     meta_ops_.set_output_tag(tag);
-    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+    meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                   ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       meta_ops_.set_compression_level(ctx_->compression_level());
@@ -904,7 +904,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
     EnsureInitialMetadataSent(&write_ops_);
     options.set_buffer_hint();
     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
-    write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    write_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&write_ops_);
   }
 
@@ -922,7 +922,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
   void Finish(const Status& status, void* tag) override {
     finish_ops_.set_output_tag(tag);
     EnsureInitialMetadataSent(&finish_ops_);
-    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_ops_);
   }
 
@@ -932,7 +932,7 @@ class ServerAsyncWriter final : public ServerAsyncWriterInterface<W> {
   template <class T>
   void EnsureInitialMetadataSent(T* ops) {
     if (!ctx_->sent_initial_metadata_) {
-      ops->SendInitialMetadata(ctx_->initial_metadata_,
+      ops->SendInitialMetadata(&ctx_->initial_metadata_,
                                ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         ops->set_compression_level(ctx_->compression_level());
@@ -1025,7 +1025,7 @@ class ServerAsyncReaderWriter final
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     meta_ops_.set_output_tag(tag);
-    meta_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+    meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                   ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       meta_ops_.set_compression_level(ctx_->compression_level());
@@ -1075,7 +1075,7 @@ class ServerAsyncReaderWriter final
     EnsureInitialMetadataSent(&write_ops_);
     options.set_buffer_hint();
     GPR_CODEGEN_ASSERT(write_ops_.SendMessage(msg, options).ok());
-    write_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    write_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&write_ops_);
   }
 
@@ -1094,7 +1094,7 @@ class ServerAsyncReaderWriter final
     finish_ops_.set_output_tag(tag);
     EnsureInitialMetadataSent(&finish_ops_);
 
-    finish_ops_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_ops_);
   }
 
@@ -1106,7 +1106,7 @@ class ServerAsyncReaderWriter final
   template <class T>
   void EnsureInitialMetadataSent(T* ops) {
     if (!ctx_->sent_initial_metadata_) {
-      ops->SendInitialMetadata(ctx_->initial_metadata_,
+      ops->SendInitialMetadata(&ctx_->initial_metadata_,
                                ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         ops->set_compression_level(ctx_->compression_level());
diff --git a/include/grpcpp/impl/codegen/async_unary_call.h b/include/grpcpp/impl/codegen/async_unary_call.h
index 60ff8e2f05..89dcb12418 100644
--- a/include/grpcpp/impl/codegen/async_unary_call.h
+++ b/include/grpcpp/impl/codegen/async_unary_call.h
@@ -174,7 +174,7 @@ class ClientAsyncResponseReader final
   }
 
   void StartCallInternal() {
-    single_buf.SendInitialMetadata(context_->send_initial_metadata_,
+    single_buf.SendInitialMetadata(&context_->send_initial_metadata_,
                                    context_->initial_metadata_flags());
   }
 
@@ -214,7 +214,7 @@ class ServerAsyncResponseWriter final
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     meta_buf_.set_output_tag(tag);
-    meta_buf_.SendInitialMetadata(ctx_->initial_metadata_,
+    meta_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
                                   ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       meta_buf_.set_compression_level(ctx_->compression_level());
@@ -240,8 +240,9 @@ class ServerAsyncResponseWriter final
   /// metadata.
   void Finish(const W& msg, const Status& status, void* tag) {
     finish_buf_.set_output_tag(tag);
+    finish_buf_.set_core_cq_tag(&finish_buf_);
     if (!ctx_->sent_initial_metadata_) {
-      finish_buf_.SendInitialMetadata(ctx_->initial_metadata_,
+      finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
                                       ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         finish_buf_.set_compression_level(ctx_->compression_level());
@@ -250,10 +251,10 @@ class ServerAsyncResponseWriter final
     }
     // The response is dropped if the status is not OK.
     if (status.ok()) {
-      finish_buf_.ServerSendStatus(ctx_->trailing_metadata_,
+      finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_,
                                    finish_buf_.SendMessage(msg));
     } else {
-      finish_buf_.ServerSendStatus(ctx_->trailing_metadata_, status);
+      finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     }
     call_.PerformOps(&finish_buf_);
   }
@@ -274,14 +275,14 @@ class ServerAsyncResponseWriter final
     GPR_CODEGEN_ASSERT(!status.ok());
     finish_buf_.set_output_tag(tag);
     if (!ctx_->sent_initial_metadata_) {
-      finish_buf_.SendInitialMetadata(ctx_->initial_metadata_,
+      finish_buf_.SendInitialMetadata(&ctx_->initial_metadata_,
                                       ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         finish_buf_.set_compression_level(ctx_->compression_level());
       }
       ctx_->sent_initial_metadata_ = true;
     }
-    finish_buf_.ServerSendStatus(ctx_->trailing_metadata_, status);
+    finish_buf_.ServerSendStatus(&ctx_->trailing_metadata_, status);
     call_.PerformOps(&finish_buf_);
   }
 
diff --git a/include/grpcpp/impl/codegen/byte_buffer.h b/include/grpcpp/impl/codegen/byte_buffer.h
index 8cc5158115..53ecb53371 100644
--- a/include/grpcpp/impl/codegen/byte_buffer.h
+++ b/include/grpcpp/impl/codegen/byte_buffer.h
@@ -45,11 +45,20 @@ template <class ServiceType, class RequestType, class ResponseType>
 class RpcMethodHandler;
 template <class ServiceType, class RequestType, class ResponseType>
 class ServerStreamingHandler;
+template <class RequestType, class ResponseType>
+class CallbackUnaryHandler;
+template <class RequestType, class ResponseType>
+class CallbackServerStreamingHandler;
 template <StatusCode code>
 class ErrorMethodHandler;
 template <class R>
 class DeserializeFuncType;
 class GrpcByteBufferPeer;
+template <class ServiceType, class RequestType, class ResponseType>
+class RpcMethodHandler;
+template <class ServiceType, class RequestType, class ResponseType>
+class ServerStreamingHandler;
+
 }  // namespace internal
 /// A sequence of bytes.
 class ByteBuffer final {
@@ -141,11 +150,18 @@ class ByteBuffer final {
   template <class R>
   friend class internal::CallOpRecvMessage;
   friend class internal::CallOpGenericRecvMessage;
-  friend class internal::MethodHandler;
+  template <class ServiceType, class RequestType, class ResponseType>
+  friend class RpcMethodHandler;
+  template <class ServiceType, class RequestType, class ResponseType>
+  friend class ServerStreamingHandler;
   template <class ServiceType, class RequestType, class ResponseType>
   friend class internal::RpcMethodHandler;
   template <class ServiceType, class RequestType, class ResponseType>
   friend class internal::ServerStreamingHandler;
+  template <class RequestType, class ResponseType>
+  friend class internal::CallbackUnaryHandler;
+  template <class RequestType, class ResponseType>
+  friend class ::grpc::internal::CallbackServerStreamingHandler;
   template <StatusCode code>
   friend class internal::ErrorMethodHandler;
   template <class R>
diff --git a/include/grpcpp/impl/codegen/call.h b/include/grpcpp/impl/codegen/call.h
index 789ea805a3..c040c30dd9 100644
--- a/include/grpcpp/impl/codegen/call.h
+++ b/include/grpcpp/impl/codegen/call.h
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015 gRPC authors.
+ * Copyright 2018 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,671 +15,31 @@
  * limitations under the License.
  *
  */
-
 #ifndef GRPCPP_IMPL_CODEGEN_CALL_H
 #define GRPCPP_IMPL_CODEGEN_CALL_H
 
-#include <assert.h>
-#include <cstring>
-#include <functional>
-#include <map>
-#include <memory>
-
-#include <grpcpp/impl/codegen/byte_buffer.h>
-#include <grpcpp/impl/codegen/call_hook.h>
-#include <grpcpp/impl/codegen/client_context.h>
-#include <grpcpp/impl/codegen/completion_queue_tag.h>
-#include <grpcpp/impl/codegen/config.h>
-#include <grpcpp/impl/codegen/core_codegen_interface.h>
-#include <grpcpp/impl/codegen/serialization_traits.h>
-#include <grpcpp/impl/codegen/slice.h>
-#include <grpcpp/impl/codegen/status.h>
-#include <grpcpp/impl/codegen/string_ref.h>
-
-#include <grpc/impl/codegen/atm.h>
-#include <grpc/impl/codegen/compression_types.h>
 #include <grpc/impl/codegen/grpc_types.h>
+#include <grpcpp/impl/codegen/call_hook.h>
 
 namespace grpc {
-
-class ByteBuffer;
 class CompletionQueue;
-extern CoreCodegenInterface* g_core_codegen_interface;
 
+namespace experimental {
+class ClientRpcInfo;
+class ServerRpcInfo;
+}  // namespace experimental
 namespace internal {
-class Call;
 class CallHook;
-
-// TODO(yangg) if the map is changed before we send, the pointers will be a
-// mess. Make sure it does not happen.
-inline grpc_metadata* FillMetadataArray(
-    const std::multimap<grpc::string, grpc::string>& metadata,
-    size_t* metadata_count, const grpc::string& optional_error_details) {
-  *metadata_count = metadata.size() + (optional_error_details.empty() ? 0 : 1);
-  if (*metadata_count == 0) {
-    return nullptr;
-  }
-  grpc_metadata* metadata_array =
-      (grpc_metadata*)(g_core_codegen_interface->gpr_malloc(
-          (*metadata_count) * sizeof(grpc_metadata)));
-  size_t i = 0;
-  for (auto iter = metadata.cbegin(); iter != metadata.cend(); ++iter, ++i) {
-    metadata_array[i].key = SliceReferencingString(iter->first);
-    metadata_array[i].value = SliceReferencingString(iter->second);
-  }
-  if (!optional_error_details.empty()) {
-    metadata_array[i].key =
-        g_core_codegen_interface->grpc_slice_from_static_buffer(
-            kBinaryErrorDetailsKey, sizeof(kBinaryErrorDetailsKey) - 1);
-    metadata_array[i].value = SliceReferencingString(optional_error_details);
-  }
-  return metadata_array;
-}
-}  // namespace internal
-
-/// Per-message write options.
-class WriteOptions {
- public:
-  WriteOptions() : flags_(0), last_message_(false) {}
-  WriteOptions(const WriteOptions& other)
-      : flags_(other.flags_), last_message_(other.last_message_) {}
-
-  /// Clear all flags.
-  inline void Clear() { flags_ = 0; }
-
-  /// Returns raw flags bitset.
-  inline uint32_t flags() const { return flags_; }
-
-  /// Sets flag for the disabling of compression for the next message write.
-  ///
-  /// \sa GRPC_WRITE_NO_COMPRESS
-  inline WriteOptions& set_no_compression() {
-    SetBit(GRPC_WRITE_NO_COMPRESS);
-    return *this;
-  }
-
-  /// Clears flag for the disabling of compression for the next message write.
-  ///
-  /// \sa GRPC_WRITE_NO_COMPRESS
-  inline WriteOptions& clear_no_compression() {
-    ClearBit(GRPC_WRITE_NO_COMPRESS);
-    return *this;
-  }
-
-  /// Get value for the flag indicating whether compression for the next
-  /// message write is forcefully disabled.
-  ///
-  /// \sa GRPC_WRITE_NO_COMPRESS
-  inline bool get_no_compression() const {
-    return GetBit(GRPC_WRITE_NO_COMPRESS);
-  }
-
-  /// Sets flag indicating that the write may be buffered and need not go out on
-  /// the wire immediately.
-  ///
-  /// \sa GRPC_WRITE_BUFFER_HINT
-  inline WriteOptions& set_buffer_hint() {
-    SetBit(GRPC_WRITE_BUFFER_HINT);
-    return *this;
-  }
-
-  /// Clears flag indicating that the write may be buffered and need not go out
-  /// on the wire immediately.
-  ///
-  /// \sa GRPC_WRITE_BUFFER_HINT
-  inline WriteOptions& clear_buffer_hint() {
-    ClearBit(GRPC_WRITE_BUFFER_HINT);
-    return *this;
-  }
-
-  /// Get value for the flag indicating that the write may be buffered and need
-  /// not go out on the wire immediately.
-  ///
-  /// \sa GRPC_WRITE_BUFFER_HINT
-  inline bool get_buffer_hint() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
-
-  /// corked bit: aliases set_buffer_hint currently, with the intent that
-  /// set_buffer_hint will be removed in the future
-  inline WriteOptions& set_corked() {
-    SetBit(GRPC_WRITE_BUFFER_HINT);
-    return *this;
-  }
-
-  inline WriteOptions& clear_corked() {
-    ClearBit(GRPC_WRITE_BUFFER_HINT);
-    return *this;
-  }
-
-  inline bool is_corked() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
-
-  /// last-message bit: indicates this is the last message in a stream
-  /// client-side:  makes Write the equivalent of performing Write, WritesDone
-  /// in a single step
-  /// server-side:  hold the Write until the service handler returns (sync api)
-  /// or until Finish is called (async api)
-  inline WriteOptions& set_last_message() {
-    last_message_ = true;
-    return *this;
-  }
-
-  /// Clears flag indicating that this is the last message in a stream,
-  /// disabling coalescing.
-  inline WriteOptions& clear_last_message() {
-    last_message_ = false;
-    return *this;
-  }
-
-  /// Guarantee that all bytes have been written to the socket before completing
-  /// this write (usually writes are completed when they pass flow control).
-  inline WriteOptions& set_write_through() {
-    SetBit(GRPC_WRITE_THROUGH);
-    return *this;
-  }
-
-  inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); }
-
-  /// Get value for the flag indicating that this is the last message, and
-  /// should be coalesced with trailing metadata.
-  ///
-  /// \sa GRPC_WRITE_LAST_MESSAGE
-  bool is_last_message() const { return last_message_; }
-
-  WriteOptions& operator=(const WriteOptions& rhs) {
-    flags_ = rhs.flags_;
-    return *this;
-  }
-
- private:
-  void SetBit(const uint32_t mask) { flags_ |= mask; }
-
-  void ClearBit(const uint32_t mask) { flags_ &= ~mask; }
-
-  bool GetBit(const uint32_t mask) const { return (flags_ & mask) != 0; }
-
-  uint32_t flags_;
-  bool last_message_;
-};
-
-namespace internal {
-/// Default argument for CallOpSet. I is unused by the class, but can be
-/// used for generating multiple names for the same thing.
-template <int I>
-class CallNoOp {
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {}
-  void FinishOp(bool* status) {}
-};
-
-class CallOpSendInitialMetadata {
- public:
-  CallOpSendInitialMetadata() : send_(false) {
-    maybe_compression_level_.is_set = false;
-  }
-
-  void SendInitialMetadata(
-      const std::multimap<grpc::string, grpc::string>& metadata,
-      uint32_t flags) {
-    maybe_compression_level_.is_set = false;
-    send_ = true;
-    flags_ = flags;
-    initial_metadata_ =
-        FillMetadataArray(metadata, &initial_metadata_count_, "");
-  }
-
-  void set_compression_level(grpc_compression_level level) {
-    maybe_compression_level_.is_set = true;
-    maybe_compression_level_.level = level;
-  }
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (!send_) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_SEND_INITIAL_METADATA;
-    op->flags = flags_;
-    op->reserved = NULL;
-    op->data.send_initial_metadata.count = initial_metadata_count_;
-    op->data.send_initial_metadata.metadata = initial_metadata_;
-    op->data.send_initial_metadata.maybe_compression_level.is_set =
-        maybe_compression_level_.is_set;
-    if (maybe_compression_level_.is_set) {
-      op->data.send_initial_metadata.maybe_compression_level.level =
-          maybe_compression_level_.level;
-    }
-  }
-  void FinishOp(bool* status) {
-    if (!send_) return;
-    g_core_codegen_interface->gpr_free(initial_metadata_);
-    send_ = false;
-  }
-
-  bool send_;
-  uint32_t flags_;
-  size_t initial_metadata_count_;
-  grpc_metadata* initial_metadata_;
-  struct {
-    bool is_set;
-    grpc_compression_level level;
-  } maybe_compression_level_;
-};
-
-class CallOpSendMessage {
- public:
-  CallOpSendMessage() : send_buf_() {}
-
-  /// Send \a message using \a options for the write. The \a options are cleared
-  /// after use.
-  template <class M>
-  Status SendMessage(const M& message,
-                     WriteOptions options) GRPC_MUST_USE_RESULT;
-
-  template <class M>
-  Status SendMessage(const M& message) GRPC_MUST_USE_RESULT;
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (!send_buf_.Valid()) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_SEND_MESSAGE;
-    op->flags = write_options_.flags();
-    op->reserved = NULL;
-    op->data.send_message.send_message = send_buf_.c_buffer();
-    // Flags are per-message: clear them after use.
-    write_options_.Clear();
-  }
-  void FinishOp(bool* status) { send_buf_.Clear(); }
-
- private:
-  ByteBuffer send_buf_;
-  WriteOptions write_options_;
-};
-
-template <class M>
-Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) {
-  write_options_ = options;
-  bool own_buf;
-  // TODO(vjpai): Remove the void below when possible
-  // The void in the template parameter below should not be needed
-  // (since it should be implicit) but is needed due to an observed
-  // difference in behavior between clang and gcc for certain internal users
-  Status result = SerializationTraits<M, void>::Serialize(
-      message, send_buf_.bbuf_ptr(), &own_buf);
-  if (!own_buf) {
-    send_buf_.Duplicate();
-  }
-  return result;
-}
-
-template <class M>
-Status CallOpSendMessage::SendMessage(const M& message) {
-  return SendMessage(message, WriteOptions());
-}
-
-template <class R>
-class CallOpRecvMessage {
- public:
-  CallOpRecvMessage()
-      : got_message(false),
-        message_(nullptr),
-        allow_not_getting_message_(false) {}
-
-  void RecvMessage(R* message) { message_ = message; }
-
-  // Do not change status if no message is received.
-  void AllowNoMessage() { allow_not_getting_message_ = true; }
-
-  bool got_message;
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (message_ == nullptr) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_RECV_MESSAGE;
-    op->flags = 0;
-    op->reserved = NULL;
-    op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr();
-  }
-
-  void FinishOp(bool* status) {
-    if (message_ == nullptr) return;
-    if (recv_buf_.Valid()) {
-      if (*status) {
-        got_message = *status =
-            SerializationTraits<R>::Deserialize(recv_buf_.bbuf_ptr(), message_)
-                .ok();
-        recv_buf_.Release();
-      } else {
-        got_message = false;
-        recv_buf_.Clear();
-      }
-    } else {
-      got_message = false;
-      if (!allow_not_getting_message_) {
-        *status = false;
-      }
-    }
-    message_ = nullptr;
-  }
-
- private:
-  R* message_;
-  ByteBuffer recv_buf_;
-  bool allow_not_getting_message_;
-};
-
-class DeserializeFunc {
- public:
-  virtual Status Deserialize(ByteBuffer* buf) = 0;
-  virtual ~DeserializeFunc() {}
-};
-
-template <class R>
-class DeserializeFuncType final : public DeserializeFunc {
- public:
-  DeserializeFuncType(R* message) : message_(message) {}
-  Status Deserialize(ByteBuffer* buf) override {
-    return SerializationTraits<R>::Deserialize(buf->bbuf_ptr(), message_);
-  }
-
-  ~DeserializeFuncType() override {}
-
- private:
-  R* message_;  // Not a managed pointer because management is external to this
-};
-
-class CallOpGenericRecvMessage {
- public:
-  CallOpGenericRecvMessage()
-      : got_message(false), allow_not_getting_message_(false) {}
-
-  template <class R>
-  void RecvMessage(R* message) {
-    // Use an explicit base class pointer to avoid resolution error in the
-    // following unique_ptr::reset for some old implementations.
-    DeserializeFunc* func = new DeserializeFuncType<R>(message);
-    deserialize_.reset(func);
-  }
-
-  // Do not change status if no message is received.
-  void AllowNoMessage() { allow_not_getting_message_ = true; }
-
-  bool got_message;
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (!deserialize_) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_RECV_MESSAGE;
-    op->flags = 0;
-    op->reserved = NULL;
-    op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr();
-  }
-
-  void FinishOp(bool* status) {
-    if (!deserialize_) return;
-    if (recv_buf_.Valid()) {
-      if (*status) {
-        got_message = true;
-        *status = deserialize_->Deserialize(&recv_buf_).ok();
-        recv_buf_.Release();
-      } else {
-        got_message = false;
-        recv_buf_.Clear();
-      }
-    } else {
-      got_message = false;
-      if (!allow_not_getting_message_) {
-        *status = false;
-      }
-    }
-    deserialize_.reset();
-  }
-
- private:
-  std::unique_ptr<DeserializeFunc> deserialize_;
-  ByteBuffer recv_buf_;
-  bool allow_not_getting_message_;
-};
-
-class CallOpClientSendClose {
- public:
-  CallOpClientSendClose() : send_(false) {}
-
-  void ClientSendClose() { send_ = true; }
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (!send_) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
-    op->flags = 0;
-    op->reserved = NULL;
-  }
-  void FinishOp(bool* status) { send_ = false; }
-
- private:
-  bool send_;
-};
-
-class CallOpServerSendStatus {
- public:
-  CallOpServerSendStatus() : send_status_available_(false) {}
-
-  void ServerSendStatus(
-      const std::multimap<grpc::string, grpc::string>& trailing_metadata,
-      const Status& status) {
-    send_error_details_ = status.error_details();
-    trailing_metadata_ = FillMetadataArray(
-        trailing_metadata, &trailing_metadata_count_, send_error_details_);
-    send_status_available_ = true;
-    send_status_code_ = static_cast<grpc_status_code>(status.error_code());
-    send_error_message_ = status.error_message();
-  }
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (!send_status_available_) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
-    op->data.send_status_from_server.trailing_metadata_count =
-        trailing_metadata_count_;
-    op->data.send_status_from_server.trailing_metadata = trailing_metadata_;
-    op->data.send_status_from_server.status = send_status_code_;
-    error_message_slice_ = SliceReferencingString(send_error_message_);
-    op->data.send_status_from_server.status_details =
-        send_error_message_.empty() ? nullptr : &error_message_slice_;
-    op->flags = 0;
-    op->reserved = NULL;
-  }
-
-  void FinishOp(bool* status) {
-    if (!send_status_available_) return;
-    g_core_codegen_interface->gpr_free(trailing_metadata_);
-    send_status_available_ = false;
-  }
-
- private:
-  bool send_status_available_;
-  grpc_status_code send_status_code_;
-  grpc::string send_error_details_;
-  grpc::string send_error_message_;
-  size_t trailing_metadata_count_;
-  grpc_metadata* trailing_metadata_;
-  grpc_slice error_message_slice_;
-};
-
-class CallOpRecvInitialMetadata {
- public:
-  CallOpRecvInitialMetadata() : metadata_map_(nullptr) {}
-
-  void RecvInitialMetadata(ClientContext* context) {
-    context->initial_metadata_received_ = true;
-    metadata_map_ = &context->recv_initial_metadata_;
-  }
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (metadata_map_ == nullptr) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_RECV_INITIAL_METADATA;
-    op->data.recv_initial_metadata.recv_initial_metadata = metadata_map_->arr();
-    op->flags = 0;
-    op->reserved = NULL;
-  }
-
-  void FinishOp(bool* status) {
-    if (metadata_map_ == nullptr) return;
-    metadata_map_ = nullptr;
-  }
-
- private:
-  MetadataMap* metadata_map_;
-};
-
-class CallOpClientRecvStatus {
- public:
-  CallOpClientRecvStatus()
-      : recv_status_(nullptr), debug_error_string_(nullptr) {}
-
-  void ClientRecvStatus(ClientContext* context, Status* status) {
-    client_context_ = context;
-    metadata_map_ = &client_context_->trailing_metadata_;
-    recv_status_ = status;
-    error_message_ = g_core_codegen_interface->grpc_empty_slice();
-  }
-
- protected:
-  void AddOp(grpc_op* ops, size_t* nops) {
-    if (recv_status_ == nullptr) return;
-    grpc_op* op = &ops[(*nops)++];
-    op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
-    op->data.recv_status_on_client.trailing_metadata = metadata_map_->arr();
-    op->data.recv_status_on_client.status = &status_code_;
-    op->data.recv_status_on_client.status_details = &error_message_;
-    op->data.recv_status_on_client.error_string = &debug_error_string_;
-    op->flags = 0;
-    op->reserved = NULL;
-  }
-
-  void FinishOp(bool* status) {
-    if (recv_status_ == nullptr) return;
-    grpc::string binary_error_details = metadata_map_->GetBinaryErrorDetails();
-    *recv_status_ =
-        Status(static_cast<StatusCode>(status_code_),
-               GRPC_SLICE_IS_EMPTY(error_message_)
-                   ? grpc::string()
-                   : grpc::string(GRPC_SLICE_START_PTR(error_message_),
-                                  GRPC_SLICE_END_PTR(error_message_)),
-               binary_error_details);
-    client_context_->set_debug_error_string(
-        debug_error_string_ != nullptr ? debug_error_string_ : "");
-    g_core_codegen_interface->grpc_slice_unref(error_message_);
-    if (debug_error_string_ != nullptr) {
-      g_core_codegen_interface->gpr_free((void*)debug_error_string_);
-    }
-    recv_status_ = nullptr;
-  }
-
- private:
-  ClientContext* client_context_;
-  MetadataMap* metadata_map_;
-  Status* recv_status_;
-  const char* debug_error_string_;
-  grpc_status_code status_code_;
-  grpc_slice error_message_;
-};
-
-/// An abstract collection of call ops, used to generate the
-/// grpc_call_op structure to pass down to the lower layers,
-/// and as it is-a CompletionQueueTag, also massages the final
-/// completion into the correct form for consumption in the C++
-/// API.
-class CallOpSetInterface : public CompletionQueueTag {
- public:
-  /// Fills in grpc_op, starting from ops[*nops] and moving
-  /// upwards.
-  virtual void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) = 0;
-
-  /// Get the tag to be used at the core completion queue. Generally, the
-  /// value of cq_tag will be "this". However, it can be overridden if we
-  /// want core to process the tag differently (e.g., as a core callback)
-  virtual void* cq_tag() = 0;
-};
-
-/// Primary implementation of CallOpSetInterface.
-/// Since we cannot use variadic templates, we declare slots up to
-/// the maximum count of ops we'll need in a set. We leverage the
-/// empty base class optimization to slim this class (especially
-/// when there are many unused slots used). To avoid duplicate base classes,
-/// the template parmeter for CallNoOp is varied by argument position.
-template <class Op1 = CallNoOp<1>, class Op2 = CallNoOp<2>,
-          class Op3 = CallNoOp<3>, class Op4 = CallNoOp<4>,
-          class Op5 = CallNoOp<5>, class Op6 = CallNoOp<6>>
-class CallOpSet : public CallOpSetInterface,
-                  public Op1,
-                  public Op2,
-                  public Op3,
-                  public Op4,
-                  public Op5,
-                  public Op6 {
- public:
-  CallOpSet() : cq_tag_(this), return_tag_(this), call_(nullptr) {}
-
-  // The copy constructor and assignment operator reset the value of
-  // cq_tag_ and return_tag_ since those are only meaningful on a specific
-  // object, not across objects.
-  CallOpSet(const CallOpSet& other)
-      : cq_tag_(this), return_tag_(this), call_(other.call_) {}
-  CallOpSet& operator=(const CallOpSet& other) {
-    cq_tag_ = this;
-    return_tag_ = this;
-    call_ = other.call_;
-    return *this;
-  }
-
-  void FillOps(grpc_call* call, grpc_op* ops, size_t* nops) override {
-    this->Op1::AddOp(ops, nops);
-    this->Op2::AddOp(ops, nops);
-    this->Op3::AddOp(ops, nops);
-    this->Op4::AddOp(ops, nops);
-    this->Op5::AddOp(ops, nops);
-    this->Op6::AddOp(ops, nops);
-    g_core_codegen_interface->grpc_call_ref(call);
-    call_ = call;
-  }
-
-  bool FinalizeResult(void** tag, bool* status) override {
-    this->Op1::FinishOp(status);
-    this->Op2::FinishOp(status);
-    this->Op3::FinishOp(status);
-    this->Op4::FinishOp(status);
-    this->Op5::FinishOp(status);
-    this->Op6::FinishOp(status);
-    *tag = return_tag_;
-
-    g_core_codegen_interface->grpc_call_unref(call_);
-    return true;
-  }
-
-  void set_output_tag(void* return_tag) { return_tag_ = return_tag; }
-
-  void* cq_tag() override { return cq_tag_; }
-
-  /// set_cq_tag is used to provide a different core CQ tag than "this".
-  /// This is used for callback-based tags, where the core tag is the core
-  /// callback function. It does not change the use or behavior of any other
-  /// function (such as FinalizeResult)
-  void set_cq_tag(void* cq_tag) { cq_tag_ = cq_tag; }
-
- private:
-  void* cq_tag_;
-  void* return_tag_;
-  grpc_call* call_;
-};
+class CallOpSetInterface;
 
 /// Straightforward wrapping of the C call object
 class Call final {
  public:
+  Call()
+      : call_hook_(nullptr),
+        cq_(nullptr),
+        call_(nullptr),
+        max_receive_message_size_(-1) {}
   /** call is owned by the caller */
   Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq)
       : call_hook_(call_hook),
@@ -688,11 +48,20 @@ class Call final {
         max_receive_message_size_(-1) {}
 
   Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
-       int max_receive_message_size)
+       experimental::ClientRpcInfo* rpc_info)
       : call_hook_(call_hook),
         cq_(cq),
         call_(call),
-        max_receive_message_size_(max_receive_message_size) {}
+        max_receive_message_size_(-1),
+        client_rpc_info_(rpc_info) {}
+
+  Call(grpc_call* call, CallHook* call_hook, CompletionQueue* cq,
+       int max_receive_message_size, experimental::ServerRpcInfo* rpc_info)
+      : call_hook_(call_hook),
+        cq_(cq),
+        call_(call),
+        max_receive_message_size_(max_receive_message_size),
+        server_rpc_info_(rpc_info) {}
 
   void PerformOps(CallOpSetInterface* ops) {
     call_hook_->PerformOpsOnCall(ops, this);
@@ -703,11 +72,21 @@ class Call final {
 
   int max_receive_message_size() const { return max_receive_message_size_; }
 
+  experimental::ClientRpcInfo* client_rpc_info() const {
+    return client_rpc_info_;
+  }
+
+  experimental::ServerRpcInfo* server_rpc_info() const {
+    return server_rpc_info_;
+  }
+
  private:
   CallHook* call_hook_;
   CompletionQueue* cq_;
   grpc_call* call_;
   int max_receive_message_size_;
+  experimental::ClientRpcInfo* client_rpc_info_ = nullptr;
+  experimental::ServerRpcInfo* server_rpc_info_ = nullptr;
 };
 }  // namespace internal
 }  // namespace grpc
diff --git a/include/grpcpp/impl/codegen/call_op_set.h b/include/grpcpp/impl/codegen/call_op_set.h
new file mode 100644
index 0000000000..b4c34a01c9
--- /dev/null
+++ b/include/grpcpp/impl/codegen/call_op_set.h
@@ -0,0 +1,919 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H
+#define GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H
+
+#include <assert.h>
+#include <array>
+#include <cstring>
+#include <functional>
+#include <map>
+#include <memory>
+#include <vector>
+
+#include <grpcpp/impl/codegen/byte_buffer.h>
+#include <grpcpp/impl/codegen/call.h>
+#include <grpcpp/impl/codegen/call_hook.h>
+#include <grpcpp/impl/codegen/call_op_set_interface.h>
+#include <grpcpp/impl/codegen/client_context.h>
+#include <grpcpp/impl/codegen/completion_queue_tag.h>
+#include <grpcpp/impl/codegen/config.h>
+#include <grpcpp/impl/codegen/core_codegen_interface.h>
+#include <grpcpp/impl/codegen/intercepted_channel.h>
+#include <grpcpp/impl/codegen/interceptor_common.h>
+#include <grpcpp/impl/codegen/serialization_traits.h>
+#include <grpcpp/impl/codegen/slice.h>
+#include <grpcpp/impl/codegen/string_ref.h>
+
+#include <grpc/impl/codegen/atm.h>
+#include <grpc/impl/codegen/compression_types.h>
+#include <grpc/impl/codegen/grpc_types.h>
+
+namespace grpc {
+
+class CompletionQueue;
+extern CoreCodegenInterface* g_core_codegen_interface;
+
+namespace internal {
+class Call;
+class CallHook;
+
+// TODO(yangg) if the map is changed before we send, the pointers will be a
+// mess. Make sure it does not happen.
+inline grpc_metadata* FillMetadataArray(
+    const std::multimap<grpc::string, grpc::string>& metadata,
+    size_t* metadata_count, const grpc::string& optional_error_details) {
+  *metadata_count = metadata.size() + (optional_error_details.empty() ? 0 : 1);
+  if (*metadata_count == 0) {
+    return nullptr;
+  }
+  grpc_metadata* metadata_array =
+      (grpc_metadata*)(g_core_codegen_interface->gpr_malloc(
+          (*metadata_count) * sizeof(grpc_metadata)));
+  size_t i = 0;
+  for (auto iter = metadata.cbegin(); iter != metadata.cend(); ++iter, ++i) {
+    metadata_array[i].key = SliceReferencingString(iter->first);
+    metadata_array[i].value = SliceReferencingString(iter->second);
+  }
+  if (!optional_error_details.empty()) {
+    metadata_array[i].key =
+        g_core_codegen_interface->grpc_slice_from_static_buffer(
+            kBinaryErrorDetailsKey, sizeof(kBinaryErrorDetailsKey) - 1);
+    metadata_array[i].value = SliceReferencingString(optional_error_details);
+  }
+  return metadata_array;
+}
+}  // namespace internal
+
+/// Per-message write options.
+class WriteOptions {
+ public:
+  WriteOptions() : flags_(0), last_message_(false) {}
+  WriteOptions(const WriteOptions& other)
+      : flags_(other.flags_), last_message_(other.last_message_) {}
+
+  /// Clear all flags.
+  inline void Clear() { flags_ = 0; }
+
+  /// Returns raw flags bitset.
+  inline uint32_t flags() const { return flags_; }
+
+  /// Sets flag for the disabling of compression for the next message write.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline WriteOptions& set_no_compression() {
+    SetBit(GRPC_WRITE_NO_COMPRESS);
+    return *this;
+  }
+
+  /// Clears flag for the disabling of compression for the next message write.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline WriteOptions& clear_no_compression() {
+    ClearBit(GRPC_WRITE_NO_COMPRESS);
+    return *this;
+  }
+
+  /// Get value for the flag indicating whether compression for the next
+  /// message write is forcefully disabled.
+  ///
+  /// \sa GRPC_WRITE_NO_COMPRESS
+  inline bool get_no_compression() const {
+    return GetBit(GRPC_WRITE_NO_COMPRESS);
+  }
+
+  /// Sets flag indicating that the write may be buffered and need not go out on
+  /// the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline WriteOptions& set_buffer_hint() {
+    SetBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  /// Clears flag indicating that the write may be buffered and need not go out
+  /// on the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline WriteOptions& clear_buffer_hint() {
+    ClearBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  /// Get value for the flag indicating that the write may be buffered and need
+  /// not go out on the wire immediately.
+  ///
+  /// \sa GRPC_WRITE_BUFFER_HINT
+  inline bool get_buffer_hint() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
+
+  /// corked bit: aliases set_buffer_hint currently, with the intent that
+  /// set_buffer_hint will be removed in the future
+  inline WriteOptions& set_corked() {
+    SetBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  inline WriteOptions& clear_corked() {
+    ClearBit(GRPC_WRITE_BUFFER_HINT);
+    return *this;
+  }
+
+  inline bool is_corked() const { return GetBit(GRPC_WRITE_BUFFER_HINT); }
+
+  /// last-message bit: indicates this is the last message in a stream
+  /// client-side:  makes Write the equivalent of performing Write, WritesDone
+  /// in a single step
+  /// server-side:  hold the Write until the service handler returns (sync api)
+  /// or until Finish is called (async api)
+  inline WriteOptions& set_last_message() {
+    last_message_ = true;
+    return *this;
+  }
+
+  /// Clears flag indicating that this is the last message in a stream,
+  /// disabling coalescing.
+  inline WriteOptions& clear_last_message() {
+    last_message_ = false;
+    return *this;
+  }
+
+  /// Guarantee that all bytes have been written to the socket before completing
+  /// this write (usually writes are completed when they pass flow control).
+  inline WriteOptions& set_write_through() {
+    SetBit(GRPC_WRITE_THROUGH);
+    return *this;
+  }
+
+  inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); }
+
+  /// Get value for the flag indicating that this is the last message, and
+  /// should be coalesced with trailing metadata.
+  ///
+  /// \sa GRPC_WRITE_LAST_MESSAGE
+  bool is_last_message() const { return last_message_; }
+
+  WriteOptions& operator=(const WriteOptions& rhs) {
+    flags_ = rhs.flags_;
+    return *this;
+  }
+
+ private:
+  void SetBit(const uint32_t mask) { flags_ |= mask; }
+
+  void ClearBit(const uint32_t mask) { flags_ &= ~mask; }
+
+  bool GetBit(const uint32_t mask) const { return (flags_ & mask) != 0; }
+
+  uint32_t flags_;
+  bool last_message_;
+};
+
+namespace internal {
+
+/// Default argument for CallOpSet. I is unused by the class, but can be
+/// used for generating multiple names for the same thing.
+template <int I>
+class CallNoOp {
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {}
+  void FinishOp(bool* status) {}
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {}
+};
+
+class CallOpSendInitialMetadata {
+ public:
+  CallOpSendInitialMetadata() : send_(false) {
+    maybe_compression_level_.is_set = false;
+  }
+
+  void SendInitialMetadata(std::multimap<grpc::string, grpc::string>* metadata,
+                           uint32_t flags) {
+    maybe_compression_level_.is_set = false;
+    send_ = true;
+    flags_ = flags;
+    metadata_map_ = metadata;
+  }
+
+  void set_compression_level(grpc_compression_level level) {
+    maybe_compression_level_.is_set = true;
+    maybe_compression_level_.level = level;
+  }
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (!send_ || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_SEND_INITIAL_METADATA;
+    op->flags = flags_;
+    op->reserved = NULL;
+    initial_metadata_ =
+        FillMetadataArray(*metadata_map_, &initial_metadata_count_, "");
+    op->data.send_initial_metadata.count = initial_metadata_count_;
+    op->data.send_initial_metadata.metadata = initial_metadata_;
+    op->data.send_initial_metadata.maybe_compression_level.is_set =
+        maybe_compression_level_.is_set;
+    if (maybe_compression_level_.is_set) {
+      op->data.send_initial_metadata.maybe_compression_level.level =
+          maybe_compression_level_.level;
+    }
+  }
+  void FinishOp(bool* status) {
+    if (!send_ || hijacked_) return;
+    g_core_codegen_interface->gpr_free(initial_metadata_);
+    send_ = false;
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!send_) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA);
+    interceptor_methods->SetSendInitialMetadata(metadata_map_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+  }
+
+  bool hijacked_ = false;
+  bool send_;
+  uint32_t flags_;
+  size_t initial_metadata_count_;
+  std::multimap<grpc::string, grpc::string>* metadata_map_;
+  grpc_metadata* initial_metadata_;
+  struct {
+    bool is_set;
+    grpc_compression_level level;
+  } maybe_compression_level_;
+};
+
+class CallOpSendMessage {
+ public:
+  CallOpSendMessage() : send_buf_() {}
+
+  /// Send \a message using \a options for the write. The \a options are cleared
+  /// after use.
+  template <class M>
+  Status SendMessage(const M& message,
+                     WriteOptions options) GRPC_MUST_USE_RESULT;
+
+  template <class M>
+  Status SendMessage(const M& message) GRPC_MUST_USE_RESULT;
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (!send_buf_.Valid() || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_SEND_MESSAGE;
+    op->flags = write_options_.flags();
+    op->reserved = NULL;
+    op->data.send_message.send_message = send_buf_.c_buffer();
+    // Flags are per-message: clear them after use.
+    write_options_.Clear();
+  }
+  void FinishOp(bool* status) { send_buf_.Clear(); }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!send_buf_.Valid()) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_SEND_MESSAGE);
+    interceptor_methods->SetSendMessage(&send_buf_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+  }
+
+ private:
+  bool hijacked_ = false;
+  ByteBuffer send_buf_;
+  WriteOptions write_options_;
+};
+
+template <class M>
+Status CallOpSendMessage::SendMessage(const M& message, WriteOptions options) {
+  write_options_ = options;
+  bool own_buf;
+  // TODO(vjpai): Remove the void below when possible
+  // The void in the template parameter below should not be needed
+  // (since it should be implicit) but is needed due to an observed
+  // difference in behavior between clang and gcc for certain internal users
+  Status result = SerializationTraits<M, void>::Serialize(
+      message, send_buf_.bbuf_ptr(), &own_buf);
+  if (!own_buf) {
+    send_buf_.Duplicate();
+  }
+  return result;
+}
+
+template <class M>
+Status CallOpSendMessage::SendMessage(const M& message) {
+  return SendMessage(message, WriteOptions());
+}
+
+template <class R>
+class CallOpRecvMessage {
+ public:
+  CallOpRecvMessage()
+      : got_message(false),
+        message_(nullptr),
+        allow_not_getting_message_(false) {}
+
+  void RecvMessage(R* message) { message_ = message; }
+
+  // Do not change status if no message is received.
+  void AllowNoMessage() { allow_not_getting_message_ = true; }
+
+  bool got_message;
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (message_ == nullptr || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_RECV_MESSAGE;
+    op->flags = 0;
+    op->reserved = NULL;
+    op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr();
+  }
+
+  void FinishOp(bool* status) {
+    if (message_ == nullptr || hijacked_) return;
+    if (recv_buf_.Valid()) {
+      if (*status) {
+        got_message = *status =
+            SerializationTraits<R>::Deserialize(recv_buf_.bbuf_ptr(), message_)
+                .ok();
+        recv_buf_.Release();
+      } else {
+        got_message = false;
+        recv_buf_.Clear();
+      }
+    } else {
+      got_message = false;
+      if (!allow_not_getting_message_) {
+        *status = false;
+      }
+    }
+    message_ = nullptr;
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    interceptor_methods->SetRecvMessage(message_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!got_message) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+  }
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+    if (message_ == nullptr) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_RECV_MESSAGE);
+    got_message = true;
+  }
+
+ private:
+  R* message_;
+  ByteBuffer recv_buf_;
+  bool allow_not_getting_message_;
+  bool hijacked_ = false;
+};
+
+class DeserializeFunc {
+ public:
+  virtual Status Deserialize(ByteBuffer* buf) = 0;
+  virtual ~DeserializeFunc() {}
+};
+
+template <class R>
+class DeserializeFuncType final : public DeserializeFunc {
+ public:
+  DeserializeFuncType(R* message) : message_(message) {}
+  Status Deserialize(ByteBuffer* buf) override {
+    return SerializationTraits<R>::Deserialize(buf->bbuf_ptr(), message_);
+  }
+
+  ~DeserializeFuncType() override {}
+
+ private:
+  R* message_;  // Not a managed pointer because management is external to this
+};
+
+class CallOpGenericRecvMessage {
+ public:
+  CallOpGenericRecvMessage()
+      : got_message(false), allow_not_getting_message_(false) {}
+
+  template <class R>
+  void RecvMessage(R* message) {
+    // Use an explicit base class pointer to avoid resolution error in the
+    // following unique_ptr::reset for some old implementations.
+    DeserializeFunc* func = new DeserializeFuncType<R>(message);
+    deserialize_.reset(func);
+    message_ = message;
+  }
+
+  // Do not change status if no message is received.
+  void AllowNoMessage() { allow_not_getting_message_ = true; }
+
+  bool got_message;
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (!deserialize_ || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_RECV_MESSAGE;
+    op->flags = 0;
+    op->reserved = NULL;
+    op->data.recv_message.recv_message = recv_buf_.c_buffer_ptr();
+  }
+
+  void FinishOp(bool* status) {
+    if (!deserialize_ || hijacked_) return;
+    if (recv_buf_.Valid()) {
+      if (*status) {
+        got_message = true;
+        *status = deserialize_->Deserialize(&recv_buf_).ok();
+        recv_buf_.Release();
+      } else {
+        got_message = false;
+        recv_buf_.Clear();
+      }
+    } else {
+      got_message = false;
+      if (!allow_not_getting_message_) {
+        *status = false;
+      }
+    }
+    deserialize_.reset();
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    interceptor_methods->SetRecvMessage(message_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!got_message) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+  }
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+    if (!deserialize_) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_RECV_MESSAGE);
+  }
+
+ private:
+  void* message_;
+  bool hijacked_ = false;
+  std::unique_ptr<DeserializeFunc> deserialize_;
+  ByteBuffer recv_buf_;
+  bool allow_not_getting_message_;
+};
+
+class CallOpClientSendClose {
+ public:
+  CallOpClientSendClose() : send_(false) {}
+
+  void ClientSendClose() { send_ = true; }
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (!send_ || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+    op->flags = 0;
+    op->reserved = NULL;
+  }
+  void FinishOp(bool* status) { send_ = false; }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!send_) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_SEND_CLOSE);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+  }
+
+ private:
+  bool hijacked_ = false;
+  bool send_;
+};
+
+class CallOpServerSendStatus {
+ public:
+  CallOpServerSendStatus() : send_status_available_(false) {}
+
+  void ServerSendStatus(
+      std::multimap<grpc::string, grpc::string>* trailing_metadata,
+      const Status& status) {
+    send_error_details_ = status.error_details();
+    metadata_map_ = trailing_metadata;
+    send_status_available_ = true;
+    send_status_code_ = static_cast<grpc_status_code>(status.error_code());
+    send_error_message_ = status.error_message();
+  }
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (!send_status_available_ || hijacked_) return;
+    trailing_metadata_ = FillMetadataArray(
+        *metadata_map_, &trailing_metadata_count_, send_error_details_);
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+    op->data.send_status_from_server.trailing_metadata_count =
+        trailing_metadata_count_;
+    op->data.send_status_from_server.trailing_metadata = trailing_metadata_;
+    op->data.send_status_from_server.status = send_status_code_;
+    error_message_slice_ = SliceReferencingString(send_error_message_);
+    op->data.send_status_from_server.status_details =
+        send_error_message_.empty() ? nullptr : &error_message_slice_;
+    op->flags = 0;
+    op->reserved = NULL;
+  }
+
+  void FinishOp(bool* status) {
+    if (!send_status_available_ || hijacked_) return;
+    g_core_codegen_interface->gpr_free(trailing_metadata_);
+    send_status_available_ = false;
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (!send_status_available_) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_SEND_STATUS);
+    interceptor_methods->SetSendTrailingMetadata(metadata_map_);
+    interceptor_methods->SetSendStatus(&send_status_code_, &send_error_details_,
+                                       &send_error_message_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {}
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+  }
+
+ private:
+  bool hijacked_ = false;
+  bool send_status_available_;
+  grpc_status_code send_status_code_;
+  grpc::string send_error_details_;
+  grpc::string send_error_message_;
+  size_t trailing_metadata_count_;
+  std::multimap<grpc::string, grpc::string>* metadata_map_;
+  grpc_metadata* trailing_metadata_;
+  grpc_slice error_message_slice_;
+};
+
+class CallOpRecvInitialMetadata {
+ public:
+  CallOpRecvInitialMetadata() : metadata_map_(nullptr) {}
+
+  void RecvInitialMetadata(ClientContext* context) {
+    context->initial_metadata_received_ = true;
+    metadata_map_ = &context->recv_initial_metadata_;
+  }
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (metadata_map_ == nullptr || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_RECV_INITIAL_METADATA;
+    op->data.recv_initial_metadata.recv_initial_metadata = metadata_map_->arr();
+    op->flags = 0;
+    op->reserved = NULL;
+  }
+
+  void FinishOp(bool* status) {
+    if (metadata_map_ == nullptr || hijacked_) return;
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    interceptor_methods->SetRecvInitialMetadata(metadata_map_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (metadata_map_ == nullptr) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA);
+    metadata_map_ = nullptr;
+  }
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+    if (metadata_map_ == nullptr) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA);
+  }
+
+ private:
+  bool hijacked_ = false;
+  MetadataMap* metadata_map_;
+};
+
+class CallOpClientRecvStatus {
+ public:
+  CallOpClientRecvStatus()
+      : recv_status_(nullptr), debug_error_string_(nullptr) {}
+
+  void ClientRecvStatus(ClientContext* context, Status* status) {
+    client_context_ = context;
+    metadata_map_ = &client_context_->trailing_metadata_;
+    recv_status_ = status;
+    error_message_ = g_core_codegen_interface->grpc_empty_slice();
+  }
+
+ protected:
+  void AddOp(grpc_op* ops, size_t* nops) {
+    if (recv_status_ == nullptr || hijacked_) return;
+    grpc_op* op = &ops[(*nops)++];
+    op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+    op->data.recv_status_on_client.trailing_metadata = metadata_map_->arr();
+    op->data.recv_status_on_client.status = &status_code_;
+    op->data.recv_status_on_client.status_details = &error_message_;
+    op->data.recv_status_on_client.error_string = &debug_error_string_;
+    op->flags = 0;
+    op->reserved = NULL;
+  }
+
+  void FinishOp(bool* status) {
+    if (recv_status_ == nullptr || hijacked_) return;
+    grpc::string binary_error_details = metadata_map_->GetBinaryErrorDetails();
+    *recv_status_ =
+        Status(static_cast<StatusCode>(status_code_),
+               GRPC_SLICE_IS_EMPTY(error_message_)
+                   ? grpc::string()
+                   : grpc::string(GRPC_SLICE_START_PTR(error_message_),
+                                  GRPC_SLICE_END_PTR(error_message_)),
+               binary_error_details);
+    client_context_->set_debug_error_string(
+        debug_error_string_ != nullptr ? debug_error_string_ : "");
+    g_core_codegen_interface->grpc_slice_unref(error_message_);
+    if (debug_error_string_ != nullptr) {
+      g_core_codegen_interface->gpr_free((void*)debug_error_string_);
+    }
+  }
+
+  void SetInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    interceptor_methods->SetRecvStatus(recv_status_);
+    interceptor_methods->SetRecvTrailingMetadata(metadata_map_);
+  }
+
+  void SetFinishInterceptionHookPoint(
+      InterceptorBatchMethodsImpl* interceptor_methods) {
+    if (recv_status_ == nullptr) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::POST_RECV_STATUS);
+    recv_status_ = nullptr;
+  }
+
+  void SetHijackingState(InterceptorBatchMethodsImpl* interceptor_methods) {
+    hijacked_ = true;
+    if (recv_status_ == nullptr) return;
+    interceptor_methods->AddInterceptionHookPoint(
+        experimental::InterceptionHookPoints::PRE_RECV_STATUS);
+  }
+
+ private:
+  bool hijacked_ = false;
+  ClientContext* client_context_;
+  MetadataMap* metadata_map_;
+  Status* recv_status_;
+  const char* debug_error_string_;
+  grpc_status_code status_code_;
+  grpc_slice error_message_;
+};
+
+template <class Op1 = CallNoOp<1>, class Op2 = CallNoOp<2>,
+          class Op3 = CallNoOp<3>, class Op4 = CallNoOp<4>,
+          class Op5 = CallNoOp<5>, class Op6 = CallNoOp<6>>
+class CallOpSet;
+
+/// Primary implementation of CallOpSetInterface.
+/// Since we cannot use variadic templates, we declare slots up to
+/// the maximum count of ops we'll need in a set. We leverage the
+/// empty base class optimization to slim this class (especially
+/// when there are many unused slots used). To avoid duplicate base classes,
+/// the template parmeter for CallNoOp is varied by argument position.
+template <class Op1, class Op2, class Op3, class Op4, class Op5, class Op6>
+class CallOpSet : public CallOpSetInterface,
+                  public Op1,
+                  public Op2,
+                  public Op3,
+                  public Op4,
+                  public Op5,
+                  public Op6 {
+ public:
+  CallOpSet() : core_cq_tag_(this), return_tag_(this) {}
+  // The copy constructor and assignment operator reset the value of
+  // core_cq_tag_, return_tag_, done_intercepting_ and interceptor_methods_
+  // since those are only meaningful on a specific object, not across objects.
+  CallOpSet(const CallOpSet& other)
+      : core_cq_tag_(this),
+        return_tag_(this),
+        call_(other.call_),
+        done_intercepting_(false),
+        interceptor_methods_(InterceptorBatchMethodsImpl()) {}
+
+  CallOpSet& operator=(const CallOpSet& other) {
+    core_cq_tag_ = this;
+    return_tag_ = this;
+    call_ = other.call_;
+    done_intercepting_ = false;
+    interceptor_methods_ = InterceptorBatchMethodsImpl();
+    return *this;
+  }
+
+  void FillOps(Call* call) override {
+    done_intercepting_ = false;
+    g_core_codegen_interface->grpc_call_ref(call->call());
+    call_ =
+        *call;  // It's fine to create a copy of call since it's just pointers
+
+    if (RunInterceptors()) {
+      ContinueFillOpsAfterInterception();
+    } else {
+      // After the interceptors are run, ContinueFillOpsAfterInterception will
+      // be run
+    }
+  }
+
+  bool FinalizeResult(void** tag, bool* status) override {
+    if (done_intercepting_) {
+      // We have already finished intercepting and filling in the results. This
+      // round trip from the core needed to be made because interceptors were
+      // run
+      *tag = return_tag_;
+      *status = saved_status_;
+      g_core_codegen_interface->grpc_call_unref(call_.call());
+      return true;
+    }
+
+    this->Op1::FinishOp(status);
+    this->Op2::FinishOp(status);
+    this->Op3::FinishOp(status);
+    this->Op4::FinishOp(status);
+    this->Op5::FinishOp(status);
+    this->Op6::FinishOp(status);
+    saved_status_ = *status;
+    if (RunInterceptorsPostRecv()) {
+      *tag = return_tag_;
+      g_core_codegen_interface->grpc_call_unref(call_.call());
+      return true;
+    }
+    // Interceptors are going to be run, so we can't return the tag just yet.
+    // After the interceptors are run, ContinueFinalizeResultAfterInterception
+    return false;
+  }
+
+  void set_output_tag(void* return_tag) { return_tag_ = return_tag; }
+
+  void* core_cq_tag() override { return core_cq_tag_; }
+
+  /// set_core_cq_tag is used to provide a different core CQ tag than "this".
+  /// This is used for callback-based tags, where the core tag is the core
+  /// callback function. It does not change the use or behavior of any other
+  /// function (such as FinalizeResult)
+  void set_core_cq_tag(void* core_cq_tag) { core_cq_tag_ = core_cq_tag; }
+
+  // This will be called while interceptors are run if the RPC is a hijacked
+  // RPC. This should set hijacking state for each of the ops.
+  void SetHijackingState() override {
+    this->Op1::SetHijackingState(&interceptor_methods_);
+    this->Op2::SetHijackingState(&interceptor_methods_);
+    this->Op3::SetHijackingState(&interceptor_methods_);
+    this->Op4::SetHijackingState(&interceptor_methods_);
+    this->Op5::SetHijackingState(&interceptor_methods_);
+    this->Op6::SetHijackingState(&interceptor_methods_);
+  }
+
+  // Should be called after interceptors are done running
+  void ContinueFillOpsAfterInterception() override {
+    static const size_t MAX_OPS = 6;
+    grpc_op ops[MAX_OPS];
+    size_t nops = 0;
+    this->Op1::AddOp(ops, &nops);
+    this->Op2::AddOp(ops, &nops);
+    this->Op3::AddOp(ops, &nops);
+    this->Op4::AddOp(ops, &nops);
+    this->Op5::AddOp(ops, &nops);
+    this->Op6::AddOp(ops, &nops);
+    GPR_CODEGEN_ASSERT(GRPC_CALL_OK ==
+                       g_core_codegen_interface->grpc_call_start_batch(
+                           call_.call(), ops, nops, core_cq_tag(), nullptr));
+  }
+
+  // Should be called after interceptors are done running on the finalize result
+  // path
+  void ContinueFinalizeResultAfterInterception() override {
+    done_intercepting_ = true;
+    GPR_CODEGEN_ASSERT(GRPC_CALL_OK ==
+                       g_core_codegen_interface->grpc_call_start_batch(
+                           call_.call(), nullptr, 0, core_cq_tag(), nullptr));
+  }
+
+ private:
+  // Returns true if no interceptors need to be run
+  bool RunInterceptors() {
+    interceptor_methods_.ClearState();
+    interceptor_methods_.SetCallOpSetInterface(this);
+    interceptor_methods_.SetCall(&call_);
+    this->Op1::SetInterceptionHookPoint(&interceptor_methods_);
+    this->Op2::SetInterceptionHookPoint(&interceptor_methods_);
+    this->Op3::SetInterceptionHookPoint(&interceptor_methods_);
+    this->Op4::SetInterceptionHookPoint(&interceptor_methods_);
+    this->Op5::SetInterceptionHookPoint(&interceptor_methods_);
+    this->Op6::SetInterceptionHookPoint(&interceptor_methods_);
+    return interceptor_methods_.RunInterceptors();
+  }
+  // Returns true if no interceptors need to be run
+  bool RunInterceptorsPostRecv() {
+    // Call and OpSet had already been set on the set state.
+    // SetReverse also clears previously set hook points
+    interceptor_methods_.SetReverse();
+    this->Op1::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    this->Op2::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    this->Op3::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    this->Op4::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    this->Op5::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    this->Op6::SetFinishInterceptionHookPoint(&interceptor_methods_);
+    return interceptor_methods_.RunInterceptors();
+  }
+
+  void* core_cq_tag_;
+  void* return_tag_;
+  Call call_;
+  bool done_intercepting_ = false;
+  InterceptorBatchMethodsImpl interceptor_methods_;
+  bool saved_status_;
+};
+
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_CALL_OP_SET_H
diff --git a/include/grpcpp/impl/codegen/call_op_set_interface.h b/include/grpcpp/impl/codegen/call_op_set_interface.h
new file mode 100644
index 0000000000..3b74566a6d
--- /dev/null
+++ b/include/grpcpp/impl/codegen/call_op_set_interface.h
@@ -0,0 +1,59 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_CALL_OP_SET_INTERFACE_H
+#define GRPCPP_IMPL_CODEGEN_CALL_OP_SET_INTERFACE_H
+
+#include <grpcpp/impl/codegen/completion_queue_tag.h>
+
+namespace grpc {
+namespace internal {
+
+class Call;
+
+/// An abstract collection of call ops, used to generate the
+/// grpc_call_op structure to pass down to the lower layers,
+/// and as it is-a CompletionQueueTag, also massages the final
+/// completion into the correct form for consumption in the C++
+/// API.
+class CallOpSetInterface : public CompletionQueueTag {
+ public:
+  /// Fills in grpc_op, starting from ops[*nops] and moving
+  /// upwards.
+  virtual void FillOps(internal::Call* call) = 0;
+
+  /// Get the tag to be used at the core completion queue. Generally, the
+  /// value of core_cq_tag will be "this". However, it can be overridden if we
+  /// want core to process the tag differently (e.g., as a core callback)
+  virtual void* core_cq_tag() = 0;
+
+  // This will be called while interceptors are run if the RPC is a hijacked
+  // RPC. This should set hijacking state for each of the ops.
+  virtual void SetHijackingState() = 0;
+
+  // Should be called after interceptors are done running
+  virtual void ContinueFillOpsAfterInterception() = 0;
+
+  // Should be called after interceptors are done running on the finalize result
+  // path
+  virtual void ContinueFinalizeResultAfterInterception() = 0;
+};
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_CALL_OP_SET_INTERFACE_H
diff --git a/include/grpcpp/impl/codegen/callback_common.h b/include/grpcpp/impl/codegen/callback_common.h
index a9835973ac..a3c8c41246 100644
--- a/include/grpcpp/impl/codegen/callback_common.h
+++ b/include/grpcpp/impl/codegen/callback_common.h
@@ -32,6 +32,8 @@ namespace grpc {
 namespace internal {
 
 /// An exception-safe way of invoking a user-specified callback function
+// TODO(vjpai): decide whether it is better for this to take a const lvalue
+//              parameter or an rvalue parameter, or if it even matters
 template <class Func, class... Args>
 void CatchingCallback(Func&& func, Args&&... args) {
 #if GRPC_ALLOW_EXCEPTIONS
@@ -45,6 +47,20 @@ void CatchingCallback(Func&& func, Args&&... args) {
 #endif  // GRPC_ALLOW_EXCEPTIONS
 }
 
+template <class ReturnType, class Func, class... Args>
+ReturnType* CatchingReactorCreator(Func&& func, Args&&... args) {
+#if GRPC_ALLOW_EXCEPTIONS
+  try {
+    return func(std::forward<Args>(args)...);
+  } catch (...) {
+    // fail the RPC, don't crash the library
+    return nullptr;
+  }
+#else   // GRPC_ALLOW_EXCEPTIONS
+  return func(std::forward<Args>(args)...);
+#endif  // GRPC_ALLOW_EXCEPTIONS
+}
+
 // The contract on these tags is that they are single-shot. They must be
 // constructed and then fired at exactly one point. There is no expectation
 // that they can be reused without reconstruction.
@@ -94,18 +110,25 @@ class CallbackWithStatusTag
   void Run(bool ok) {
     void* ignored = ops_;
 
-    GPR_CODEGEN_ASSERT(ops_->FinalizeResult(&ignored, &ok));
+    if (!ops_->FinalizeResult(&ignored, &ok)) {
+      // The tag was swallowed
+      return;
+    }
     GPR_CODEGEN_ASSERT(ignored == ops_);
 
     // Last use of func_ or status_, so ok to move them out
-    CatchingCallback(std::move(func_), std::move(status_));
-
+    auto func = std::move(func_);
+    auto status = std::move(status_);
     func_ = nullptr;     // reset to clear this out for sure
     status_ = Status();  // reset to clear this out for sure
+    CatchingCallback(std::move(func), std::move(status));
     g_core_codegen_interface->grpc_call_unref(call_);
   }
 };
 
+/// CallbackWithSuccessTag can be reused multiple times, and will be used in
+/// this fashion for streaming operations. As a result, it shouldn't clear
+/// anything up until its destructor
 class CallbackWithSuccessTag
     : public grpc_experimental_completion_queue_functor {
  public:
@@ -121,13 +144,40 @@ class CallbackWithSuccessTag
   // there are no tests catching the compiler warning.
   static void operator delete(void*, void*) { assert(0); }
 
+  CallbackWithSuccessTag() : call_(nullptr) {}
+
   CallbackWithSuccessTag(grpc_call* call, std::function<void(bool)> f,
-                         CompletionQueueTag* ops)
-      : call_(call), func_(std::move(f)), ops_(ops) {
+                         CompletionQueueTag* ops) {
+    Set(call, f, ops);
+  }
+
+  CallbackWithSuccessTag(const CallbackWithSuccessTag&) = delete;
+  CallbackWithSuccessTag& operator=(const CallbackWithSuccessTag&) = delete;
+
+  ~CallbackWithSuccessTag() { Clear(); }
+
+  // Set can only be called on a default-constructed or Clear'ed tag.
+  // It should never be called on a tag that was constructed with arguments
+  // 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;
     functor_run = &CallbackWithSuccessTag::StaticRun;
   }
 
+  void Clear() {
+    if (call_ != nullptr) {
+      grpc_call* call = call_;
+      call_ = nullptr;
+      func_ = nullptr;
+      g_core_codegen_interface->grpc_call_unref(call);
+    }
+  }
+
   CompletionQueueTag* ops() { return ops_; }
 
   // force_run can not be performed on a tag if operations using this tag
@@ -135,6 +185,9 @@ class CallbackWithSuccessTag
   // that are detected before the operations are internally processed.
   void force_run(bool ok) { Run(ok); }
 
+  /// check if this tag is currently set
+  operator bool() const { return call_ != nullptr; }
+
  private:
   grpc_call* call_;
   std::function<void(bool)> func_;
@@ -146,15 +199,15 @@ class CallbackWithSuccessTag
   }
   void Run(bool ok) {
     void* ignored = ops_;
-    bool new_ok = ok;
-    GPR_CODEGEN_ASSERT(ops_->FinalizeResult(&ignored, &new_ok));
-    GPR_CODEGEN_ASSERT(ignored == ops_);
-
-    // Last use of func_, so ok to move it out for rvalue call above
-    CatchingCallback(std::move(func_), ok);
-
-    func_ = nullptr;  // reset to clear this out for sure
-    g_core_codegen_interface->grpc_call_unref(call_);
+    // Allow a "false" return value from FinalizeResult to silence the
+    // callback, just as it silences a CQ tag in the async cases
+    auto* ops = ops_;
+    bool do_callback = ops_->FinalizeResult(&ignored, &ok);
+    GPR_CODEGEN_ASSERT(ignored == ops);
+
+    if (do_callback) {
+      CatchingCallback(func_, ok);
+    }
   }
 };
 
diff --git a/include/grpcpp/impl/codegen/channel_interface.h b/include/grpcpp/impl/codegen/channel_interface.h
index b257acc1ab..5353f5feaa 100644
--- a/include/grpcpp/impl/codegen/channel_interface.h
+++ b/include/grpcpp/impl/codegen/channel_interface.h
@@ -20,6 +20,7 @@
 #define GRPCPP_IMPL_CODEGEN_CHANNEL_INTERFACE_H
 
 #include <grpc/impl/codegen/connectivity_state.h>
+#include <grpcpp/impl/codegen/call.h>
 #include <grpcpp/impl/codegen/status.h>
 #include <grpcpp/impl/codegen/time.h>
 
@@ -51,6 +52,13 @@ 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
 
 /// Codegen interface for \a grpc::Channel.
@@ -103,11 +111,18 @@ 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>
   friend class ::grpc::internal::CallbackUnaryCallImpl;
   friend class ::grpc::internal::RpcMethod;
+  friend class ::grpc::internal::InterceptedChannel;
   virtual internal::Call CreateCall(const internal::RpcMethod& method,
                                     ClientContext* context,
                                     CompletionQueue* cq) = 0;
@@ -121,11 +136,25 @@ class ChannelInterface {
                                       gpr_timespec deadline) = 0;
 
   // EXPERIMENTAL
+  // This is needed to keep codegen_test_minimal happy. InterceptedChannel needs
+  // to make use of this but can't directly call Channel's implementation
+  // because of the test.
+  // Returns an empty Call object (rather than being pure) since this is a new
+  // method and adding a new pure method to an interface would be a breaking
+  // change (even though this is private and non-API)
+  virtual internal::Call CreateCallInternal(const internal::RpcMethod& method,
+                                            ClientContext* context,
+                                            CompletionQueue* cq,
+                                            size_t interceptor_pos) {
+    return internal::Call();
+  }
+
+  // EXPERIMENTAL
   // A method to get the callbackable completion queue associated with this
   // channel. If the return value is nullptr, this channel doesn't support
   // callback operations.
   // TODO(vjpai): Consider a better default like using a global CQ
-  // Returns nullptr (rather than being pure) since this is a new method
+  // Returns nullptr (rather than being pure) since this is a post-1.0 method
   // and adding a new pure method to an interface would be a breaking change
   // (even though this is private and non-API)
   virtual CompletionQueue* CallbackCQ() { return nullptr; }
diff --git a/include/grpcpp/impl/codegen/client_callback.h b/include/grpcpp/impl/codegen/client_callback.h
index 4d4faea063..66cf9b7754 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>
@@ -77,17 +78,671 @@ class CallbackUnaryCallImpl {
       tag->force_run(s);
       return;
     }
-    ops->SendInitialMetadata(context->send_initial_metadata_,
+    ops->SendInitialMetadata(&context->send_initial_metadata_,
                              context->initial_metadata_flags());
     ops->RecvInitialMetadata(context);
     ops->RecvMessage(result);
     ops->AllowNoMessage();
     ops->ClientSendClose();
     ops->ClientRecvStatus(context, tag->status_ptr());
-    ops->set_cq_tag(tag);
+    ops->set_core_cq_tag(tag);
     call.PerformOps(ops);
   }
 };
+}  // namespace internal
+
+namespace experimental {
+
+// Forward declarations
+template <class Request, class Response>
+class ClientBidiReactor;
+template <class Response>
+class ClientReadReactor;
+template <class Request>
+class ClientWriteReactor;
+
+// NOTE: The streaming objects are not actually implemented in the public API.
+//       These interfaces are provided for mocking only. Typical applications
+//       will interact exclusively with the reactors that they define.
+template <class Request, class Response>
+class ClientCallbackReaderWriter {
+ public:
+  virtual ~ClientCallbackReaderWriter() {}
+  virtual void StartCall() = 0;
+  virtual void Write(const Request* req, WriteOptions options) = 0;
+  virtual void WritesDone() = 0;
+  virtual void Read(Response* resp) = 0;
+
+ protected:
+  void BindReactor(ClientBidiReactor<Request, Response>* reactor) {
+    reactor->BindStream(this);
+  }
+};
+
+template <class Response>
+class ClientCallbackReader {
+ public:
+  virtual ~ClientCallbackReader() {}
+  virtual void StartCall() = 0;
+  virtual void Read(Response* resp) = 0;
+
+ protected:
+  void BindReactor(ClientReadReactor<Response>* reactor) {
+    reactor->BindReader(this);
+  }
+};
+
+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;
+
+ protected:
+  void BindReactor(ClientWriteReactor<Request>* reactor) {
+    reactor->BindWriter(this);
+  }
+};
+
+// 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
+template <class Request, class Response>
+class ClientBidiReactor {
+ public:
+  virtual ~ClientBidiReactor() {}
+  virtual void OnDone(const Status& s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+  virtual void OnWritesDoneDone(bool ok) {}
+
+  void StartCall() { stream_->StartCall(); }
+  void StartRead(Response* resp) { stream_->Read(resp); }
+  void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); }
+  void StartWrite(const Request* req, WriteOptions options) {
+    stream_->Write(req, std::move(options));
+  }
+  void StartWriteLast(const Request* req, WriteOptions options) {
+    StartWrite(req, std::move(options.set_last_message()));
+  }
+  void StartWritesDone() { stream_->WritesDone(); }
+
+ private:
+  friend class ClientCallbackReaderWriter<Request, Response>;
+  void BindStream(ClientCallbackReaderWriter<Request, Response>* stream) {
+    stream_ = stream;
+  }
+  ClientCallbackReaderWriter<Request, Response>* stream_;
+};
+
+template <class Response>
+class ClientReadReactor {
+ public:
+  virtual ~ClientReadReactor() {}
+  virtual void OnDone(const Status& s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+
+  void StartCall() { reader_->StartCall(); }
+  void StartRead(Response* resp) { reader_->Read(resp); }
+
+ private:
+  friend class ClientCallbackReader<Response>;
+  void BindReader(ClientCallbackReader<Response>* reader) { reader_ = reader; }
+  ClientCallbackReader<Response>* reader_;
+};
+
+template <class Request>
+class ClientWriteReactor {
+ public:
+  virtual ~ClientWriteReactor() {}
+  virtual void OnDone(const Status& s) {}
+  virtual void OnReadInitialMetadataDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+  virtual void OnWritesDoneDone(bool ok) {}
+
+  void StartCall() { writer_->StartCall(); }
+  void StartWrite(const Request* req) { StartWrite(req, WriteOptions()); }
+  void StartWrite(const Request* req, WriteOptions options) {
+    writer_->Write(req, std::move(options));
+  }
+  void StartWriteLast(const Request* req, WriteOptions options) {
+    StartWrite(req, std::move(options.set_last_message()));
+  }
+  void StartWritesDone() { writer_->WritesDone(); }
+
+ private:
+  friend class ClientCallbackWriter<Request>;
+  void BindWriter(ClientCallbackWriter<Request>* writer) { writer_ = writer; }
+  ClientCallbackWriter<Request>* writer_;
+};
+
+}  // 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) {
+      Status s = std::move(finish_status_);
+      auto* reactor = reactor_;
+      auto* call = call_.call();
+      this->~ClientCallbackReaderWriterImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+      reactor->OnDone(s);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches, plus any backlog, each with a callback
+    // 1. Send initial metadata (unless corked) + recv initial metadata
+    // 2. Any read backlog
+    // 3. Recv trailing metadata, on_completion callback
+    // 4. Any write backlog
+    // 5. See if the call can finish (if other callbacks were triggered already)
+    started_ = true;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_);
+    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_);
+
+    // 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_);
+    if (read_ops_at_start_) {
+      call_.PerformOps(&read_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_);
+
+    if (write_ops_at_start_) {
+      call_.PerformOps(&write_ops_);
+    }
+
+    if (writes_done_ops_at_start_) {
+      call_.PerformOps(&writes_done_ops_);
+    }
+    MaybeFinish();
+  }
+
+  void Read(Response* msg) override {
+    read_ops_.RecvMessage(msg);
+    callbacks_outstanding_++;
+    if (started_) {
+      call_.PerformOps(&read_ops_);
+    } else {
+      read_ops_at_start_ = true;
+    }
+  }
+
+  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_++;
+    if (started_) {
+      call_.PerformOps(&write_ops_);
+    } else {
+      write_ops_at_start_ = true;
+    }
+  }
+  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_++;
+    if (started_) {
+      call_.PerformOps(&writes_done_ops_);
+    } else {
+      writes_done_ops_at_start_ = true;
+    }
+  }
+
+ private:
+  friend class ClientCallbackReaderWriterFactory<Request, Response>;
+
+  ClientCallbackReaderWriterImpl(
+      Call call, ClientContext* context,
+      ::grpc::experimental::ClientBidiReactor<Request, Response>* reactor)
+      : context_(context),
+        call_(call),
+        reactor_(reactor),
+        start_corked_(context_->initial_metadata_corked_) {
+    this->BindReactor(reactor);
+  }
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientBidiReactor<Request, Response>* 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_;
+  bool write_ops_at_start_{false};
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
+  CallbackWithSuccessTag writes_done_tag_;
+  bool writes_done_ops_at_start_{false};
+
+  CallOpSet<CallOpRecvMessage<Response>> read_ops_;
+  CallbackWithSuccessTag read_tag_;
+  bool read_ops_at_start_{false};
+
+  // Minimum of 3 callbacks to pre-register for StartCall, start, and finish
+  std::atomic_int callbacks_outstanding_{3};
+  bool started_{false};
+};
+
+template <class Request, class Response>
+class ClientCallbackReaderWriterFactory {
+ public:
+  static void Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context,
+      ::grpc::experimental::ClientBidiReactor<Request, Response>* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    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) {
+      Status s = std::move(finish_status_);
+      auto* reactor = reactor_;
+      auto* call = call_.call();
+      this->~ClientCallbackReaderImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+      reactor->OnDone(s);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches, plus any backlog, each with a callback
+    // 1. Send initial metadata (unless corked) + recv initial metadata
+    // 2. Any backlog
+    // 3. Recv trailing metadata, on_completion callback
+    // 4. See if the call can finish (if other callbacks were triggered already)
+    started_ = true;
+
+    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_);
+
+    // 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_);
+    if (read_ops_at_start_) {
+      call_.PerformOps(&read_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_);
+
+    MaybeFinish();
+  }
+
+  void Read(Response* msg) override {
+    read_ops_.RecvMessage(msg);
+    callbacks_outstanding_++;
+    if (started_) {
+      call_.PerformOps(&read_ops_);
+    } else {
+      read_ops_at_start_ = true;
+    }
+  }
+
+ private:
+  friend class ClientCallbackReaderFactory<Response>;
+
+  template <class Request>
+  ClientCallbackReaderImpl(
+      Call call, ClientContext* context, Request* request,
+      ::grpc::experimental::ClientReadReactor<Response>* reactor)
+      : context_(context), call_(call), reactor_(reactor) {
+    this->BindReactor(reactor);
+    // TODO(vjpai): don't assert
+    GPR_CODEGEN_ASSERT(start_ops_.SendMessage(*request).ok());
+    start_ops_.ClientSendClose();
+  }
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientReadReactor<Response>* reactor_;
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage, CallOpClientSendClose,
+            CallOpRecvInitialMetadata>
+      start_ops_;
+  CallbackWithSuccessTag start_tag_;
+
+  CallOpSet<CallOpClientRecvStatus> finish_ops_;
+  CallbackWithSuccessTag finish_tag_;
+  Status finish_status_;
+
+  CallOpSet<CallOpRecvMessage<Response>> read_ops_;
+  CallbackWithSuccessTag read_tag_;
+  bool read_ops_at_start_{false};
+
+  // Minimum of 3 callbacks to pre-register for StartCall, start, and finish
+  std::atomic_int callbacks_outstanding_{3};
+  bool started_{false};
+};
+
+template <class Response>
+class ClientCallbackReaderFactory {
+ public:
+  template <class Request>
+  static void Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context, const Request* request,
+      ::grpc::experimental::ClientReadReactor<Response>* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    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) {
+      Status s = std::move(finish_status_);
+      auto* reactor = reactor_;
+      auto* call = call_.call();
+      this->~ClientCallbackWriterImpl();
+      g_core_codegen_interface->grpc_call_unref(call);
+      reactor->OnDone(s);
+    }
+  }
+
+  void StartCall() override {
+    // This call initiates two batches, plus any backlog, each with a callback
+    // 1. Send initial metadata (unless corked) + recv initial metadata
+    // 2. Recv trailing metadata, on_completion callback
+    // 3. Any backlog
+    // 4. See if the call can finish (if other callbacks were triggered already)
+    started_ = true;
+
+    start_tag_.Set(call_.call(),
+                   [this](bool ok) {
+                     reactor_->OnReadInitialMetadataDone(ok);
+                     MaybeFinish();
+                   },
+                   &start_ops_);
+    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_);
+
+    // 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_);
+
+    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_);
+
+    if (write_ops_at_start_) {
+      call_.PerformOps(&write_ops_);
+    }
+
+    if (writes_done_ops_at_start_) {
+      call_.PerformOps(&writes_done_ops_);
+    }
+
+    MaybeFinish();
+  }
+
+  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_++;
+    if (started_) {
+      call_.PerformOps(&write_ops_);
+    } else {
+      write_ops_at_start_ = true;
+    }
+  }
+  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_++;
+    if (started_) {
+      call_.PerformOps(&writes_done_ops_);
+    } else {
+      writes_done_ops_at_start_ = true;
+    }
+  }
+
+ private:
+  friend class ClientCallbackWriterFactory<Request>;
+
+  template <class Response>
+  ClientCallbackWriterImpl(
+      Call call, ClientContext* context, Response* response,
+      ::grpc::experimental::ClientWriteReactor<Request>* reactor)
+      : context_(context),
+        call_(call),
+        reactor_(reactor),
+        start_corked_(context_->initial_metadata_corked_) {
+    this->BindReactor(reactor);
+    finish_ops_.RecvMessage(response);
+    finish_ops_.AllowNoMessage();
+  }
+
+  ClientContext* context_;
+  Call call_;
+  ::grpc::experimental::ClientWriteReactor<Request>* 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_;
+  bool write_ops_at_start_{false};
+
+  CallOpSet<CallOpSendInitialMetadata, CallOpClientSendClose> writes_done_ops_;
+  CallbackWithSuccessTag writes_done_tag_;
+  bool writes_done_ops_at_start_{false};
+
+  // Minimum of 3 callbacks to pre-register for StartCall, start, and finish
+  std::atomic_int callbacks_outstanding_{3};
+  bool started_{false};
+};
+
+template <class Request>
+class ClientCallbackWriterFactory {
+ public:
+  template <class Response>
+  static void Create(
+      ChannelInterface* channel, const ::grpc::internal::RpcMethod& method,
+      ClientContext* context, Response* response,
+      ::grpc::experimental::ClientWriteReactor<Request>* reactor) {
+    Call call = channel->CreateCall(method, context, channel->CallbackCQ());
+
+    g_core_codegen_interface->grpc_call_ref(call.call());
+    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 24f5c431ce..0a71f3d9b6 100644
--- a/include/grpcpp/impl/codegen/client_context.h
+++ b/include/grpcpp/impl/codegen/client_context.h
@@ -41,10 +41,12 @@
 
 #include <grpc/impl/codegen/compression_types.h>
 #include <grpc/impl/codegen/propagation_bits.h>
+#include <grpcpp/impl/codegen/client_interceptor.h>
 #include <grpcpp/impl/codegen/config.h>
 #include <grpcpp/impl/codegen/core_codegen_interface.h>
 #include <grpcpp/impl/codegen/create_auth_context.h>
 #include <grpcpp/impl/codegen/metadata_map.h>
+#include <grpcpp/impl/codegen/rpc_method.h>
 #include <grpcpp/impl/codegen/security/auth_context.h>
 #include <grpcpp/impl/codegen/slice.h>
 #include <grpcpp/impl/codegen/status.h>
@@ -70,6 +72,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>
@@ -161,6 +169,8 @@ class InteropClientContextInspector;
 /// (see \a grpc::CreateCustomChannel).
 ///
 /// \warning ClientContext instances should \em not be reused across rpcs.
+/// \warning The ClientContext instance used for creating an rpc must remain
+///          alive and valid for the lifetime of the rpc.
 class ClientContext {
  public:
   ClientContext();
@@ -393,6 +403,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) {
@@ -402,6 +418,18 @@ class ClientContext {
   grpc_call* call() const { return call_; }
   void set_call(grpc_call* call, const std::shared_ptr<Channel>& channel);
 
+  experimental::ClientRpcInfo* set_client_rpc_info(
+      const char* method, internal::RpcMethod::RpcType type,
+      grpc::ChannelInterface* channel,
+      const std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>&
+          creators,
+      size_t interceptor_pos) {
+    rpc_info_ = experimental::ClientRpcInfo(this, type, method, channel);
+    rpc_info_.RegisterInterceptors(creators, interceptor_pos);
+    return &rpc_info_;
+  }
+
   uint32_t initial_metadata_flags() const {
     return (idempotent_ ? GRPC_INITIAL_METADATA_IDEMPOTENT_REQUEST : 0) |
            (wait_for_ready_ ? GRPC_INITIAL_METADATA_WAIT_FOR_READY : 0) |
@@ -414,6 +442,8 @@ class ClientContext {
 
   grpc::string authority() { return authority_; }
 
+  void SendCancelToInterceptors();
+
   bool initial_metadata_received_;
   bool wait_for_ready_;
   bool wait_for_ready_explicitly_set_;
@@ -439,6 +469,8 @@ class ClientContext {
   bool initial_metadata_corked_;
 
   grpc::string debug_error_string_;
+
+  experimental::ClientRpcInfo rpc_info_;
 };
 
 }  // namespace grpc
diff --git a/include/grpcpp/impl/codegen/client_interceptor.h b/include/grpcpp/impl/codegen/client_interceptor.h
index f460c5ac0c..2bae11a251 100644
--- a/include/grpcpp/impl/codegen/client_interceptor.h
+++ b/include/grpcpp/impl/codegen/client_interceptor.h
@@ -19,25 +19,144 @@
 #ifndef GRPCPP_IMPL_CODEGEN_CLIENT_INTERCEPTOR_H
 #define GRPCPP_IMPL_CODEGEN_CLIENT_INTERCEPTOR_H
 
+#include <memory>
+#include <vector>
+
 #include <grpcpp/impl/codegen/interceptor.h>
+#include <grpcpp/impl/codegen/rpc_method.h>
+#include <grpcpp/impl/codegen/string_ref.h>
 
 namespace grpc {
-namespace experimental {
-class ClientInterceptor {
- public:
-  virtual ~ClientInterceptor() {}
 
-  virtual void Intercept(InterceptorBatchMethods* methods) = 0;
-};
+class ClientContext;
+class Channel;
+
+namespace internal {
+class InterceptorBatchMethodsImpl;
+}
 
-class ClientRpcInfo {};
+namespace experimental {
+class ClientRpcInfo;
 
 class ClientInterceptorFactoryInterface {
  public:
   virtual ~ClientInterceptorFactoryInterface() {}
-  virtual ClientInterceptor* CreateClientInterceptor(ClientRpcInfo* info) = 0;
+  virtual Interceptor* CreateClientInterceptor(ClientRpcInfo* info) = 0;
+};
+}  // namespace experimental
+
+namespace internal {
+extern experimental::ClientInterceptorFactoryInterface*
+    g_global_client_interceptor_factory;
+}
+
+namespace experimental {
+class ClientRpcInfo {
+ public:
+  // TODO(yashykt): Stop default-constructing ClientRpcInfo and remove UNKNOWN
+  //                from the list of possible Types.
+  enum class Type {
+    UNARY,
+    CLIENT_STREAMING,
+    SERVER_STREAMING,
+    BIDI_STREAMING,
+    UNKNOWN  // UNKNOWN is not API and will be removed later
+  };
+
+  ~ClientRpcInfo(){};
+
+  ClientRpcInfo(const ClientRpcInfo&) = delete;
+  ClientRpcInfo(ClientRpcInfo&&) = default;
+
+  // Getter methods
+  const char* method() const { return method_; }
+  ChannelInterface* channel() { return channel_; }
+  grpc::ClientContext* client_context() { return ctx_; }
+  Type type() const { return type_; }
+
+ private:
+  static_assert(Type::UNARY ==
+                    static_cast<Type>(internal::RpcMethod::NORMAL_RPC),
+                "violated expectation about Type enum");
+  static_assert(Type::CLIENT_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::CLIENT_STREAMING),
+                "violated expectation about Type enum");
+  static_assert(Type::SERVER_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::SERVER_STREAMING),
+                "violated expectation about Type enum");
+  static_assert(Type::BIDI_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::BIDI_STREAMING),
+                "violated expectation about Type enum");
+
+  // Default constructor should only be used by ClientContext
+  ClientRpcInfo() = default;
+
+  // Constructor will only be called from ClientContext
+  ClientRpcInfo(grpc::ClientContext* ctx, internal::RpcMethod::RpcType type,
+                const char* method, grpc::ChannelInterface* channel)
+      : ctx_(ctx),
+        type_(static_cast<Type>(type)),
+        method_(method),
+        channel_(channel) {}
+
+  // Move assignment should only be used by ClientContext
+  // TODO(yashykt): Delete move assignment
+  ClientRpcInfo& operator=(ClientRpcInfo&&) = default;
+
+  // Runs interceptor at pos \a pos.
+  void RunInterceptor(
+      experimental::InterceptorBatchMethods* interceptor_methods, size_t pos) {
+    GPR_CODEGEN_ASSERT(pos < interceptors_.size());
+    interceptors_[pos]->Intercept(interceptor_methods);
+  }
+
+  void RegisterInterceptors(
+      const std::vector<std::unique_ptr<
+          experimental::ClientInterceptorFactoryInterface>>& creators,
+      size_t interceptor_pos) {
+    if (interceptor_pos > creators.size()) {
+      // No interceptors to register
+      return;
+    }
+    for (auto it = creators.begin() + interceptor_pos; it != creators.end();
+         ++it) {
+      interceptors_.push_back(std::unique_ptr<experimental::Interceptor>(
+          (*it)->CreateClientInterceptor(this)));
+    }
+    if (internal::g_global_client_interceptor_factory != nullptr) {
+      interceptors_.push_back(std::unique_ptr<experimental::Interceptor>(
+          internal::g_global_client_interceptor_factory
+              ->CreateClientInterceptor(this)));
+    }
+  }
+
+  grpc::ClientContext* ctx_ = nullptr;
+  // TODO(yashykt): make type_ const once move-assignment is deleted
+  Type type_{Type::UNKNOWN};
+  const char* method_ = nullptr;
+  grpc::ChannelInterface* channel_ = nullptr;
+  std::vector<std::unique_ptr<experimental::Interceptor>> interceptors_;
+  bool hijacked_ = false;
+  size_t hijacked_interceptor_ = 0;
+
+  friend class internal::InterceptorBatchMethodsImpl;
+  friend class grpc::ClientContext;
 };
 
+// PLEASE DO NOT USE THIS. ALWAYS PREFER PER CHANNEL INTERCEPTORS OVER A GLOBAL
+// INTERCEPTOR. IF USAGE IS ABSOLUTELY NECESSARY, PLEASE READ THE SAFETY NOTES.
+// Registers a global client interceptor factory object, which is used for all
+// RPCs made in this process.  If the argument is nullptr, the global
+// interceptor factory is deregistered. The application is responsible for
+// maintaining the life of the object while gRPC operations are in progress. It
+// is unsafe to try to register/deregister if any gRPC operation is in progress.
+// For safety, it is in the best interests of the developer to register the
+// global interceptor factory once at the start of the process before any gRPC
+// operations have begun. Deregistration is optional since gRPC does not
+// maintain any references to the object.
+void RegisterGlobalClientInterceptorFactory(
+    ClientInterceptorFactoryInterface* factory);
+
 }  // namespace experimental
 }  // namespace grpc
 
diff --git a/include/grpcpp/impl/codegen/client_unary_call.h b/include/grpcpp/impl/codegen/client_unary_call.h
index e4e8364e07..5151839412 100644
--- a/include/grpcpp/impl/codegen/client_unary_call.h
+++ b/include/grpcpp/impl/codegen/client_unary_call.h
@@ -61,7 +61,7 @@ class BlockingUnaryCallImpl {
     if (!status_.ok()) {
       return;
     }
-    ops.SendInitialMetadata(context->send_initial_metadata_,
+    ops.SendInitialMetadata(&context->send_initial_metadata_,
                             context->initial_metadata_flags());
     ops.RecvInitialMetadata(context);
     ops.RecvMessage(result);
@@ -69,13 +69,17 @@ class BlockingUnaryCallImpl {
     ops.ClientSendClose();
     ops.ClientRecvStatus(context, &status_);
     call.PerformOps(&ops);
-    if (cq.Pluck(&ops)) {
-      if (!ops.got_message && status_.ok()) {
-        status_ = Status(StatusCode::UNIMPLEMENTED,
-                         "No message returned for unary request");
-      }
-    } else {
-      GPR_CODEGEN_ASSERT(!status_.ok());
+    cq.Pluck(&ops);
+    // Some of the ops might fail. If the ops fail in the core layer, status
+    // would reflect the error. But, if the ops fail in the C++ layer, the
+    // status would still be the same as the one returned by gRPC Core. This can
+    // happen if deserialization of the message fails.
+    // TODO(yashykt): If deserialization fails, but the status received is OK,
+    // then it might be a good idea to change the status to something better
+    // than StatusCode::UNIMPLEMENTED to reflect this.
+    if (!ops.got_message && status_.ok()) {
+      status_ = Status(StatusCode::UNIMPLEMENTED,
+                       "No message returned for unary request");
     }
   }
   Status status() { return status_; }
diff --git a/include/grpcpp/impl/codegen/completion_queue.h b/include/grpcpp/impl/codegen/completion_queue.h
index a62d48abe7..fb38788f7d 100644
--- a/include/grpcpp/impl/codegen/completion_queue.h
+++ b/include/grpcpp/impl/codegen/completion_queue.h
@@ -122,8 +122,8 @@ class CompletionQueue : private GrpcLibraryCodegen {
   /// Read from the queue, blocking until an event is available or the queue is
   /// shutting down.
   ///
-  /// \param tag[out] Updated to point to the read event's tag.
-  /// \param ok[out] true if read a successful event, false otherwise.
+  /// \param tag [out] Updated to point to the read event's tag.
+  /// \param ok [out] true if read a successful event, false otherwise.
   ///
   /// Note that each tag sent to the completion queue (through RPC operations
   /// or alarms) will be delivered out of the completion queue by a call to
@@ -178,10 +178,10 @@ class CompletionQueue : private GrpcLibraryCodegen {
   /// within the \a deadline).  A \a tag points to an arbitrary location usually
   /// employed to uniquely identify an event.
   ///
-  /// \param tag[out] Upon sucess, updated to point to the event's tag.
-  /// \param ok[out] Upon sucess, true if a successful event, false otherwise
+  /// \param tag [out] Upon sucess, updated to point to the event's tag.
+  /// \param ok [out] Upon sucess, true if a successful event, false otherwise
   ///        See documentation for CompletionQueue::Next for explanation of ok
-  /// \param deadline[in] How long to block in wait for an event.
+  /// \param deadline [in] How long to block in wait for an event.
   ///
   /// \return The type of event read.
   template <typename T>
@@ -197,10 +197,11 @@ class CompletionQueue : private GrpcLibraryCodegen {
   /// within the \a deadline).  A \a tag points to an arbitrary location usually
   /// employed to uniquely identify an event.
   ///
-  /// \param F[in] Function to execute before calling AsyncNext on this queue.
-  /// \param tag[out] Upon sucess, updated to point to the event's tag.
-  /// \param ok[out] Upon sucess, true if read a regular event, false otherwise.
-  /// \param deadline[in] How long to block in wait for an event.
+  /// \param f [in] Function to execute before calling AsyncNext on this queue.
+  /// \param tag [out] Upon sucess, updated to point to the event's tag.
+  /// \param ok [out] Upon sucess, true if read a regular event, false
+  /// otherwise.
+  /// \param deadline [in] How long to block in wait for an event.
   ///
   /// \return The type of event read.
   template <typename T, typename F>
@@ -299,14 +300,16 @@ class CompletionQueue : private GrpcLibraryCodegen {
   bool Pluck(internal::CompletionQueueTag* tag) {
     auto deadline =
         g_core_codegen_interface->gpr_inf_future(GPR_CLOCK_REALTIME);
-    auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
-        cq_, tag, deadline, nullptr);
-    bool ok = ev.success != 0;
-    void* ignored = tag;
-    GPR_CODEGEN_ASSERT(tag->FinalizeResult(&ignored, &ok));
-    GPR_CODEGEN_ASSERT(ignored == tag);
-    // Ignore mutations by FinalizeResult: Pluck returns the C API status
-    return ev.success != 0;
+    while (true) {
+      auto ev = g_core_codegen_interface->grpc_completion_queue_pluck(
+          cq_, tag, deadline, nullptr);
+      bool ok = ev.success != 0;
+      void* ignored = tag;
+      if (tag->FinalizeResult(&ignored, &ok)) {
+        GPR_CODEGEN_ASSERT(ignored == tag);
+        return ok;
+      }
+    }
   }
 
   /// Performs a single polling pluck on \a tag.
@@ -376,13 +379,18 @@ class ServerCompletionQueue : public CompletionQueue {
   ServerCompletionQueue() : polling_type_(GRPC_CQ_DEFAULT_POLLING) {}
 
  private:
-  /// \param is_frequently_polled Informs the GRPC library about whether the
-  /// server completion queue would be actively polled (by calling Next() or
-  /// AsyncNext()). By default all server completion queues are assumed to be
-  /// frequently polled.
-  ServerCompletionQueue(grpc_cq_polling_type polling_type)
+  /// \param completion_type indicates whether this is a NEXT or CALLBACK
+  /// completion queue.
+  /// \param polling_type Informs the GRPC library about the type of polling
+  /// allowed on this completion queue. See grpc_cq_polling_type's description
+  /// in grpc_types.h for more details.
+  /// \param shutdown_cb is the shutdown callback used for CALLBACK api queues
+  ServerCompletionQueue(grpc_cq_completion_type completion_type,
+                        grpc_cq_polling_type polling_type,
+                        grpc_experimental_completion_queue_functor* shutdown_cb)
       : CompletionQueue(grpc_completion_queue_attributes{
-            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type, nullptr}),
+            GRPC_CQ_CURRENT_VERSION, completion_type, polling_type,
+            shutdown_cb}),
         polling_type_(polling_type) {}
 
   grpc_cq_polling_type polling_type_;
diff --git a/include/grpcpp/impl/codegen/core_codegen.h b/include/grpcpp/impl/codegen/core_codegen.h
index e9df96bf04..6ef184d01a 100644
--- a/include/grpcpp/impl/codegen/core_codegen.h
+++ b/include/grpcpp/impl/codegen/core_codegen.h
@@ -63,6 +63,9 @@ class CoreCodegen final : public CoreCodegenInterface {
   void gpr_cv_signal(gpr_cv* cv) override;
   void gpr_cv_broadcast(gpr_cv* cv) override;
 
+  grpc_call_error grpc_call_start_batch(grpc_call* call, const grpc_op* ops,
+                                        size_t nops, void* tag,
+                                        void* reserved) override;
   grpc_call_error grpc_call_cancel_with_status(grpc_call* call,
                                                grpc_status_code status,
                                                const char* description,
diff --git a/include/grpcpp/impl/codegen/core_codegen_interface.h b/include/grpcpp/impl/codegen/core_codegen_interface.h
index 1167a188a2..20a5b3300c 100644
--- a/include/grpcpp/impl/codegen/core_codegen_interface.h
+++ b/include/grpcpp/impl/codegen/core_codegen_interface.h
@@ -100,6 +100,9 @@ class CoreCodegenInterface {
   virtual grpc_slice grpc_slice_new_with_len(void* p, size_t len,
                                              void (*destroy)(void*,
                                                              size_t)) = 0;
+  virtual grpc_call_error grpc_call_start_batch(grpc_call* call,
+                                                const grpc_op* ops, size_t nops,
+                                                void* tag, void* reserved) = 0;
   virtual grpc_call_error grpc_call_cancel_with_status(grpc_call* call,
                                                        grpc_status_code status,
                                                        const char* description,
@@ -142,6 +145,15 @@ extern CoreCodegenInterface* g_core_codegen_interface;
     }                                                                      \
   } while (0)
 
+/// Codegen specific version of \a GPR_DEBUG_ASSERT.
+#ifndef NDEBUG
+#define GPR_CODEGEN_DEBUG_ASSERT(x) GPR_CODEGEN_ASSERT(x)
+#else
+#define GPR_CODEGEN_DEBUG_ASSERT(x) \
+  do {                              \
+  } while (0)
+#endif
+
 }  // namespace grpc
 
 #endif  // GRPCPP_IMPL_CODEGEN_CORE_CODEGEN_INTERFACE_H
diff --git a/include/grpcpp/impl/codegen/intercepted_channel.h b/include/grpcpp/impl/codegen/intercepted_channel.h
new file mode 100644
index 0000000000..5255a6d147
--- /dev/null
+++ b/include/grpcpp/impl/codegen/intercepted_channel.h
@@ -0,0 +1,80 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_INTERCEPTED_CHANNEL_H
+#define GRPCPP_IMPL_CODEGEN_INTERCEPTED_CHANNEL_H
+
+#include <grpcpp/impl/codegen/channel_interface.h>
+
+namespace grpc {
+
+namespace internal {
+
+class InterceptorBatchMethodsImpl;
+
+/// An InterceptedChannel is available to client Interceptors. An
+/// InterceptedChannel is unique to an interceptor, and when an RPC is started
+/// on this channel, only those interceptors that come after this interceptor
+/// see the RPC.
+class InterceptedChannel : public ChannelInterface {
+ public:
+  virtual ~InterceptedChannel() { channel_ = nullptr; }
+
+  /// Get the current channel state. If the channel is in IDLE and
+  /// \a try_to_connect is set to true, try to connect.
+  grpc_connectivity_state GetState(bool try_to_connect) override {
+    return channel_->GetState(try_to_connect);
+  }
+
+ private:
+  InterceptedChannel(ChannelInterface* channel, size_t pos)
+      : channel_(channel), interceptor_pos_(pos) {}
+
+  Call CreateCall(const RpcMethod& method, ClientContext* context,
+                  CompletionQueue* cq) override {
+    return channel_->CreateCallInternal(method, context, cq, interceptor_pos_);
+  }
+
+  void PerformOpsOnCall(CallOpSetInterface* ops, Call* call) override {
+    return channel_->PerformOpsOnCall(ops, call);
+  }
+  void* RegisterMethod(const char* method) override {
+    return channel_->RegisterMethod(method);
+  }
+
+  void NotifyOnStateChangeImpl(grpc_connectivity_state last_observed,
+                               gpr_timespec deadline, CompletionQueue* cq,
+                               void* tag) override {
+    return channel_->NotifyOnStateChangeImpl(last_observed, deadline, cq, tag);
+  }
+  bool WaitForStateChangeImpl(grpc_connectivity_state last_observed,
+                              gpr_timespec deadline) override {
+    return channel_->WaitForStateChangeImpl(last_observed, deadline);
+  }
+
+  CompletionQueue* CallbackCQ() override { return channel_->CallbackCQ(); }
+
+  ChannelInterface* channel_;
+  size_t interceptor_pos_;
+
+  friend class InterceptorBatchMethodsImpl;
+};
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_INTERCEPTED_CHANNEL_H
diff --git a/include/grpcpp/impl/codegen/interceptor.h b/include/grpcpp/impl/codegen/interceptor.h
index 6402a3a946..e449e44a23 100644
--- a/include/grpcpp/impl/codegen/interceptor.h
+++ b/include/grpcpp/impl/codegen/interceptor.h
@@ -19,7 +19,17 @@
 #ifndef GRPCPP_IMPL_CODEGEN_INTERCEPTOR_H
 #define GRPCPP_IMPL_CODEGEN_INTERCEPTOR_H
 
+#include <grpc/impl/codegen/grpc_types.h>
+#include <grpcpp/impl/codegen/byte_buffer.h>
+#include <grpcpp/impl/codegen/config.h>
+#include <grpcpp/impl/codegen/core_codegen_interface.h>
+#include <grpcpp/impl/codegen/metadata_map.h>
+
 namespace grpc {
+
+class ChannelInterface;
+class Status;
+
 namespace experimental {
 class InterceptedMessage {
  public:
@@ -35,6 +45,7 @@ enum class InterceptionHookPoints {
   PRE_SEND_INITIAL_METADATA,
   PRE_SEND_MESSAGE,
   PRE_SEND_STATUS /* server only */,
+  PRE_SEND_CLOSE /* client only */,
   /* The following three are for hijacked clients only and can only be
      registered by the global interceptor */
   PRE_RECV_INITIAL_METADATA,
@@ -45,22 +56,77 @@ enum class InterceptionHookPoints {
   POST_RECV_MESSAGE,
   POST_RECV_STATUS /* client only */,
   POST_RECV_CLOSE /* server only */,
+  /* This is a special hook point available to both clients and servers when
+     TryCancel() is performed.
+     - No other hook points will be present along with this.
+     - It is illegal for an interceptor to block/delay this operation.
+     - ALL interceptors see this hook point irrespective of whether the RPC was
+     hijacked or not. */
+  PRE_SEND_CANCEL,
   NUM_INTERCEPTION_HOOKS
 };
 
 class InterceptorBatchMethods {
  public:
-  virtual ~InterceptorBatchMethods();
+  virtual ~InterceptorBatchMethods(){};
   // Queries to check whether the current batch has an interception hook point
   // of type \a type
   virtual bool QueryInterceptionHookPoint(InterceptionHookPoints type) = 0;
   // Calling this will signal that the interceptor is done intercepting the
-  // current batch of the RPC
+  // current batch of the RPC.
+  // Proceed is a no-op if the batch contains PRE_SEND_CANCEL. Simply returning
+  // from the Intercept method does the job of continuing the RPC in this case.
   virtual void Proceed() = 0;
   // Calling this indicates that the interceptor has hijacked the RPC (only
   // valid if the batch contains send_initial_metadata on the client side)
   virtual void Hijack() = 0;
+
+  // Returns a modifable ByteBuffer holding serialized form of the message to be
+  // sent
+  virtual ByteBuffer* GetSendMessage() = 0;
+
+  // Returns a modifiable multimap of the initial metadata to be sent
+  virtual std::multimap<grpc::string, grpc::string>*
+  GetSendInitialMetadata() = 0;
+
+  // Returns the status to be sent
+  virtual Status GetSendStatus() = 0;
+
+  // Modifies the status with \a status
+  virtual void ModifySendStatus(const Status& status) = 0;
+
+  // Returns a modifiable multimap of the trailing metadata to be sent
+  virtual std::multimap<grpc::string, grpc::string>*
+  GetSendTrailingMetadata() = 0;
+
+  // Returns a pointer to the modifiable received message. Note that the message
+  // is already deserialized
+  virtual void* GetRecvMessage() = 0;
+
+  // Returns a modifiable multimap of the received initial metadata
+  virtual std::multimap<grpc::string_ref, grpc::string_ref>*
+  GetRecvInitialMetadata() = 0;
+
+  // Returns a modifiable view of the received status
+  virtual Status* GetRecvStatus() = 0;
+
+  // Returns a modifiable multimap of the received trailing metadata
+  virtual std::multimap<grpc::string_ref, grpc::string_ref>*
+  GetRecvTrailingMetadata() = 0;
+
+  // Gets an intercepted channel. When a call is started on this interceptor,
+  // only interceptors after the current interceptor are created from the
+  // factory objects registered with the channel.
+  virtual std::unique_ptr<ChannelInterface> GetInterceptedChannel() = 0;
 };
+
+class Interceptor {
+ public:
+  virtual ~Interceptor() {}
+
+  virtual void Intercept(InterceptorBatchMethods* methods) = 0;
+};
+
 }  // namespace experimental
 }  // namespace grpc
 
diff --git a/include/grpcpp/impl/codegen/interceptor_common.h b/include/grpcpp/impl/codegen/interceptor_common.h
new file mode 100644
index 0000000000..d0aa23cb0a
--- /dev/null
+++ b/include/grpcpp/impl/codegen/interceptor_common.h
@@ -0,0 +1,458 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_INTERCEPTOR_COMMON_H
+#define GRPCPP_IMPL_CODEGEN_INTERCEPTOR_COMMON_H
+
+#include <array>
+#include <functional>
+
+#include <grpcpp/impl/codegen/call.h>
+#include <grpcpp/impl/codegen/call_op_set_interface.h>
+#include <grpcpp/impl/codegen/client_interceptor.h>
+#include <grpcpp/impl/codegen/intercepted_channel.h>
+#include <grpcpp/impl/codegen/server_interceptor.h>
+
+#include <grpc/impl/codegen/grpc_types.h>
+
+namespace grpc {
+namespace internal {
+
+class InterceptorBatchMethodsImpl
+    : public experimental::InterceptorBatchMethods {
+ public:
+  InterceptorBatchMethodsImpl() {
+    for (auto i = static_cast<experimental::InterceptionHookPoints>(0);
+         i < experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS;
+         i = static_cast<experimental::InterceptionHookPoints>(
+             static_cast<size_t>(i) + 1)) {
+      hooks_[static_cast<size_t>(i)] = false;
+    }
+  }
+
+  ~InterceptorBatchMethodsImpl() {}
+
+  bool QueryInterceptionHookPoint(
+      experimental::InterceptionHookPoints type) override {
+    return hooks_[static_cast<size_t>(type)];
+  }
+
+  void Proceed() override {
+    if (call_->client_rpc_info() != nullptr) {
+      return ProceedClient();
+    }
+    GPR_CODEGEN_ASSERT(call_->server_rpc_info() != nullptr);
+    ProceedServer();
+  }
+
+  void Hijack() override {
+    // Only the client can hijack when sending down initial metadata
+    GPR_CODEGEN_ASSERT(!reverse_ && ops_ != nullptr &&
+                       call_->client_rpc_info() != nullptr);
+    // It is illegal to call Hijack twice
+    GPR_CODEGEN_ASSERT(!ran_hijacking_interceptor_);
+    auto* rpc_info = call_->client_rpc_info();
+    rpc_info->hijacked_ = true;
+    rpc_info->hijacked_interceptor_ = current_interceptor_index_;
+    ClearHookPoints();
+    ops_->SetHijackingState();
+    ran_hijacking_interceptor_ = true;
+    rpc_info->RunInterceptor(this, current_interceptor_index_);
+  }
+
+  void AddInterceptionHookPoint(experimental::InterceptionHookPoints type) {
+    hooks_[static_cast<size_t>(type)] = true;
+  }
+
+  ByteBuffer* GetSendMessage() override { return send_message_; }
+
+  std::multimap<grpc::string, grpc::string>* GetSendInitialMetadata() override {
+    return send_initial_metadata_;
+  }
+
+  Status GetSendStatus() override {
+    return Status(static_cast<StatusCode>(*code_), *error_message_,
+                  *error_details_);
+  }
+
+  void ModifySendStatus(const Status& status) override {
+    *code_ = static_cast<grpc_status_code>(status.error_code());
+    *error_details_ = status.error_details();
+    *error_message_ = status.error_message();
+  }
+
+  std::multimap<grpc::string, grpc::string>* GetSendTrailingMetadata()
+      override {
+    return send_trailing_metadata_;
+  }
+
+  void* GetRecvMessage() override { return recv_message_; }
+
+  std::multimap<grpc::string_ref, grpc::string_ref>* GetRecvInitialMetadata()
+      override {
+    return recv_initial_metadata_->map();
+  }
+
+  Status* GetRecvStatus() override { return recv_status_; }
+
+  std::multimap<grpc::string_ref, grpc::string_ref>* GetRecvTrailingMetadata()
+      override {
+    return recv_trailing_metadata_->map();
+  }
+
+  void SetSendMessage(ByteBuffer* buf) { send_message_ = buf; }
+
+  void SetSendInitialMetadata(
+      std::multimap<grpc::string, grpc::string>* metadata) {
+    send_initial_metadata_ = metadata;
+  }
+
+  void SetSendStatus(grpc_status_code* code, grpc::string* error_details,
+                     grpc::string* error_message) {
+    code_ = code;
+    error_details_ = error_details;
+    error_message_ = error_message;
+  }
+
+  void SetSendTrailingMetadata(
+      std::multimap<grpc::string, grpc::string>* metadata) {
+    send_trailing_metadata_ = metadata;
+  }
+
+  void SetRecvMessage(void* message) { recv_message_ = message; }
+
+  void SetRecvInitialMetadata(MetadataMap* map) {
+    recv_initial_metadata_ = map;
+  }
+
+  void SetRecvStatus(Status* status) { recv_status_ = status; }
+
+  void SetRecvTrailingMetadata(MetadataMap* map) {
+    recv_trailing_metadata_ = map;
+  }
+
+  std::unique_ptr<ChannelInterface> GetInterceptedChannel() override {
+    auto* info = call_->client_rpc_info();
+    if (info == nullptr) {
+      return std::unique_ptr<ChannelInterface>(nullptr);
+    }
+    // The intercepted channel starts from the interceptor just after the
+    // current interceptor
+    return std::unique_ptr<ChannelInterface>(new InterceptedChannel(
+        info->channel(), current_interceptor_index_ + 1));
+  }
+
+  // Clears all state
+  void ClearState() {
+    reverse_ = false;
+    ran_hijacking_interceptor_ = false;
+    ClearHookPoints();
+  }
+
+  // Prepares for Post_recv operations
+  void SetReverse() {
+    reverse_ = true;
+    ran_hijacking_interceptor_ = false;
+    ClearHookPoints();
+  }
+
+  // This needs to be set before interceptors are run
+  void SetCall(Call* call) { call_ = call; }
+
+  // This needs to be set before interceptors are run using RunInterceptors().
+  // Alternatively, RunInterceptors(std::function<void(void)> f) can be used.
+  void SetCallOpSetInterface(CallOpSetInterface* ops) { ops_ = ops; }
+
+  // Returns true if no interceptors are run. This should be used only by
+  // subclasses of CallOpSetInterface. SetCall and SetCallOpSetInterface should
+  // have been called before this. After all the interceptors are done running,
+  // either ContinueFillOpsAfterInterception or
+  // ContinueFinalizeOpsAfterInterception will be called. Note that neither of
+  // them is invoked if there were no interceptors registered.
+  bool RunInterceptors() {
+    GPR_CODEGEN_ASSERT(ops_);
+    auto* client_rpc_info = call_->client_rpc_info();
+    if (client_rpc_info != nullptr) {
+      if (client_rpc_info->interceptors_.size() == 0) {
+        return true;
+      } else {
+        RunClientInterceptors();
+        return false;
+      }
+    }
+
+    auto* server_rpc_info = call_->server_rpc_info();
+    if (server_rpc_info == nullptr ||
+        server_rpc_info->interceptors_.size() == 0) {
+      return true;
+    }
+    RunServerInterceptors();
+    return false;
+  }
+
+  // Returns true if no interceptors are run. Returns false otherwise if there
+  // are interceptors registered. After the interceptors are done running \a f
+  // will be invoked. This is to be used only by BaseAsyncRequest and
+  // SyncRequest.
+  bool RunInterceptors(std::function<void(void)> f) {
+    // This is used only by the server for initial call request
+    GPR_CODEGEN_ASSERT(reverse_ == true);
+    GPR_CODEGEN_ASSERT(call_->client_rpc_info() == nullptr);
+    auto* server_rpc_info = call_->server_rpc_info();
+    if (server_rpc_info == nullptr ||
+        server_rpc_info->interceptors_.size() == 0) {
+      return true;
+    }
+    callback_ = std::move(f);
+    RunServerInterceptors();
+    return false;
+  }
+
+ private:
+  void RunClientInterceptors() {
+    auto* rpc_info = call_->client_rpc_info();
+    if (!reverse_) {
+      current_interceptor_index_ = 0;
+    } else {
+      if (rpc_info->hijacked_) {
+        current_interceptor_index_ = rpc_info->hijacked_interceptor_;
+      } else {
+        current_interceptor_index_ = rpc_info->interceptors_.size() - 1;
+      }
+    }
+    rpc_info->RunInterceptor(this, current_interceptor_index_);
+  }
+
+  void RunServerInterceptors() {
+    auto* rpc_info = call_->server_rpc_info();
+    if (!reverse_) {
+      current_interceptor_index_ = 0;
+    } else {
+      current_interceptor_index_ = rpc_info->interceptors_.size() - 1;
+    }
+    rpc_info->RunInterceptor(this, current_interceptor_index_);
+  }
+
+  void ProceedClient() {
+    auto* rpc_info = call_->client_rpc_info();
+    if (rpc_info->hijacked_ && !reverse_ &&
+        current_interceptor_index_ == rpc_info->hijacked_interceptor_ &&
+        !ran_hijacking_interceptor_) {
+      // We now need to provide hijacked recv ops to this interceptor
+      ClearHookPoints();
+      ops_->SetHijackingState();
+      ran_hijacking_interceptor_ = true;
+      rpc_info->RunInterceptor(this, current_interceptor_index_);
+      return;
+    }
+    if (!reverse_) {
+      current_interceptor_index_++;
+      // We are going down the stack of interceptors
+      if (current_interceptor_index_ < rpc_info->interceptors_.size()) {
+        if (rpc_info->hijacked_ &&
+            current_interceptor_index_ > rpc_info->hijacked_interceptor_) {
+          // This is a hijacked RPC and we are done with hijacking
+          ops_->ContinueFillOpsAfterInterception();
+        } else {
+          rpc_info->RunInterceptor(this, current_interceptor_index_);
+        }
+      } else {
+        // we are done running all the interceptors without any hijacking
+        ops_->ContinueFillOpsAfterInterception();
+      }
+    } else {
+      // We are going up the stack of interceptors
+      if (current_interceptor_index_ > 0) {
+        // Continue running interceptors
+        current_interceptor_index_--;
+        rpc_info->RunInterceptor(this, current_interceptor_index_);
+      } else {
+        // we are done running all the interceptors without any hijacking
+        ops_->ContinueFinalizeResultAfterInterception();
+      }
+    }
+  }
+
+  void ProceedServer() {
+    auto* rpc_info = call_->server_rpc_info();
+    if (!reverse_) {
+      current_interceptor_index_++;
+      if (current_interceptor_index_ < rpc_info->interceptors_.size()) {
+        return rpc_info->RunInterceptor(this, current_interceptor_index_);
+      } else if (ops_) {
+        return ops_->ContinueFillOpsAfterInterception();
+      }
+    } else {
+      // We are going up the stack of interceptors
+      if (current_interceptor_index_ > 0) {
+        // Continue running interceptors
+        current_interceptor_index_--;
+        return rpc_info->RunInterceptor(this, current_interceptor_index_);
+      } else if (ops_) {
+        return ops_->ContinueFinalizeResultAfterInterception();
+      }
+    }
+    GPR_CODEGEN_ASSERT(callback_);
+    callback_();
+  }
+
+  void ClearHookPoints() {
+    for (auto i = static_cast<experimental::InterceptionHookPoints>(0);
+         i < experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS;
+         i = static_cast<experimental::InterceptionHookPoints>(
+             static_cast<size_t>(i) + 1)) {
+      hooks_[static_cast<size_t>(i)] = false;
+    }
+  }
+
+  std::array<bool,
+             static_cast<size_t>(
+                 experimental::InterceptionHookPoints::NUM_INTERCEPTION_HOOKS)>
+      hooks_;
+
+  size_t current_interceptor_index_ = 0;  // Current iterator
+  bool reverse_ = false;
+  bool ran_hijacking_interceptor_ = false;
+  Call* call_ = nullptr;  // The Call object is present along with CallOpSet
+                          // object/callback
+  CallOpSetInterface* ops_ = nullptr;
+  std::function<void(void)> callback_;
+
+  ByteBuffer* send_message_ = nullptr;
+
+  std::multimap<grpc::string, grpc::string>* send_initial_metadata_;
+
+  grpc_status_code* code_ = nullptr;
+  grpc::string* error_details_ = nullptr;
+  grpc::string* error_message_ = nullptr;
+  Status send_status_;
+
+  std::multimap<grpc::string, grpc::string>* send_trailing_metadata_ = nullptr;
+
+  void* recv_message_ = nullptr;
+
+  MetadataMap* recv_initial_metadata_ = nullptr;
+
+  Status* recv_status_ = nullptr;
+
+  MetadataMap* recv_trailing_metadata_ = nullptr;
+};
+
+// A special implementation of InterceptorBatchMethods to send a Cancel
+// notification down the interceptor stack
+class CancelInterceptorBatchMethods
+    : public experimental::InterceptorBatchMethods {
+ public:
+  bool QueryInterceptionHookPoint(
+      experimental::InterceptionHookPoints type) override {
+    if (type == experimental::InterceptionHookPoints::PRE_SEND_CANCEL) {
+      return true;
+    } else {
+      return false;
+    }
+  }
+
+  void Proceed() override {
+    // This is a no-op. For actual continuation of the RPC simply needs to
+    // return from the Intercept method
+  }
+
+  void Hijack() override {
+    // Only the client can hijack when sending down initial metadata
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call Hijack on a method which has a "
+                       "Cancel notification");
+  }
+
+  ByteBuffer* GetSendMessage() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetSendMessage on a method which "
+                       "has a Cancel notification");
+    return nullptr;
+  }
+
+  std::multimap<grpc::string, grpc::string>* GetSendInitialMetadata() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetSendInitialMetadata on a "
+                       "method which has a Cancel notification");
+    return nullptr;
+  }
+
+  Status GetSendStatus() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetSendStatus on a method which "
+                       "has a Cancel notification");
+    return Status();
+  }
+
+  void ModifySendStatus(const Status& status) override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call ModifySendStatus on a method "
+                       "which has a Cancel notification");
+    return;
+  }
+
+  std::multimap<grpc::string, grpc::string>* GetSendTrailingMetadata()
+      override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetSendTrailingMetadata on a "
+                       "method which has a Cancel notification");
+    return nullptr;
+  }
+
+  void* GetRecvMessage() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetRecvMessage on a method which "
+                       "has a Cancel notification");
+    return nullptr;
+  }
+
+  std::multimap<grpc::string_ref, grpc::string_ref>* GetRecvInitialMetadata()
+      override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetRecvInitialMetadata on a "
+                       "method which has a Cancel notification");
+    return nullptr;
+  }
+
+  Status* GetRecvStatus() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetRecvStatus on a method which "
+                       "has a Cancel notification");
+    return nullptr;
+  }
+
+  std::multimap<grpc::string_ref, grpc::string_ref>* GetRecvTrailingMetadata()
+      override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetRecvTrailingMetadata on a "
+                       "method which has a Cancel notification");
+    return nullptr;
+  }
+
+  std::unique_ptr<ChannelInterface> GetInterceptedChannel() override {
+    GPR_CODEGEN_ASSERT(false &&
+                       "It is illegal to call GetInterceptedChannel on a "
+                       "method which has a Cancel notification");
+    return std::unique_ptr<ChannelInterface>(nullptr);
+  }
+};
+}  // namespace internal
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_INTERCEPTOR_COMMON_H
diff --git a/include/grpcpp/impl/codegen/metadata_map.h b/include/grpcpp/impl/codegen/metadata_map.h
index 5e062a50f8..9cec54d9f0 100644
--- a/include/grpcpp/impl/codegen/metadata_map.h
+++ b/include/grpcpp/impl/codegen/metadata_map.h
@@ -19,6 +19,8 @@
 #ifndef GRPCPP_IMPL_CODEGEN_METADATA_MAP_H
 #define GRPCPP_IMPL_CODEGEN_METADATA_MAP_H
 
+#include <map>
+
 #include <grpc/impl/codegen/log.h>
 #include <grpcpp/impl/codegen/slice.h>
 
@@ -30,11 +32,9 @@ const char kBinaryErrorDetailsKey[] = "grpc-status-details-bin";
 
 class MetadataMap {
  public:
-  MetadataMap() { memset(&arr_, 0, sizeof(arr_)); }
+  MetadataMap() { Setup(); }
 
-  ~MetadataMap() {
-    g_core_codegen_interface->grpc_metadata_array_destroy(&arr_);
-  }
+  ~MetadataMap() { Destroy(); }
 
   grpc::string GetBinaryErrorDetails() {
     // if filled_, extract from the multimap for O(log(n))
@@ -69,11 +69,24 @@ class MetadataMap {
   }
   grpc_metadata_array* arr() { return &arr_; }
 
+  void Reset() {
+    filled_ = false;
+    map_.clear();
+    Destroy();
+    Setup();
+  }
+
  private:
   bool filled_ = false;
   grpc_metadata_array arr_;
   std::multimap<grpc::string_ref, grpc::string_ref> map_;
 
+  void Destroy() {
+    g_core_codegen_interface->grpc_metadata_array_destroy(&arr_);
+  }
+
+  void Setup() { memset(&arr_, 0, sizeof(arr_)); }
+
   void FillMap() {
     if (filled_) return;
     filled_ = true;
diff --git a/include/grpcpp/impl/codegen/method_handler_impl.h b/include/grpcpp/impl/codegen/method_handler_impl.h
index 53117f941b..dd53f975f6 100644
--- a/include/grpcpp/impl/codegen/method_handler_impl.h
+++ b/include/grpcpp/impl/codegen/method_handler_impl.h
@@ -59,21 +59,21 @@ class RpcMethodHandler : public MethodHandler {
       : func_(func), service_(service) {}
 
   void RunHandler(const HandlerParameter& param) final {
-    RequestType req;
-    Status status = SerializationTraits<RequestType>::Deserialize(
-        param.request.bbuf_ptr(), &req);
     ResponseType rsp;
+    Status status = param.status;
     if (status.ok()) {
-      status = CatchingFunctionHandler([this, &param, &req, &rsp] {
-        return func_(service_, param.server_context, &req, &rsp);
+      status = CatchingFunctionHandler([this, &param, &rsp] {
+        return func_(service_, param.server_context,
+                     static_cast<RequestType*>(param.request), &rsp);
       });
+      static_cast<RequestType*>(param.request)->~RequestType();
     }
 
     GPR_CODEGEN_ASSERT(!param.server_context->sent_initial_metadata_);
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
               CallOpServerSendStatus>
         ops;
-    ops.SendInitialMetadata(param.server_context->initial_metadata_,
+    ops.SendInitialMetadata(&param.server_context->initial_metadata_,
                             param.server_context->initial_metadata_flags());
     if (param.server_context->compression_level_set()) {
       ops.set_compression_level(param.server_context->compression_level());
@@ -81,11 +81,26 @@ class RpcMethodHandler : public MethodHandler {
     if (status.ok()) {
       status = ops.SendMessage(rsp);
     }
-    ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
+    ops.ServerSendStatus(&param.server_context->trailing_metadata_, status);
     param.call->PerformOps(&ops);
     param.call->cq()->Pluck(&ops);
   }
 
+  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                    Status* status) final {
+    ByteBuffer buf;
+    buf.set_buffer(req);
+    auto* request = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call, sizeof(RequestType))) RequestType();
+    *status = SerializationTraits<RequestType>::Deserialize(&buf, request);
+    buf.Release();
+    if (status->ok()) {
+      return request;
+    }
+    request->~RequestType();
+    return nullptr;
+  }
+
  private:
   /// Application provided rpc handler function.
   std::function<Status(ServiceType*, ServerContext*, const RequestType*,
@@ -117,7 +132,7 @@ class ClientStreamingHandler : public MethodHandler {
               CallOpServerSendStatus>
         ops;
     if (!param.server_context->sent_initial_metadata_) {
-      ops.SendInitialMetadata(param.server_context->initial_metadata_,
+      ops.SendInitialMetadata(&param.server_context->initial_metadata_,
                               param.server_context->initial_metadata_flags());
       if (param.server_context->compression_level_set()) {
         ops.set_compression_level(param.server_context->compression_level());
@@ -126,7 +141,7 @@ class ClientStreamingHandler : public MethodHandler {
     if (status.ok()) {
       status = ops.SendMessage(rsp);
     }
-    ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
+    ops.ServerSendStatus(&param.server_context->trailing_metadata_, status);
     param.call->PerformOps(&ops);
     param.call->cq()->Pluck(&ops);
   }
@@ -150,26 +165,25 @@ class ServerStreamingHandler : public MethodHandler {
       : func_(func), service_(service) {}
 
   void RunHandler(const HandlerParameter& param) final {
-    RequestType req;
-    Status status = SerializationTraits<RequestType>::Deserialize(
-        param.request.bbuf_ptr(), &req);
-
+    Status status = param.status;
     if (status.ok()) {
       ServerWriter<ResponseType> writer(param.call, param.server_context);
-      status = CatchingFunctionHandler([this, &param, &req, &writer] {
-        return func_(service_, param.server_context, &req, &writer);
+      status = CatchingFunctionHandler([this, &param, &writer] {
+        return func_(service_, param.server_context,
+                     static_cast<RequestType*>(param.request), &writer);
       });
+      static_cast<RequestType*>(param.request)->~RequestType();
     }
 
     CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> ops;
     if (!param.server_context->sent_initial_metadata_) {
-      ops.SendInitialMetadata(param.server_context->initial_metadata_,
+      ops.SendInitialMetadata(&param.server_context->initial_metadata_,
                               param.server_context->initial_metadata_flags());
       if (param.server_context->compression_level_set()) {
         ops.set_compression_level(param.server_context->compression_level());
       }
     }
-    ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
+    ops.ServerSendStatus(&param.server_context->trailing_metadata_, status);
     param.call->PerformOps(&ops);
     if (param.server_context->has_pending_ops_) {
       param.call->cq()->Pluck(&param.server_context->pending_ops_);
@@ -177,6 +191,21 @@ class ServerStreamingHandler : public MethodHandler {
     param.call->cq()->Pluck(&ops);
   }
 
+  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                    Status* status) final {
+    ByteBuffer buf;
+    buf.set_buffer(req);
+    auto* request = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call, sizeof(RequestType))) RequestType();
+    *status = SerializationTraits<RequestType>::Deserialize(&buf, request);
+    buf.Release();
+    if (status->ok()) {
+      return request;
+    }
+    request->~RequestType();
+    return nullptr;
+  }
+
  private:
   std::function<Status(ServiceType*, ServerContext*, const RequestType*,
                        ServerWriter<ResponseType>*)>
@@ -206,7 +235,7 @@ class TemplatedBidiStreamingHandler : public MethodHandler {
 
     CallOpSet<CallOpSendInitialMetadata, CallOpServerSendStatus> ops;
     if (!param.server_context->sent_initial_metadata_) {
-      ops.SendInitialMetadata(param.server_context->initial_metadata_,
+      ops.SendInitialMetadata(&param.server_context->initial_metadata_,
                               param.server_context->initial_metadata_flags());
       if (param.server_context->compression_level_set()) {
         ops.set_compression_level(param.server_context->compression_level());
@@ -218,7 +247,7 @@ class TemplatedBidiStreamingHandler : public MethodHandler {
                         "Service did not provide response message");
       }
     }
-    ops.ServerSendStatus(param.server_context->trailing_metadata_, status);
+    ops.ServerSendStatus(&param.server_context->trailing_metadata_, status);
     param.call->PerformOps(&ops);
     if (param.server_context->has_pending_ops_) {
       param.call->cq()->Pluck(&param.server_context->pending_ops_);
@@ -281,14 +310,14 @@ class ErrorMethodHandler : public MethodHandler {
   static void FillOps(ServerContext* context, T* ops) {
     Status status(code, "");
     if (!context->sent_initial_metadata_) {
-      ops->SendInitialMetadata(context->initial_metadata_,
+      ops->SendInitialMetadata(&context->initial_metadata_,
                                context->initial_metadata_flags());
       if (context->compression_level_set()) {
         ops->set_compression_level(context->compression_level());
       }
       context->sent_initial_metadata_ = true;
     }
-    ops->ServerSendStatus(context->trailing_metadata_, status);
+    ops->ServerSendStatus(&context->trailing_metadata_, status);
   }
 
   void RunHandler(const HandlerParameter& param) final {
@@ -296,11 +325,15 @@ class ErrorMethodHandler : public MethodHandler {
     FillOps(param.server_context, &ops);
     param.call->PerformOps(&ops);
     param.call->cq()->Pluck(&ops);
-    // We also have to destroy any request payload in the handler parameter
-    ByteBuffer* payload = param.request.bbuf_ptr();
-    if (payload != nullptr) {
-      payload->Clear();
+  }
+
+  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                    Status* status) final {
+    // We have to destroy any request payload
+    if (req != nullptr) {
+      g_core_codegen_interface->grpc_byte_buffer_destroy(req);
     }
+    return nullptr;
   }
 };
 
diff --git a/include/grpcpp/impl/codegen/rpc_service_method.h b/include/grpcpp/impl/codegen/rpc_service_method.h
index 5cf88e216f..f465c5fc2f 100644
--- a/include/grpcpp/impl/codegen/rpc_service_method.h
+++ b/include/grpcpp/impl/codegen/rpc_service_method.h
@@ -40,17 +40,41 @@ class MethodHandler {
  public:
   virtual ~MethodHandler() {}
   struct HandlerParameter {
-    HandlerParameter(Call* c, ServerContext* context, grpc_byte_buffer* req)
-        : call(c), server_context(context) {
-      request.set_buffer(req);
-    }
-    ~HandlerParameter() { request.Release(); }
+    /// Constructor for HandlerParameter
+    ///
+    /// \param c : the gRPC Call structure for this server call
+    /// \param context : the ServerContext structure for this server call
+    /// \param req : the request payload, if appropriate for this RPC
+    /// \param req_status : the request status after any interceptors have run
+    /// \param rpc_requester : used only by the callback API. It is a function
+    ///        called by the RPC Controller to request another RPC (and also
+    ///        to set up the state required to make that request possible)
+    HandlerParameter(Call* c, ServerContext* context, void* req,
+                     Status req_status, std::function<void()> requester)
+        : call(c),
+          server_context(context),
+          request(req),
+          status(req_status),
+          call_requester(std::move(requester)) {}
+    ~HandlerParameter() {}
     Call* call;
     ServerContext* server_context;
-    // Handler required to destroy these contents
-    ByteBuffer request;
+    void* request;
+    Status status;
+    std::function<void()> call_requester;
   };
   virtual void RunHandler(const HandlerParameter& param) = 0;
+
+  /* Returns a pointer to the deserialized request. \a status reflects the
+     result of deserialization. This pointer and the status should be filled in
+     a HandlerParameter and passed to RunHandler. It is illegal to access the
+     pointer after calling RunHandler. Ownership of the deserialized request is
+     retained by the handler. Returns nullptr if deserialization failed. */
+  virtual void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                            Status* status) {
+    GPR_CODEGEN_ASSERT(req == nullptr);
+    return nullptr;
+  }
 };
 
 /// Server side rpc method class
@@ -61,25 +85,29 @@ class RpcServiceMethod : public RpcMethod {
                    MethodHandler* handler)
       : RpcMethod(name, type),
         server_tag_(nullptr),
-        async_type_(AsyncType::UNSET),
+        api_type_(ApiType::SYNC),
         handler_(handler) {}
 
-  enum class AsyncType {
-    UNSET,
+  enum class ApiType {
+    SYNC,
     ASYNC,
     RAW,
+    CALL_BACK,  // not CALLBACK because that is reserved in Windows
+    RAW_CALL_BACK,
   };
 
   void set_server_tag(void* tag) { server_tag_ = tag; }
   void* server_tag() const { return server_tag_; }
   /// if MethodHandler is nullptr, then this is an async method
   MethodHandler* handler() const { return handler_.get(); }
+  ApiType api_type() const { return api_type_; }
   void SetHandler(MethodHandler* handler) { handler_.reset(handler); }
-  void SetServerAsyncType(RpcServiceMethod::AsyncType type) {
-    if (async_type_ == AsyncType::UNSET) {
+  void SetServerApiType(RpcServiceMethod::ApiType type) {
+    if ((api_type_ == ApiType::SYNC) &&
+        (type == ApiType::ASYNC || type == ApiType::RAW)) {
       // this marks this method as async
       handler_.reset();
-    } else {
+    } else if (api_type_ != ApiType::SYNC) {
       // this is not an error condition, as it allows users to declare a server
       // like WithRawMethod_foo<AsyncService>. However since it
       // overwrites behavior, it should be logged.
@@ -88,24 +116,28 @@ class RpcServiceMethod : public RpcMethod {
           "You are marking method %s as '%s', even though it was "
           "previously marked '%s'. This behavior will overwrite the original "
           "behavior. If you expected this then ignore this message.",
-          name(), TypeToString(async_type_), TypeToString(type));
+          name(), TypeToString(api_type_), TypeToString(type));
     }
-    async_type_ = type;
+    api_type_ = type;
   }
 
  private:
   void* server_tag_;
-  AsyncType async_type_;
+  ApiType api_type_;
   std::unique_ptr<MethodHandler> handler_;
 
-  const char* TypeToString(RpcServiceMethod::AsyncType type) {
+  const char* TypeToString(RpcServiceMethod::ApiType type) {
     switch (type) {
-      case AsyncType::UNSET:
-        return "unset";
-      case AsyncType::ASYNC:
+      case ApiType::SYNC:
+        return "sync";
+      case ApiType::ASYNC:
         return "async";
-      case AsyncType::RAW:
+      case ApiType::RAW:
         return "raw";
+      case ApiType::CALL_BACK:
+        return "callback";
+      case ApiType::RAW_CALL_BACK:
+        return "raw_callback";
       default:
         GPR_UNREACHABLE_CODE(return "unknown");
     }
diff --git a/include/grpcpp/impl/codegen/server_callback.h b/include/grpcpp/impl/codegen/server_callback.h
new file mode 100644
index 0000000000..1854f6ef2f
--- /dev/null
+++ b/include/grpcpp/impl/codegen/server_callback.h
@@ -0,0 +1,890 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_H
+#define GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_H
+
+#include <atomic>
+#include <functional>
+#include <type_traits>
+
+#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/config.h>
+#include <grpcpp/impl/codegen/core_codegen_interface.h>
+#include <grpcpp/impl/codegen/server_context.h>
+#include <grpcpp/impl/codegen/server_interface.h>
+#include <grpcpp/impl/codegen/status.h>
+
+namespace grpc {
+
+// Declare base class of all reactors as internal
+namespace internal {
+
+class ServerReactor {
+ public:
+  virtual ~ServerReactor() = default;
+  virtual void OnDone() {}
+  virtual void OnCancel() {}
+};
+
+}  // namespace internal
+
+namespace experimental {
+
+// Forward declarations
+template <class Request, class Response>
+class ServerReadReactor;
+template <class Request, class Response>
+class ServerWriteReactor;
+template <class Request, class Response>
+class ServerBidiReactor;
+
+// For unary RPCs, the exposed controller class is only an interface
+// and the actual implementation is an internal class.
+class ServerCallbackRpcController {
+ public:
+  virtual ~ServerCallbackRpcController() = default;
+
+  // The method handler must call this function when it is done so that
+  // the library knows to free its resources
+  virtual void Finish(Status s) = 0;
+
+  // Allow the method handler to push out the initial metadata before
+  // the response and status are ready
+  virtual void SendInitialMetadata(std::function<void(bool)>) = 0;
+};
+
+// NOTE: The actual streaming object classes are provided
+// as API only to support mocking. There are no implementations of
+// these class interfaces in the API.
+template <class Request>
+class ServerCallbackReader {
+ public:
+  virtual ~ServerCallbackReader() {}
+  virtual void Finish(Status s) = 0;
+  virtual void SendInitialMetadata() = 0;
+  virtual void Read(Request* msg) = 0;
+
+ protected:
+  template <class Response>
+  void BindReactor(ServerReadReactor<Request, Response>* reactor) {
+    reactor->BindReader(this);
+  }
+};
+
+template <class Response>
+class ServerCallbackWriter {
+ public:
+  virtual ~ServerCallbackWriter() {}
+
+  virtual void Finish(Status s) = 0;
+  virtual void SendInitialMetadata() = 0;
+  virtual void Write(const Response* msg, WriteOptions options) = 0;
+  virtual void WriteAndFinish(const Response* msg, WriteOptions options,
+                              Status s) {
+    // Default implementation that can/should be overridden
+    Write(msg, std::move(options));
+    Finish(std::move(s));
+  };
+
+ protected:
+  template <class Request>
+  void BindReactor(ServerWriteReactor<Request, Response>* reactor) {
+    reactor->BindWriter(this);
+  }
+};
+
+template <class Request, class Response>
+class ServerCallbackReaderWriter {
+ public:
+  virtual ~ServerCallbackReaderWriter() {}
+
+  virtual void Finish(Status s) = 0;
+  virtual void SendInitialMetadata() = 0;
+  virtual void Read(Request* msg) = 0;
+  virtual void Write(const Response* msg, WriteOptions options) = 0;
+  virtual void WriteAndFinish(const Response* msg, WriteOptions options,
+                              Status s) {
+    // Default implementation that can/should be overridden
+    Write(msg, std::move(options));
+    Finish(std::move(s));
+  };
+
+ protected:
+  void BindReactor(ServerBidiReactor<Request, Response>* reactor) {
+    reactor->BindStream(this);
+  }
+};
+
+// The following classes are reactors that are to be implemented
+// by the user, returned as the result of the method handler for
+// a callback method, and activated by the call to OnStarted
+template <class Request, class Response>
+class ServerBidiReactor : public internal::ServerReactor {
+ public:
+  ~ServerBidiReactor() = default;
+  virtual void OnStarted(ServerContext*) {}
+  virtual void OnSendInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+
+  void StartSendInitialMetadata() { stream_->SendInitialMetadata(); }
+  void StartRead(Request* msg) { stream_->Read(msg); }
+  void StartWrite(const Response* msg) { StartWrite(msg, WriteOptions()); }
+  void StartWrite(const Response* msg, WriteOptions options) {
+    stream_->Write(msg, std::move(options));
+  }
+  void StartWriteAndFinish(const Response* msg, WriteOptions options,
+                           Status s) {
+    stream_->WriteAndFinish(msg, std::move(options), std::move(s));
+  }
+  void StartWriteLast(const Response* msg, WriteOptions options) {
+    StartWrite(msg, std::move(options.set_last_message()));
+  }
+  void Finish(Status s) { stream_->Finish(std::move(s)); }
+
+ private:
+  friend class ServerCallbackReaderWriter<Request, Response>;
+  void BindStream(ServerCallbackReaderWriter<Request, Response>* stream) {
+    stream_ = stream;
+  }
+
+  ServerCallbackReaderWriter<Request, Response>* stream_;
+};
+
+template <class Request, class Response>
+class ServerReadReactor : public internal::ServerReactor {
+ public:
+  ~ServerReadReactor() = default;
+  virtual void OnStarted(ServerContext*, Response* resp) {}
+  virtual void OnSendInitialMetadataDone(bool ok) {}
+  virtual void OnReadDone(bool ok) {}
+
+  void StartSendInitialMetadata() { reader_->SendInitialMetadata(); }
+  void StartRead(Request* msg) { reader_->Read(msg); }
+  void Finish(Status s) { reader_->Finish(std::move(s)); }
+
+ private:
+  friend class ServerCallbackReader<Request>;
+  void BindReader(ServerCallbackReader<Request>* reader) { reader_ = reader; }
+
+  ServerCallbackReader<Request>* reader_;
+};
+
+template <class Request, class Response>
+class ServerWriteReactor : public internal::ServerReactor {
+ public:
+  ~ServerWriteReactor() = default;
+  virtual void OnStarted(ServerContext*, const Request* req) {}
+  virtual void OnSendInitialMetadataDone(bool ok) {}
+  virtual void OnWriteDone(bool ok) {}
+
+  void StartSendInitialMetadata() { writer_->SendInitialMetadata(); }
+  void StartWrite(const Response* msg) { StartWrite(msg, WriteOptions()); }
+  void StartWrite(const Response* msg, WriteOptions options) {
+    writer_->Write(msg, std::move(options));
+  }
+  void StartWriteAndFinish(const Response* msg, WriteOptions options,
+                           Status s) {
+    writer_->WriteAndFinish(msg, std::move(options), std::move(s));
+  }
+  void StartWriteLast(const Response* msg, WriteOptions options) {
+    StartWrite(msg, std::move(options.set_last_message()));
+  }
+  void Finish(Status s) { writer_->Finish(std::move(s)); }
+
+ private:
+  friend class ServerCallbackWriter<Response>;
+  void BindWriter(ServerCallbackWriter<Response>* writer) { writer_ = writer; }
+
+  ServerCallbackWriter<Response>* writer_;
+};
+
+}  // namespace experimental
+
+namespace internal {
+
+template <class Request, class Response>
+class UnimplementedReadReactor
+    : public experimental::ServerReadReactor<Request, Response> {
+ public:
+  void OnDone() override { delete this; }
+  void OnStarted(ServerContext*, Response*) override {
+    this->Finish(Status(StatusCode::UNIMPLEMENTED, ""));
+  }
+};
+
+template <class Request, class Response>
+class UnimplementedWriteReactor
+    : public experimental::ServerWriteReactor<Request, Response> {
+ public:
+  void OnDone() override { delete this; }
+  void OnStarted(ServerContext*, const Request*) override {
+    this->Finish(Status(StatusCode::UNIMPLEMENTED, ""));
+  }
+};
+
+template <class Request, class Response>
+class UnimplementedBidiReactor
+    : public experimental::ServerBidiReactor<Request, Response> {
+ public:
+  void OnDone() override { delete this; }
+  void OnStarted(ServerContext*) override {
+    this->Finish(Status(StatusCode::UNIMPLEMENTED, ""));
+  }
+};
+
+template <class RequestType, class ResponseType>
+class CallbackUnaryHandler : public MethodHandler {
+ public:
+  CallbackUnaryHandler(
+      std::function<void(ServerContext*, const RequestType*, ResponseType*,
+                         experimental::ServerCallbackRpcController*)>
+          func)
+      : func_(func) {}
+  void RunHandler(const HandlerParameter& param) final {
+    // Arena allocate a controller structure (that includes request/response)
+    g_core_codegen_interface->grpc_call_ref(param.call->call());
+    auto* controller = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        param.call->call(), sizeof(ServerCallbackRpcControllerImpl)))
+        ServerCallbackRpcControllerImpl(
+            param.server_context, param.call,
+            static_cast<RequestType*>(param.request),
+            std::move(param.call_requester));
+    Status status = param.status;
+
+    if (status.ok()) {
+      // Call the actual function handler and expect the user to call finish
+      CatchingCallback(func_, param.server_context, controller->request(),
+                       controller->response(), controller);
+    } else {
+      // if deserialization failed, we need to fail the call
+      controller->Finish(status);
+    }
+  }
+
+  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                    Status* status) final {
+    ByteBuffer buf;
+    buf.set_buffer(req);
+    auto* request = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call, sizeof(RequestType))) RequestType();
+    *status = SerializationTraits<RequestType>::Deserialize(&buf, request);
+    buf.Release();
+    if (status->ok()) {
+      return request;
+    }
+    request->~RequestType();
+    return nullptr;
+  }
+
+ private:
+  std::function<void(ServerContext*, const RequestType*, ResponseType*,
+                     experimental::ServerCallbackRpcController*)>
+      func_;
+
+  // The implementation class of ServerCallbackRpcController is a private member
+  // of CallbackUnaryHandler since it is never exposed anywhere, and this allows
+  // it to take advantage of CallbackUnaryHandler's friendships.
+  class ServerCallbackRpcControllerImpl
+      : public experimental::ServerCallbackRpcController {
+   public:
+    void Finish(Status s) override {
+      finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); },
+                      &finish_ops_);
+      if (!ctx_->sent_initial_metadata_) {
+        finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                        ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          finish_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      // The response is dropped if the status is not OK.
+      if (s.ok()) {
+        finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_,
+                                     finish_ops_.SendMessage(resp_));
+      } else {
+        finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s);
+      }
+      finish_ops_.set_core_cq_tag(&finish_tag_);
+      call_.PerformOps(&finish_ops_);
+    }
+
+    void SendInitialMetadata(std::function<void(bool)> f) override {
+      GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
+      callbacks_outstanding_++;
+      // TODO(vjpai): Consider taking f as a move-capture if we adopt C++14
+      //              and if performance of this operation matters
+      meta_tag_.Set(call_.call(),
+                    [this, f](bool ok) {
+                      f(ok);
+                      MaybeDone();
+                    },
+                    &meta_ops_);
+      meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                    ctx_->initial_metadata_flags());
+      if (ctx_->compression_level_set()) {
+        meta_ops_.set_compression_level(ctx_->compression_level());
+      }
+      ctx_->sent_initial_metadata_ = true;
+      meta_ops_.set_core_cq_tag(&meta_tag_);
+      call_.PerformOps(&meta_ops_);
+    }
+
+   private:
+    friend class CallbackUnaryHandler<RequestType, ResponseType>;
+
+    ServerCallbackRpcControllerImpl(ServerContext* ctx, Call* call,
+                                    const RequestType* req,
+                                    std::function<void()> call_requester)
+        : ctx_(ctx),
+          call_(*call),
+          req_(req),
+          call_requester_(std::move(call_requester)) {
+      ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, nullptr);
+    }
+
+    ~ServerCallbackRpcControllerImpl() { req_->~RequestType(); }
+
+    const RequestType* request() { return req_; }
+    ResponseType* response() { return &resp_; }
+
+    void MaybeDone() {
+      if (--callbacks_outstanding_ == 0) {
+        grpc_call* call = call_.call();
+        auto call_requester = std::move(call_requester_);
+        this->~ServerCallbackRpcControllerImpl();  // explicitly call destructor
+        g_core_codegen_interface->grpc_call_unref(call);
+        call_requester();
+      }
+    }
+
+    CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+    CallbackWithSuccessTag meta_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+              CallOpServerSendStatus>
+        finish_ops_;
+    CallbackWithSuccessTag finish_tag_;
+
+    ServerContext* ctx_;
+    Call call_;
+    const RequestType* req_;
+    ResponseType resp_;
+    std::function<void()> call_requester_;
+    std::atomic_int callbacks_outstanding_{
+        2};  // reserve for Finish and CompletionOp
+  };
+};
+
+template <class RequestType, class ResponseType>
+class CallbackClientStreamingHandler : public MethodHandler {
+ public:
+  CallbackClientStreamingHandler(
+      std::function<
+          experimental::ServerReadReactor<RequestType, ResponseType>*()>
+          func)
+      : func_(std::move(func)) {}
+  void RunHandler(const HandlerParameter& param) final {
+    // Arena allocate a reader structure (that includes response)
+    g_core_codegen_interface->grpc_call_ref(param.call->call());
+
+    experimental::ServerReadReactor<RequestType, ResponseType>* reactor =
+        param.status.ok()
+            ? CatchingReactorCreator<
+                  experimental::ServerReadReactor<RequestType, ResponseType>>(
+                  func_)
+            : nullptr;
+
+    if (reactor == nullptr) {
+      // if deserialization or reactor creator failed, we need to fail the call
+      reactor = new UnimplementedReadReactor<RequestType, ResponseType>;
+    }
+
+    auto* reader = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        param.call->call(), sizeof(ServerCallbackReaderImpl)))
+        ServerCallbackReaderImpl(param.server_context, param.call,
+                                 std::move(param.call_requester), reactor);
+
+    reader->BindReactor(reactor);
+    reactor->OnStarted(param.server_context, reader->response());
+    reader->MaybeDone();
+  }
+
+ private:
+  std::function<experimental::ServerReadReactor<RequestType, ResponseType>*()>
+      func_;
+
+  class ServerCallbackReaderImpl
+      : public experimental::ServerCallbackReader<RequestType> {
+   public:
+    void Finish(Status s) override {
+      finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); },
+                      &finish_ops_);
+      if (!ctx_->sent_initial_metadata_) {
+        finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                        ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          finish_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      // The response is dropped if the status is not OK.
+      if (s.ok()) {
+        finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_,
+                                     finish_ops_.SendMessage(resp_));
+      } else {
+        finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s);
+      }
+      finish_ops_.set_core_cq_tag(&finish_tag_);
+      call_.PerformOps(&finish_ops_);
+    }
+
+    void SendInitialMetadata() override {
+      GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
+      callbacks_outstanding_++;
+      meta_tag_.Set(call_.call(),
+                    [this](bool ok) {
+                      reactor_->OnSendInitialMetadataDone(ok);
+                      MaybeDone();
+                    },
+                    &meta_ops_);
+      meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                    ctx_->initial_metadata_flags());
+      if (ctx_->compression_level_set()) {
+        meta_ops_.set_compression_level(ctx_->compression_level());
+      }
+      ctx_->sent_initial_metadata_ = true;
+      meta_ops_.set_core_cq_tag(&meta_tag_);
+      call_.PerformOps(&meta_ops_);
+    }
+
+    void Read(RequestType* req) override {
+      callbacks_outstanding_++;
+      read_ops_.RecvMessage(req);
+      call_.PerformOps(&read_ops_);
+    }
+
+   private:
+    friend class CallbackClientStreamingHandler<RequestType, ResponseType>;
+
+    ServerCallbackReaderImpl(
+        ServerContext* ctx, Call* call, std::function<void()> call_requester,
+        experimental::ServerReadReactor<RequestType, ResponseType>* reactor)
+        : ctx_(ctx),
+          call_(*call),
+          call_requester_(std::move(call_requester)),
+          reactor_(reactor) {
+      ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor);
+      read_tag_.Set(call_.call(),
+                    [this](bool ok) {
+                      reactor_->OnReadDone(ok);
+                      MaybeDone();
+                    },
+                    &read_ops_);
+      read_ops_.set_core_cq_tag(&read_tag_);
+    }
+
+    ~ServerCallbackReaderImpl() {}
+
+    ResponseType* response() { return &resp_; }
+
+    void MaybeDone() {
+      if (--callbacks_outstanding_ == 0) {
+        reactor_->OnDone();
+        grpc_call* call = call_.call();
+        auto call_requester = std::move(call_requester_);
+        this->~ServerCallbackReaderImpl();  // explicitly call destructor
+        g_core_codegen_interface->grpc_call_unref(call);
+        call_requester();
+      }
+    }
+
+    CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+    CallbackWithSuccessTag meta_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+              CallOpServerSendStatus>
+        finish_ops_;
+    CallbackWithSuccessTag finish_tag_;
+    CallOpSet<CallOpRecvMessage<RequestType>> read_ops_;
+    CallbackWithSuccessTag read_tag_;
+
+    ServerContext* ctx_;
+    Call call_;
+    ResponseType resp_;
+    std::function<void()> call_requester_;
+    experimental::ServerReadReactor<RequestType, ResponseType>* reactor_;
+    std::atomic_int callbacks_outstanding_{
+        3};  // reserve for OnStarted, Finish, and CompletionOp
+  };
+};
+
+template <class RequestType, class ResponseType>
+class CallbackServerStreamingHandler : public MethodHandler {
+ public:
+  CallbackServerStreamingHandler(
+      std::function<
+          experimental::ServerWriteReactor<RequestType, ResponseType>*()>
+          func)
+      : func_(std::move(func)) {}
+  void RunHandler(const HandlerParameter& param) final {
+    // Arena allocate a writer structure
+    g_core_codegen_interface->grpc_call_ref(param.call->call());
+
+    experimental::ServerWriteReactor<RequestType, ResponseType>* reactor =
+        param.status.ok()
+            ? CatchingReactorCreator<
+                  experimental::ServerWriteReactor<RequestType, ResponseType>>(
+                  func_)
+            : nullptr;
+
+    if (reactor == nullptr) {
+      // if deserialization or reactor creator failed, we need to fail the call
+      reactor = new UnimplementedWriteReactor<RequestType, ResponseType>;
+    }
+
+    auto* writer = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        param.call->call(), sizeof(ServerCallbackWriterImpl)))
+        ServerCallbackWriterImpl(param.server_context, param.call,
+                                 static_cast<RequestType*>(param.request),
+                                 std::move(param.call_requester), reactor);
+    writer->BindReactor(reactor);
+    reactor->OnStarted(param.server_context, writer->request());
+    writer->MaybeDone();
+  }
+
+  void* Deserialize(grpc_call* call, grpc_byte_buffer* req,
+                    Status* status) final {
+    ByteBuffer buf;
+    buf.set_buffer(req);
+    auto* request = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        call, sizeof(RequestType))) RequestType();
+    *status = SerializationTraits<RequestType>::Deserialize(&buf, request);
+    buf.Release();
+    if (status->ok()) {
+      return request;
+    }
+    request->~RequestType();
+    return nullptr;
+  }
+
+ private:
+  std::function<experimental::ServerWriteReactor<RequestType, ResponseType>*()>
+      func_;
+
+  class ServerCallbackWriterImpl
+      : public experimental::ServerCallbackWriter<ResponseType> {
+   public:
+    void Finish(Status s) override {
+      finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); },
+                      &finish_ops_);
+      finish_ops_.set_core_cq_tag(&finish_tag_);
+
+      if (!ctx_->sent_initial_metadata_) {
+        finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                        ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          finish_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s);
+      call_.PerformOps(&finish_ops_);
+    }
+
+    void SendInitialMetadata() override {
+      GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
+      callbacks_outstanding_++;
+      meta_tag_.Set(call_.call(),
+                    [this](bool ok) {
+                      reactor_->OnSendInitialMetadataDone(ok);
+                      MaybeDone();
+                    },
+                    &meta_ops_);
+      meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                    ctx_->initial_metadata_flags());
+      if (ctx_->compression_level_set()) {
+        meta_ops_.set_compression_level(ctx_->compression_level());
+      }
+      ctx_->sent_initial_metadata_ = true;
+      meta_ops_.set_core_cq_tag(&meta_tag_);
+      call_.PerformOps(&meta_ops_);
+    }
+
+    void Write(const ResponseType* resp, WriteOptions options) override {
+      callbacks_outstanding_++;
+      if (options.is_last_message()) {
+        options.set_buffer_hint();
+      }
+      if (!ctx_->sent_initial_metadata_) {
+        write_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                       ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          write_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      // TODO(vjpai): don't assert
+      GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*resp, options).ok());
+      call_.PerformOps(&write_ops_);
+    }
+
+    void WriteAndFinish(const ResponseType* resp, WriteOptions options,
+                        Status s) override {
+      // This combines the write into the finish callback
+      // Don't send any message if the status is bad
+      if (s.ok()) {
+        // TODO(vjpai): don't assert
+        GPR_CODEGEN_ASSERT(finish_ops_.SendMessage(*resp, options).ok());
+      }
+      Finish(std::move(s));
+    }
+
+   private:
+    friend class CallbackServerStreamingHandler<RequestType, ResponseType>;
+
+    ServerCallbackWriterImpl(
+        ServerContext* ctx, Call* call, const RequestType* req,
+        std::function<void()> call_requester,
+        experimental::ServerWriteReactor<RequestType, ResponseType>* reactor)
+        : ctx_(ctx),
+          call_(*call),
+          req_(req),
+          call_requester_(std::move(call_requester)),
+          reactor_(reactor) {
+      ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor);
+      write_tag_.Set(call_.call(),
+                     [this](bool ok) {
+                       reactor_->OnWriteDone(ok);
+                       MaybeDone();
+                     },
+                     &write_ops_);
+      write_ops_.set_core_cq_tag(&write_tag_);
+    }
+    ~ServerCallbackWriterImpl() { req_->~RequestType(); }
+
+    const RequestType* request() { return req_; }
+
+    void MaybeDone() {
+      if (--callbacks_outstanding_ == 0) {
+        reactor_->OnDone();
+        grpc_call* call = call_.call();
+        auto call_requester = std::move(call_requester_);
+        this->~ServerCallbackWriterImpl();  // explicitly call destructor
+        g_core_codegen_interface->grpc_call_unref(call);
+        call_requester();
+      }
+    }
+
+    CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+    CallbackWithSuccessTag meta_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+              CallOpServerSendStatus>
+        finish_ops_;
+    CallbackWithSuccessTag finish_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
+    CallbackWithSuccessTag write_tag_;
+
+    ServerContext* ctx_;
+    Call call_;
+    const RequestType* req_;
+    std::function<void()> call_requester_;
+    experimental::ServerWriteReactor<RequestType, ResponseType>* reactor_;
+    std::atomic_int callbacks_outstanding_{
+        3};  // reserve for OnStarted, Finish, and CompletionOp
+  };
+};
+
+template <class RequestType, class ResponseType>
+class CallbackBidiHandler : public MethodHandler {
+ public:
+  CallbackBidiHandler(
+      std::function<
+          experimental::ServerBidiReactor<RequestType, ResponseType>*()>
+          func)
+      : func_(std::move(func)) {}
+  void RunHandler(const HandlerParameter& param) final {
+    g_core_codegen_interface->grpc_call_ref(param.call->call());
+
+    experimental::ServerBidiReactor<RequestType, ResponseType>* reactor =
+        param.status.ok()
+            ? CatchingReactorCreator<
+                  experimental::ServerBidiReactor<RequestType, ResponseType>>(
+                  func_)
+            : nullptr;
+
+    if (reactor == nullptr) {
+      // if deserialization or reactor creator failed, we need to fail the call
+      reactor = new UnimplementedBidiReactor<RequestType, ResponseType>;
+    }
+
+    auto* stream = new (g_core_codegen_interface->grpc_call_arena_alloc(
+        param.call->call(), sizeof(ServerCallbackReaderWriterImpl)))
+        ServerCallbackReaderWriterImpl(param.server_context, param.call,
+                                       std::move(param.call_requester),
+                                       reactor);
+
+    stream->BindReactor(reactor);
+    reactor->OnStarted(param.server_context);
+    stream->MaybeDone();
+  }
+
+ private:
+  std::function<experimental::ServerBidiReactor<RequestType, ResponseType>*()>
+      func_;
+
+  class ServerCallbackReaderWriterImpl
+      : public experimental::ServerCallbackReaderWriter<RequestType,
+                                                        ResponseType> {
+   public:
+    void Finish(Status s) override {
+      finish_tag_.Set(call_.call(), [this](bool) { MaybeDone(); },
+                      &finish_ops_);
+      finish_ops_.set_core_cq_tag(&finish_tag_);
+
+      if (!ctx_->sent_initial_metadata_) {
+        finish_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                        ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          finish_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      finish_ops_.ServerSendStatus(&ctx_->trailing_metadata_, s);
+      call_.PerformOps(&finish_ops_);
+    }
+
+    void SendInitialMetadata() override {
+      GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
+      callbacks_outstanding_++;
+      meta_tag_.Set(call_.call(),
+                    [this](bool ok) {
+                      reactor_->OnSendInitialMetadataDone(ok);
+                      MaybeDone();
+                    },
+                    &meta_ops_);
+      meta_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                    ctx_->initial_metadata_flags());
+      if (ctx_->compression_level_set()) {
+        meta_ops_.set_compression_level(ctx_->compression_level());
+      }
+      ctx_->sent_initial_metadata_ = true;
+      meta_ops_.set_core_cq_tag(&meta_tag_);
+      call_.PerformOps(&meta_ops_);
+    }
+
+    void Write(const ResponseType* resp, WriteOptions options) override {
+      callbacks_outstanding_++;
+      if (options.is_last_message()) {
+        options.set_buffer_hint();
+      }
+      if (!ctx_->sent_initial_metadata_) {
+        write_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
+                                       ctx_->initial_metadata_flags());
+        if (ctx_->compression_level_set()) {
+          write_ops_.set_compression_level(ctx_->compression_level());
+        }
+        ctx_->sent_initial_metadata_ = true;
+      }
+      // TODO(vjpai): don't assert
+      GPR_CODEGEN_ASSERT(write_ops_.SendMessage(*resp, options).ok());
+      call_.PerformOps(&write_ops_);
+    }
+
+    void WriteAndFinish(const ResponseType* resp, WriteOptions options,
+                        Status s) override {
+      // Don't send any message if the status is bad
+      if (s.ok()) {
+        // TODO(vjpai): don't assert
+        GPR_CODEGEN_ASSERT(finish_ops_.SendMessage(*resp, options).ok());
+      }
+      Finish(std::move(s));
+    }
+
+    void Read(RequestType* req) override {
+      callbacks_outstanding_++;
+      read_ops_.RecvMessage(req);
+      call_.PerformOps(&read_ops_);
+    }
+
+   private:
+    friend class CallbackBidiHandler<RequestType, ResponseType>;
+
+    ServerCallbackReaderWriterImpl(
+        ServerContext* ctx, Call* call, std::function<void()> call_requester,
+        experimental::ServerBidiReactor<RequestType, ResponseType>* reactor)
+        : ctx_(ctx),
+          call_(*call),
+          call_requester_(std::move(call_requester)),
+          reactor_(reactor) {
+      ctx_->BeginCompletionOp(call, [this](bool) { MaybeDone(); }, reactor);
+      write_tag_.Set(call_.call(),
+                     [this](bool ok) {
+                       reactor_->OnWriteDone(ok);
+                       MaybeDone();
+                     },
+                     &write_ops_);
+      write_ops_.set_core_cq_tag(&write_tag_);
+      read_tag_.Set(call_.call(),
+                    [this](bool ok) {
+                      reactor_->OnReadDone(ok);
+                      MaybeDone();
+                    },
+                    &read_ops_);
+      read_ops_.set_core_cq_tag(&read_tag_);
+    }
+    ~ServerCallbackReaderWriterImpl() {}
+
+    void MaybeDone() {
+      if (--callbacks_outstanding_ == 0) {
+        reactor_->OnDone();
+        grpc_call* call = call_.call();
+        auto call_requester = std::move(call_requester_);
+        this->~ServerCallbackReaderWriterImpl();  // explicitly call destructor
+        g_core_codegen_interface->grpc_call_unref(call);
+        call_requester();
+      }
+    }
+
+    CallOpSet<CallOpSendInitialMetadata> meta_ops_;
+    CallbackWithSuccessTag meta_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
+              CallOpServerSendStatus>
+        finish_ops_;
+    CallbackWithSuccessTag finish_tag_;
+    CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage> write_ops_;
+    CallbackWithSuccessTag write_tag_;
+    CallOpSet<CallOpRecvMessage<RequestType>> read_ops_;
+    CallbackWithSuccessTag read_tag_;
+
+    ServerContext* ctx_;
+    Call call_;
+    std::function<void()> call_requester_;
+    experimental::ServerBidiReactor<RequestType, ResponseType>* reactor_;
+    std::atomic_int callbacks_outstanding_{
+        3};  // reserve for OnStarted, Finish, and CompletionOp
+  };
+};
+
+}  // namespace internal
+
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_SERVER_CALLBACK_H
diff --git a/include/grpcpp/impl/codegen/server_context.h b/include/grpcpp/impl/codegen/server_context.h
index b58f029de9..ccb5925e7d 100644
--- a/include/grpcpp/impl/codegen/server_context.h
+++ b/include/grpcpp/impl/codegen/server_context.h
@@ -26,11 +26,14 @@
 #include <grpc/impl/codegen/compression_types.h>
 
 #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/completion_queue_tag.h>
 #include <grpcpp/impl/codegen/config.h>
 #include <grpcpp/impl/codegen/create_auth_context.h>
 #include <grpcpp/impl/codegen/metadata_map.h>
 #include <grpcpp/impl/codegen/security/auth_context.h>
+#include <grpcpp/impl/codegen/server_interceptor.h>
 #include <grpcpp/impl/codegen/string_ref.h>
 #include <grpcpp/impl/codegen/time.h>
 
@@ -63,11 +66,20 @@ template <class ServiceType, class RequestType, class ResponseType>
 class ServerStreamingHandler;
 template <class ServiceType, class RequestType, class ResponseType>
 class BidiStreamingHandler;
+template <class RequestType, class ResponseType>
+class CallbackUnaryHandler;
+template <class RequestType, class ResponseType>
+class CallbackClientStreamingHandler;
+template <class RequestType, class ResponseType>
+class CallbackServerStreamingHandler;
+template <class RequestType, class ResponseType>
+class CallbackBidiHandler;
 template <class Streamer, bool WriteNeeded>
 class TemplatedBidiStreamingHandler;
 template <StatusCode code>
 class ErrorMethodHandler;
 class Call;
+class ServerReactor;
 }  // namespace internal
 
 class CompletionQueue;
@@ -107,7 +119,7 @@ class ServerContext {
   /// Return a \a gpr_timespec representation of the server call's deadline.
   gpr_timespec raw_deadline() const { return deadline_; }
 
-  /// Add the (\a meta_key, \a meta_value) pair to the initial metadata
+  /// Add the (\a key, \a value) pair to the initial metadata
   /// associated with a server call. These are made available at the client side
   /// by the \a grpc::ClientContext::GetServerInitialMetadata() method.
   ///
@@ -115,13 +127,13 @@ class ServerContext {
   /// to the client (which can happen explicitly, or implicitly when sending a
   /// a response message or status to the client).
   ///
-  /// \param meta_key The metadata key. If \a meta_value is binary data, it must
+  /// \param key The metadata key. If \a value is binary data, it must
   /// end in "-bin".
-  /// \param meta_value The metadata value. If its value is binary, the key name
+  /// \param value The metadata value. If its value is binary, the key name
   /// must end in "-bin".
   void AddInitialMetadata(const grpc::string& key, const grpc::string& value);
 
-  /// Add the (\a meta_key, \a meta_value) pair to the initial metadata
+  /// Add the (\a key, \a value) pair to the initial metadata
   /// associated with a server call. These are made available at the client
   /// side by the \a grpc::ClientContext::GetServerTrailingMetadata() method.
   ///
@@ -129,13 +141,13 @@ class ServerContext {
   /// metadata to the client (which happens when the call is finished and a
   /// status is sent to the client).
   ///
-  /// \param meta_key The metadata key. If \a meta_value is binary data,
+  /// \param key The metadata key. If \a value is binary data,
   /// it must end in "-bin".
-  /// \param meta_value The metadata value. If its value is binary, the key name
+  /// \param value The metadata value. If its value is binary, the key name
   /// must end in "-bin".
   void AddTrailingMetadata(const grpc::string& key, const grpc::string& value);
 
-  /// IsCancelled is always safe to call when using sync API.
+  /// IsCancelled is always safe to call when using sync or callback API.
   /// When using async API, it is only safe to call IsCancelled after
   /// the AsyncNotifyWhenDone tag has been delivered.
   bool IsCancelled() const;
@@ -177,9 +189,9 @@ class ServerContext {
     return compression_level_;
   }
 
-  /// Set \a algorithm to be the compression algorithm used for the server call.
+  /// Set \a level to be the compression level used for the server call.
   ///
-  /// \param algorithm The compression algorithm used for the server call.
+  /// \param level The compression level used for the server call.
   void set_compression_level(grpc_compression_level level) {
     compression_level_set_ = true;
     compression_level_ = level;
@@ -265,6 +277,14 @@ class ServerContext {
   friend class ::grpc::internal::ServerStreamingHandler;
   template <class Streamer, bool WriteNeeded>
   friend class ::grpc::internal::TemplatedBidiStreamingHandler;
+  template <class RequestType, class ResponseType>
+  friend class ::grpc::internal::CallbackUnaryHandler;
+  template <class RequestType, class ResponseType>
+  friend class ::grpc::internal::CallbackClientStreamingHandler;
+  template <class RequestType, class ResponseType>
+  friend class ::grpc::internal::CallbackServerStreamingHandler;
+  template <class RequestType, class ResponseType>
+  friend class ::grpc::internal::CallbackBidiHandler;
   template <StatusCode code>
   friend class internal::ErrorMethodHandler;
   friend class ::grpc::ClientContext;
@@ -275,7 +295,9 @@ class ServerContext {
 
   class CompletionOp;
 
-  void BeginCompletionOp(internal::Call* call);
+  void BeginCompletionOp(internal::Call* call,
+                         std::function<void(bool)> callback,
+                         internal::ServerReactor* reactor);
   /// Return the tag queued by BeginCompletionOp()
   internal::CompletionQueueTag* GetCompletionOpTag();
 
@@ -283,11 +305,30 @@ class ServerContext {
 
   void set_call(grpc_call* call) { call_ = call; }
 
+  void BindDeadlineAndMetadata(gpr_timespec deadline, grpc_metadata_array* arr);
+
+  void Clear();
+
+  void Setup(gpr_timespec deadline);
+
   uint32_t initial_metadata_flags() const { return 0; }
 
+  experimental::ServerRpcInfo* set_server_rpc_info(
+      const char* method, internal::RpcMethod::RpcType type,
+      const std::vector<
+          std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>&
+          creators) {
+    if (creators.size() != 0) {
+      rpc_info_ = new experimental::ServerRpcInfo(this, method, type);
+      rpc_info_->RegisterInterceptors(creators);
+    }
+    return rpc_info_;
+  }
+
   CompletionOp* completion_op_;
   bool has_notify_when_done_tag_;
   void* async_notify_when_done_tag_;
+  internal::CallbackWithSuccessTag completion_tag_;
 
   gpr_timespec deadline_;
   grpc_call* call_;
@@ -306,6 +347,8 @@ class ServerContext {
                       internal::CallOpSendMessage>
       pending_ops_;
   bool has_pending_ops_;
+
+  experimental::ServerRpcInfo* rpc_info_;
 };
 
 }  // namespace grpc
diff --git a/include/grpcpp/impl/codegen/server_interceptor.h b/include/grpcpp/impl/codegen/server_interceptor.h
new file mode 100644
index 0000000000..cd7c0600b6
--- /dev/null
+++ b/include/grpcpp/impl/codegen/server_interceptor.h
@@ -0,0 +1,118 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_IMPL_CODEGEN_SERVER_INTERCEPTOR_H
+#define GRPCPP_IMPL_CODEGEN_SERVER_INTERCEPTOR_H
+
+#include <atomic>
+#include <vector>
+
+#include <grpcpp/impl/codegen/interceptor.h>
+#include <grpcpp/impl/codegen/rpc_method.h>
+#include <grpcpp/impl/codegen/string_ref.h>
+
+namespace grpc {
+
+class ServerContext;
+
+namespace internal {
+class InterceptorBatchMethodsImpl;
+}
+
+namespace experimental {
+class ServerRpcInfo;
+
+class ServerInterceptorFactoryInterface {
+ public:
+  virtual ~ServerInterceptorFactoryInterface() {}
+  virtual Interceptor* CreateServerInterceptor(ServerRpcInfo* info) = 0;
+};
+
+class ServerRpcInfo {
+ public:
+  enum class Type { UNARY, CLIENT_STREAMING, SERVER_STREAMING, BIDI_STREAMING };
+
+  ~ServerRpcInfo(){};
+
+  ServerRpcInfo(const ServerRpcInfo&) = delete;
+  ServerRpcInfo(ServerRpcInfo&&) = default;
+  ServerRpcInfo& operator=(ServerRpcInfo&&) = default;
+
+  // Getter methods
+  const char* method() const { return method_; }
+  Type type() const { return type_; }
+  grpc::ServerContext* server_context() { return ctx_; }
+
+ private:
+  static_assert(Type::UNARY ==
+                    static_cast<Type>(internal::RpcMethod::NORMAL_RPC),
+                "violated expectation about Type enum");
+  static_assert(Type::CLIENT_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::CLIENT_STREAMING),
+                "violated expectation about Type enum");
+  static_assert(Type::SERVER_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::SERVER_STREAMING),
+                "violated expectation about Type enum");
+  static_assert(Type::BIDI_STREAMING ==
+                    static_cast<Type>(internal::RpcMethod::BIDI_STREAMING),
+                "violated expectation about Type enum");
+
+  ServerRpcInfo(grpc::ServerContext* ctx, const char* method,
+                internal::RpcMethod::RpcType type)
+      : ctx_(ctx), method_(method), type_(static_cast<Type>(type)) {
+    ref_.store(1);
+  }
+
+  // Runs interceptor at pos \a pos.
+  void RunInterceptor(
+      experimental::InterceptorBatchMethods* interceptor_methods, size_t pos) {
+    GPR_CODEGEN_ASSERT(pos < interceptors_.size());
+    interceptors_[pos]->Intercept(interceptor_methods);
+  }
+
+  void RegisterInterceptors(
+      const std::vector<
+          std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>&
+          creators) {
+    for (const auto& creator : creators) {
+      interceptors_.push_back(std::unique_ptr<experimental::Interceptor>(
+          creator->CreateServerInterceptor(this)));
+    }
+  }
+
+  void Ref() { ref_++; }
+  void Unref() {
+    if (--ref_ == 0) {
+      delete this;
+    }
+  }
+
+  grpc::ServerContext* ctx_ = nullptr;
+  const char* method_ = nullptr;
+  const Type type_;
+  std::atomic_int ref_;
+  std::vector<std::unique_ptr<experimental::Interceptor>> interceptors_;
+
+  friend class internal::InterceptorBatchMethodsImpl;
+  friend class grpc::ServerContext;
+};
+
+}  // namespace experimental
+}  // namespace grpc
+
+#endif  // GRPCPP_IMPL_CODEGEN_SERVER_INTERCEPTOR_H
diff --git a/include/grpcpp/impl/codegen/server_interface.h b/include/grpcpp/impl/codegen/server_interface.h
index 237991cde6..e0e2629827 100644
--- a/include/grpcpp/impl/codegen/server_interface.h
+++ b/include/grpcpp/impl/codegen/server_interface.h
@@ -21,10 +21,12 @@
 
 #include <grpc/impl/codegen/grpc_types.h>
 #include <grpcpp/impl/codegen/byte_buffer.h>
+#include <grpcpp/impl/codegen/call.h>
 #include <grpcpp/impl/codegen/call_hook.h>
 #include <grpcpp/impl/codegen/completion_queue_tag.h>
 #include <grpcpp/impl/codegen/core_codegen_interface.h>
 #include <grpcpp/impl/codegen/rpc_service_method.h>
+#include <grpcpp/impl/codegen/server_context.h>
 
 namespace grpc {
 
@@ -148,44 +150,69 @@ class ServerInterface : public internal::CallHook {
    public:
     BaseAsyncRequest(ServerInterface* server, ServerContext* context,
                      internal::ServerAsyncStreamingInterface* stream,
-                     CompletionQueue* call_cq, void* tag,
+                     CompletionQueue* call_cq,
+                     ServerCompletionQueue* notification_cq, void* tag,
                      bool delete_on_finalize);
     virtual ~BaseAsyncRequest();
 
     bool FinalizeResult(void** tag, bool* status) override;
 
+   private:
+    void ContinueFinalizeResultAfterInterception();
+
    protected:
     ServerInterface* const server_;
     ServerContext* const context_;
     internal::ServerAsyncStreamingInterface* const stream_;
     CompletionQueue* const call_cq_;
+    ServerCompletionQueue* const notification_cq_;
     void* const tag_;
     const bool delete_on_finalize_;
     grpc_call* call_;
+    internal::Call call_wrapper_;
+    internal::InterceptorBatchMethodsImpl interceptor_methods_;
+    bool done_intercepting_;
   };
 
+  /// RegisteredAsyncRequest is not part of the C++ API
   class RegisteredAsyncRequest : public BaseAsyncRequest {
    public:
     RegisteredAsyncRequest(ServerInterface* server, ServerContext* context,
                            internal::ServerAsyncStreamingInterface* stream,
-                           CompletionQueue* call_cq, void* tag);
-
-    // uses BaseAsyncRequest::FinalizeResult
+                           CompletionQueue* call_cq,
+                           ServerCompletionQueue* notification_cq, void* tag,
+                           const char* name, internal::RpcMethod::RpcType type);
+
+    virtual bool FinalizeResult(void** tag, bool* status) override {
+      /* If we are done intercepting, then there is nothing more for us to do */
+      if (done_intercepting_) {
+        return BaseAsyncRequest::FinalizeResult(tag, status);
+      }
+      call_wrapper_ = internal::Call(
+          call_, server_, call_cq_, server_->max_receive_message_size(),
+          context_->set_server_rpc_info(name_, type_,
+                                        *server_->interceptor_creators()));
+      return BaseAsyncRequest::FinalizeResult(tag, status);
+    }
 
    protected:
     void IssueRequest(void* registered_method, grpc_byte_buffer** payload,
                       ServerCompletionQueue* notification_cq);
+    const char* name_;
+    const internal::RpcMethod::RpcType type_;
   };
 
   class NoPayloadAsyncRequest final : public RegisteredAsyncRequest {
    public:
-    NoPayloadAsyncRequest(void* registered_method, ServerInterface* server,
-                          ServerContext* context,
+    NoPayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
+                          ServerInterface* server, ServerContext* context,
                           internal::ServerAsyncStreamingInterface* stream,
                           CompletionQueue* call_cq,
                           ServerCompletionQueue* notification_cq, void* tag)
-        : RegisteredAsyncRequest(server, context, stream, call_cq, tag) {
-      IssueRequest(registered_method, nullptr, notification_cq);
+        : RegisteredAsyncRequest(
+              server, context, stream, call_cq, notification_cq, tag,
+              registered_method->name(), registered_method->method_type()) {
+      IssueRequest(registered_method->server_tag(), nullptr, notification_cq);
     }
 
     // uses RegisteredAsyncRequest::FinalizeResult
@@ -194,13 +221,15 @@ class ServerInterface : public internal::CallHook {
   template <class Message>
   class PayloadAsyncRequest final : public RegisteredAsyncRequest {
    public:
-    PayloadAsyncRequest(void* registered_method, ServerInterface* server,
-                        ServerContext* context,
+    PayloadAsyncRequest(internal::RpcServiceMethod* registered_method,
+                        ServerInterface* server, ServerContext* context,
                         internal::ServerAsyncStreamingInterface* stream,
                         CompletionQueue* call_cq,
                         ServerCompletionQueue* notification_cq, void* tag,
                         Message* request)
-        : RegisteredAsyncRequest(server, context, stream, call_cq, tag),
+        : RegisteredAsyncRequest(
+              server, context, stream, call_cq, notification_cq, tag,
+              registered_method->name(), registered_method->method_type()),
           registered_method_(registered_method),
           server_(server),
           context_(context),
@@ -209,7 +238,8 @@ class ServerInterface : public internal::CallHook {
           notification_cq_(notification_cq),
           tag_(tag),
           request_(request) {
-      IssueRequest(registered_method, payload_.bbuf_ptr(), notification_cq);
+      IssueRequest(registered_method->server_tag(), payload_.bbuf_ptr(),
+                   notification_cq);
     }
 
     ~PayloadAsyncRequest() {
@@ -217,6 +247,10 @@ class ServerInterface : public internal::CallHook {
     }
 
     bool FinalizeResult(void** tag, bool* status) override {
+      /* If we are done intercepting, then there is nothing more for us to do */
+      if (done_intercepting_) {
+        return RegisteredAsyncRequest::FinalizeResult(tag, status);
+      }
       if (*status) {
         if (!payload_.Valid() || !SerializationTraits<Message>::Deserialize(
                                       payload_.bbuf_ptr(), request_)
@@ -235,15 +269,20 @@ class ServerInterface : public internal::CallHook {
           return false;
         }
       }
+      /* Set interception point for recv message */
+      interceptor_methods_.AddInterceptionHookPoint(
+          experimental::InterceptionHookPoints::POST_RECV_MESSAGE);
+      interceptor_methods_.SetRecvMessage(request_);
       return RegisteredAsyncRequest::FinalizeResult(tag, status);
     }
 
    private:
-    void* const registered_method_;
+    internal::RpcServiceMethod* const registered_method_;
     ServerInterface* const server_;
     ServerContext* const context_;
     internal::ServerAsyncStreamingInterface* const stream_;
     CompletionQueue* const call_cq_;
+
     ServerCompletionQueue* const notification_cq_;
     void* const tag_;
     Message* const request_;
@@ -272,9 +311,8 @@ class ServerInterface : public internal::CallHook {
                         ServerCompletionQueue* notification_cq, void* tag,
                         Message* message) {
     GPR_CODEGEN_ASSERT(method);
-    new PayloadAsyncRequest<Message>(method->server_tag(), this, context,
-                                     stream, call_cq, notification_cq, tag,
-                                     message);
+    new PayloadAsyncRequest<Message>(method, this, context, stream, call_cq,
+                                     notification_cq, tag, message);
   }
 
   void RequestAsyncCall(internal::RpcServiceMethod* method,
@@ -283,8 +321,8 @@ class ServerInterface : public internal::CallHook {
                         CompletionQueue* call_cq,
                         ServerCompletionQueue* notification_cq, void* tag) {
     GPR_CODEGEN_ASSERT(method);
-    new NoPayloadAsyncRequest(method->server_tag(), this, context, stream,
-                              call_cq, notification_cq, tag);
+    new NoPayloadAsyncRequest(method, this, context, stream, call_cq,
+                              notification_cq, tag);
   }
 
   void RequestAsyncGenericCall(GenericServerContext* context,
@@ -295,6 +333,28 @@ class ServerInterface : public internal::CallHook {
     new GenericAsyncRequest(this, context, stream, call_cq, notification_cq,
                             tag, true);
   }
+
+ private:
+  // EXPERIMENTAL
+  // Getter method for the vector of interceptor factory objects.
+  // Returns a nullptr (rather than being pure) since this is a post-1.0 method
+  // and adding a new pure method to an interface would be a breaking change
+  // (even though this is private and non-API)
+  virtual std::vector<
+      std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>*
+  interceptor_creators() {
+    return nullptr;
+  }
+
+  // EXPERIMENTAL
+  // A method to get the callbackable completion queue associated with this
+  // server. If the return value is nullptr, this server doesn't support
+  // callback operations.
+  // TODO(vjpai): Consider a better default like using a global CQ
+  // Returns nullptr (rather than being pure) since this is a post-1.0 method
+  // and adding a new pure method to an interface would be a breaking change
+  // (even though this is private and non-API)
+  virtual CompletionQueue* CallbackCQ() { return nullptr; }
 };
 
 }  // namespace grpc
diff --git a/include/grpcpp/impl/codegen/service_type.h b/include/grpcpp/impl/codegen/service_type.h
index 9f1a052168..332a04c294 100644
--- a/include/grpcpp/impl/codegen/service_type.h
+++ b/include/grpcpp/impl/codegen/service_type.h
@@ -71,7 +71,20 @@ class Service {
 
   bool has_synchronous_methods() const {
     for (auto it = methods_.begin(); it != methods_.end(); ++it) {
-      if (*it && (*it)->handler() != nullptr) {
+      if (*it &&
+          (*it)->api_type() == internal::RpcServiceMethod::ApiType::SYNC) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  bool has_callback_methods() const {
+    for (auto it = methods_.begin(); it != methods_.end(); ++it) {
+      if (*it && ((*it)->api_type() ==
+                      internal::RpcServiceMethod::ApiType::CALL_BACK ||
+                  (*it)->api_type() ==
+                      internal::RpcServiceMethod::ApiType::RAW_CALL_BACK)) {
         return true;
       }
     }
@@ -88,6 +101,43 @@ class Service {
   }
 
  protected:
+  // TODO(vjpai): Promote experimental contents once callback API is accepted
+  class experimental_type {
+   public:
+    explicit experimental_type(Service* service) : service_(service) {}
+
+    void MarkMethodCallback(int index, internal::MethodHandler* handler) {
+      // This does not have to be a hard error, however no one has approached us
+      // with a use case yet. Please file an issue if you believe you have one.
+      size_t idx = static_cast<size_t>(index);
+      GPR_CODEGEN_ASSERT(
+          service_->methods_[idx].get() != nullptr &&
+          "Cannot mark the method as 'callback' because it has already been "
+          "marked as 'generic'.");
+      service_->methods_[idx]->SetHandler(handler);
+      service_->methods_[idx]->SetServerApiType(
+          internal::RpcServiceMethod::ApiType::CALL_BACK);
+    }
+
+    void MarkMethodRawCallback(int index, internal::MethodHandler* handler) {
+      // This does not have to be a hard error, however no one has approached us
+      // with a use case yet. Please file an issue if you believe you have one.
+      size_t idx = static_cast<size_t>(index);
+      GPR_CODEGEN_ASSERT(
+          service_->methods_[idx].get() != nullptr &&
+          "Cannot mark the method as 'raw callback' because it has already "
+          "been marked as 'generic'.");
+      service_->methods_[idx]->SetHandler(handler);
+      service_->methods_[idx]->SetServerApiType(
+          internal::RpcServiceMethod::ApiType::RAW_CALL_BACK);
+    }
+
+   private:
+    Service* service_;
+  };
+
+  experimental_type experimental() { return experimental_type(this); }
+
   template <class Message>
   void RequestAsyncUnary(int index, ServerContext* context, Message* request,
                          internal::ServerAsyncStreamingInterface* stream,
@@ -138,8 +188,7 @@ class Service {
         methods_[idx].get() != nullptr &&
         "Cannot mark the method as 'async' because it has already been "
         "marked as 'generic'.");
-    methods_[idx]->SetServerAsyncType(
-        internal::RpcServiceMethod::AsyncType::ASYNC);
+    methods_[idx]->SetServerApiType(internal::RpcServiceMethod::ApiType::ASYNC);
   }
 
   void MarkMethodRaw(int index) {
@@ -149,8 +198,7 @@ class Service {
     GPR_CODEGEN_ASSERT(methods_[idx].get() != nullptr &&
                        "Cannot mark the method as 'raw' because it has already "
                        "been marked as 'generic'.");
-    methods_[idx]->SetServerAsyncType(
-        internal::RpcServiceMethod::AsyncType::RAW);
+    methods_[idx]->SetServerApiType(internal::RpcServiceMethod::ApiType::RAW);
   }
 
   void MarkMethodGeneric(int index) {
diff --git a/include/grpcpp/impl/codegen/sync_stream.h b/include/grpcpp/impl/codegen/sync_stream.h
index cbfcf25d0a..6981076f04 100644
--- a/include/grpcpp/impl/codegen/sync_stream.h
+++ b/include/grpcpp/impl/codegen/sync_stream.h
@@ -250,7 +250,7 @@ class ClientReader final : public ClientReaderInterface<R> {
                                 ::grpc::internal::CallOpSendMessage,
                                 ::grpc::internal::CallOpClientSendClose>
         ops;
-    ops.SendInitialMetadata(context->send_initial_metadata_,
+    ops.SendInitialMetadata(&context->send_initial_metadata_,
                             context->initial_metadata_flags());
     // TODO(ctiller): don't assert
     GPR_CODEGEN_ASSERT(ops.SendMessage(request).ok());
@@ -327,7 +327,7 @@ class ClientWriter : public ClientWriterInterface<W> {
       ops.ClientSendClose();
     }
     if (context_->initial_metadata_corked_) {
-      ops.SendInitialMetadata(context_->send_initial_metadata_,
+      ops.SendInitialMetadata(&context_->send_initial_metadata_,
                               context_->initial_metadata_flags());
       context_->set_initial_metadata_corked(false);
     }
@@ -386,7 +386,7 @@ class ClientWriter : public ClientWriterInterface<W> {
     if (!context_->initial_metadata_corked_) {
       ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
           ops;
-      ops.SendInitialMetadata(context->send_initial_metadata_,
+      ops.SendInitialMetadata(&context->send_initial_metadata_,
                               context->initial_metadata_flags());
       call_.PerformOps(&ops);
       cq_.Pluck(&ops);
@@ -498,7 +498,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
       ops.ClientSendClose();
     }
     if (context_->initial_metadata_corked_) {
-      ops.SendInitialMetadata(context_->send_initial_metadata_,
+      ops.SendInitialMetadata(&context_->send_initial_metadata_,
                               context_->initial_metadata_flags());
       context_->set_initial_metadata_corked(false);
     }
@@ -557,7 +557,7 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
     if (!context_->initial_metadata_corked_) {
       ::grpc::internal::CallOpSet<::grpc::internal::CallOpSendInitialMetadata>
           ops;
-      ops.SendInitialMetadata(context->send_initial_metadata_,
+      ops.SendInitialMetadata(&context->send_initial_metadata_,
                               context->initial_metadata_flags());
       call_.PerformOps(&ops);
       cq_.Pluck(&ops);
@@ -583,7 +583,7 @@ class ServerReader final : public ServerReaderInterface<R> {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     internal::CallOpSet<internal::CallOpSendInitialMetadata> ops;
-    ops.SendInitialMetadata(ctx_->initial_metadata_,
+    ops.SendInitialMetadata(&ctx_->initial_metadata_,
                             ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       ops.set_compression_level(ctx_->compression_level());
@@ -635,7 +635,7 @@ class ServerWriter final : public ServerWriterInterface<W> {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     internal::CallOpSet<internal::CallOpSendInitialMetadata> ops;
-    ops.SendInitialMetadata(ctx_->initial_metadata_,
+    ops.SendInitialMetadata(&ctx_->initial_metadata_,
                             ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       ops.set_compression_level(ctx_->compression_level());
@@ -660,7 +660,7 @@ class ServerWriter final : public ServerWriterInterface<W> {
       return false;
     }
     if (!ctx_->sent_initial_metadata_) {
-      ctx_->pending_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+      ctx_->pending_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                              ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         ctx_->pending_ops_.set_compression_level(ctx_->compression_level());
@@ -708,7 +708,7 @@ class ServerReaderWriterBody final {
     GPR_CODEGEN_ASSERT(!ctx_->sent_initial_metadata_);
 
     CallOpSet<CallOpSendInitialMetadata> ops;
-    ops.SendInitialMetadata(ctx_->initial_metadata_,
+    ops.SendInitialMetadata(&ctx_->initial_metadata_,
                             ctx_->initial_metadata_flags());
     if (ctx_->compression_level_set()) {
       ops.set_compression_level(ctx_->compression_level());
@@ -738,7 +738,7 @@ class ServerReaderWriterBody final {
       return false;
     }
     if (!ctx_->sent_initial_metadata_) {
-      ctx_->pending_ops_.SendInitialMetadata(ctx_->initial_metadata_,
+      ctx_->pending_ops_.SendInitialMetadata(&ctx_->initial_metadata_,
                                              ctx_->initial_metadata_flags());
       if (ctx_->compression_level_set()) {
         ctx_->pending_ops_.set_compression_level(ctx_->compression_level());
diff --git a/include/grpcpp/security/credentials.h b/include/grpcpp/security/credentials.h
index 8dfbdec3e6..d8c9e04d77 100644
--- a/include/grpcpp/security/credentials.h
+++ b/include/grpcpp/security/credentials.h
@@ -46,8 +46,8 @@ std::shared_ptr<Channel> CreateCustomChannelWithInterceptors(
     const grpc::string& target,
     const std::shared_ptr<ChannelCredentials>& creds,
     const ChannelArguments& args,
-    std::unique_ptr<std::vector<
-        std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+    std::vector<
+        std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
         interceptor_creators);
 }  // namespace experimental
 
@@ -80,8 +80,8 @@ class ChannelCredentials : private GrpcLibraryCodegen {
       const grpc::string& target,
       const std::shared_ptr<ChannelCredentials>& creds,
       const ChannelArguments& args,
-      std::unique_ptr<std::vector<
-          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+      std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
           interceptor_creators);
 
   virtual std::shared_ptr<Channel> CreateChannel(
@@ -91,8 +91,8 @@ class ChannelCredentials : private GrpcLibraryCodegen {
   // implemented as a virtual function so that it does not break API.
   virtual std::shared_ptr<Channel> CreateChannelWithInterceptors(
       const grpc::string& target, const ChannelArguments& args,
-      std::unique_ptr<std::vector<
-          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+      std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
           interceptor_creators) {
     return nullptr;
   };
diff --git a/include/grpcpp/server.h b/include/grpcpp/server.h
index 8d3e856502..cdcac186cb 100644
--- a/include/grpcpp/server.h
+++ b/include/grpcpp/server.h
@@ -111,8 +111,8 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
     /// interceptors
     std::shared_ptr<Channel> InProcessChannelWithInterceptors(
         const ChannelArguments& args,
-        std::unique_ptr<std::vector<
-            std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+        std::vector<
+            std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>
             interceptor_creators);
 
    private:
@@ -174,7 +174,11 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
          std::shared_ptr<std::vector<std::unique_ptr<ServerCompletionQueue>>>
              sync_server_cqs,
          int min_pollers, int max_pollers, int sync_cq_timeout_msec,
-         grpc_resource_quota* server_rq = nullptr);
+         grpc_resource_quota* server_rq = nullptr,
+         std::vector<
+             std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
+             interceptor_creators = std::vector<std::unique_ptr<
+                 experimental::ServerInterceptorFactoryInterface>>());
 
   /// Start the server.
   ///
@@ -187,11 +191,17 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
   grpc_server* server() override { return server_; };
 
  private:
+  std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>*
+  interceptor_creators() override {
+    return &interceptor_creators_;
+  }
+
   friend class AsyncGenericService;
   friend class ServerBuilder;
   friend class ServerInitializer;
 
   class SyncRequest;
+  class CallbackRequest;
   class UnimplementedAsyncRequest;
   class UnimplementedAsyncResponse;
 
@@ -214,8 +224,18 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
     return max_receive_message_size_;
   };
 
+  CompletionQueue* CallbackCQ() override;
+
   ServerInitializer* initializer();
 
+  // A vector of interceptor factory objects.
+  // This should be destroyed after health_check_service_ and this requirement
+  // is satisfied by declaring interceptor_creators_ before
+  // health_check_service_. (C++ mandates that member objects be destroyed in
+  // the reverse order of initialization.)
+  std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
+      interceptor_creators_;
+
   const int max_receive_message_size_;
 
   /// The following completion queues are ONLY used in case of Sync API
@@ -228,6 +248,9 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
   /// the \a sync_server_cqs)
   std::vector<std::unique_ptr<SyncRequestThreadManager>> sync_req_mgrs_;
 
+  /// Outstanding callback requests
+  std::vector<std::unique_ptr<CallbackRequest>> callback_reqs_;
+
   // Server status
   std::mutex mu_;
   bool started_;
@@ -251,6 +274,13 @@ class Server : public ServerInterface, private GrpcLibraryCodegen {
 
   // A special handler for resource exhausted in sync case
   std::unique_ptr<internal::MethodHandler> resource_exhausted_handler_;
+
+  // callback_cq_ references the callbackable completion queue associated
+  // with this server (if any). It is set on the first call to CallbackCQ().
+  // It is _not owned_ by the server; ownership belongs with its internal
+  // shutdown callback tag (invoked when the CQ is fully shutdown).
+  // It is protected by mu_
+  CompletionQueue* callback_cq_ = nullptr;
 };
 
 }  // namespace grpc
diff --git a/include/grpcpp/server_builder.h b/include/grpcpp/server_builder.h
index a58a59c2d8..028b8cffaa 100644
--- a/include/grpcpp/server_builder.h
+++ b/include/grpcpp/server_builder.h
@@ -28,6 +28,7 @@
 #include <grpc/support/cpu.h>
 #include <grpc/support/workaround_list.h>
 #include <grpcpp/impl/channel_argument_option.h>
+#include <grpcpp/impl/codegen/server_interceptor.h>
 #include <grpcpp/impl/server_builder_option.h>
 #include <grpcpp/impl/server_builder_plugin.h>
 #include <grpcpp/support/config.h>
@@ -212,6 +213,29 @@ class ServerBuilder {
   /// doc/workarounds.md.
   ServerBuilder& EnableWorkaround(grpc_workaround_list id);
 
+  /// NOTE: class experimental_type is not part of the public API of this class.
+  /// TODO(yashykt): Integrate into public API when this is no longer
+  /// experimental.
+  class experimental_type {
+   public:
+    explicit experimental_type(ServerBuilder* builder) : builder_(builder) {}
+
+    void SetInterceptorCreators(
+        std::vector<
+            std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
+            interceptor_creators) {
+      builder_->interceptor_creators_ = std::move(interceptor_creators);
+    }
+
+   private:
+    ServerBuilder* builder_;
+  };
+
+  /// NOTE: The function experimental() is not stable public API. It is a view
+  /// to the experimental components of this class. It may be changed or removed
+  /// at any time.
+  experimental_type experimental() { return experimental_type(this); }
+
  protected:
   /// Experimental, to be deprecated
   struct Port {
@@ -297,6 +321,8 @@ class ServerBuilder {
     grpc_compression_algorithm algorithm;
   } maybe_default_compression_algorithm_;
   uint32_t enabled_compression_algorithms_bitset_;
+  std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>>
+      interceptor_creators_;
 };
 
 }  // namespace grpc
diff --git a/include/grpcpp/support/server_callback.h b/include/grpcpp/support/server_callback.h
new file mode 100644
index 0000000000..b0aeeb53c5
--- /dev/null
+++ b/include/grpcpp/support/server_callback.h
@@ -0,0 +1,24 @@
+/*
+ *
+ * Copyright 2018 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#ifndef GRPCPP_SUPPORT_SERVER_CALLBACK_H
+#define GRPCPP_SUPPORT_SERVER_CALLBACK_H
+
+#include <grpcpp/impl/codegen/server_callback.h>
+
+#endif  // GRPCPP_SUPPORT_SERVER_CALLBACK_H