Finish hybrid server stuff, ensure it gets tested

5 files changed, 41 insertions, 22 deletions

diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h
index a5a4f3d739..4bf35ae778 100644
--- a/include/grpc++/impl/codegen/client_unary_call.h
+++ b/include/grpc++/impl/codegen/client_unary_call.h
@@ -52,7 +52,9 @@ template <class InputMessage, class OutputMessage>
 Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
                          ClientContext* context, const InputMessage& request,
                          OutputMessage* result) {
-  CompletionQueue cq(true);  // Pluckable completion queue
+  CompletionQueue cq(grpc_completion_queue_attributes{
+      GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+      GRPC_CQ_DEFAULT_POLLING});  // Pluckable completion queue
   Call call(channel->CreateCall(method, context, &cq));
   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
             CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index 906a64693e..c8ab726b0f 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -102,7 +102,9 @@ class CompletionQueue : private GrpcLibraryCodegen {
  public:
   /// Default constructor. Implicitly creates a \a grpc_completion_queue
   /// instance.
-  CompletionQueue() : CompletionQueue(false) {}
+  CompletionQueue()
+      : CompletionQueue(grpc_completion_queue_attributes{
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {}
 
   /// Wrap \a take, taking ownership of the instance.
   ///
@@ -182,6 +184,16 @@ class CompletionQueue : private GrpcLibraryCodegen {
   };
   void CompleteAvalanching();
 
+ protected:
+  /// Private constructor of CompletionQueue only visible to friend classes
+  CompletionQueue(const grpc_completion_queue_attributes& attributes) {
+    cq_ = g_core_codegen_interface->grpc_completion_queue_create(
+        g_core_codegen_interface->grpc_completion_queue_factory_lookup(
+            &attributes),
+        &attributes, NULL);
+    InitialAvalanching();  // reserve this for the future shutdown
+  }
+
  private:
   // Friend synchronous wrappers so that they can access Pluck(), which is
   // a semi-private API geared towards the synchronous implementation.
@@ -215,18 +227,6 @@ class CompletionQueue : private GrpcLibraryCodegen {
                                   const InputMessage& request,
                                   OutputMessage* result);
 
-  /// Private constructor of CompletionQueue only visible to friend classes
-  CompletionQueue(bool is_pluck) {
-    if (is_pluck) {
-      cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck(
-          nullptr);
-    } else {
-      cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next(
-          nullptr);
-    }
-    InitialAvalanching();  // reserve this for the future shutdown
-  }
-
   NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
 
   /// Wraps \a grpc_completion_queue_pluck.
@@ -299,11 +299,9 @@ class ServerCompletionQueue : public CompletionQueue {
   /// AsyncNext()). By default all server completion queues are assumed to be
   /// frequently polled.
   ServerCompletionQueue(grpc_cq_polling_type polling_type)
-      : CompletionQueue(MakeCompletionQueue(polling_type)),
+      : CompletionQueue(grpc_completion_queue_attributes{
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}),
         polling_type_(polling_type) {}
-
-  static grpc_completion_queue* MakeCompletionQueue(
-      grpc_cq_polling_type polling_type);
 };
 
 }  // namespace grpc
diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h
index 65151590b2..3cb7da8ef6 100644
--- a/include/grpc++/impl/codegen/core_codegen.h
+++ b/include/grpc++/impl/codegen/core_codegen.h
@@ -44,8 +44,15 @@
 namespace grpc {
 
 /// Implementation of the core codegen interface.
-class CoreCodegen : public CoreCodegenInterface {
+class CoreCodegen final : public CoreCodegenInterface {
  private:
+  virtual const grpc_completion_queue_factory*
+  grpc_completion_queue_factory_lookup(
+      const grpc_completion_queue_attributes* attributes) override;
+  virtual grpc_completion_queue* grpc_completion_queue_create(
+      const grpc_completion_queue_factory* factory,
+      const grpc_completion_queue_attributes* attributes,
+      void* reserved) override;
   grpc_completion_queue* grpc_completion_queue_create_for_next(
       void* reserved) override;
   grpc_completion_queue* grpc_completion_queue_create_for_pluck(
diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h
index 529bef687b..a1a0aaf3ca 100644
--- a/include/grpc++/impl/codegen/core_codegen_interface.h
+++ b/include/grpc++/impl/codegen/core_codegen_interface.h
@@ -59,6 +59,12 @@ class CoreCodegenInterface {
   virtual void assert_fail(const char* failed_assertion, const char* file,
                            int line) = 0;
 
+  virtual const grpc_completion_queue_factory*
+  grpc_completion_queue_factory_lookup(
+      const grpc_completion_queue_attributes* attributes) = 0;
+  virtual grpc_completion_queue* grpc_completion_queue_create(
+      const grpc_completion_queue_factory* factory,
+      const grpc_completion_queue_attributes* attributes, void* reserved) = 0;
   virtual grpc_completion_queue* grpc_completion_queue_create_for_next(
       void* reserved) = 0;
   virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck(
diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h
index 328d5cb1e8..a010924cef 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -156,7 +156,9 @@ class ClientReader final : public ClientReaderInterface<R> {
   ClientReader(ChannelInterface* channel, const RpcMethod& method,
                ClientContext* context, const W& request)
       : context_(context),
-        cq_(true),  // Pluckable cq
+        cq_(grpc_completion_queue_attributes{
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
               CallOpClientSendClose>
@@ -230,7 +232,9 @@ class ClientWriter : public ClientWriterInterface<W> {
   ClientWriter(ChannelInterface* channel, const RpcMethod& method,
                ClientContext* context, R* response)
       : context_(context),
-        cq_(true),  // Pluckable cq
+        cq_(grpc_completion_queue_attributes{
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     finish_ops_.RecvMessage(response);
     finish_ops_.AllowNoMessage();
@@ -330,7 +334,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
   ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
                      ClientContext* context)
       : context_(context),
-        cq_(true),  // Pluckable cq
+        cq_(grpc_completion_queue_attributes{
+            GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+            GRPC_CQ_DEFAULT_POLLING}),  // Pluckable cq
         call_(channel->CreateCall(method, context, &cq_)) {
     if (!context_->initial_metadata_corked_) {
       CallOpSet<CallOpSendInitialMetadata> ops;