Merge remote-tracking branch 'upstream/master' into run_interop_tests_go

69 files changed, 1864 insertions, 1275 deletions

diff --git a/src/ruby/.rubocop.yml b/src/ruby/.rubocop.yml
index 34bb477543..0f61ccfa81 100644
--- a/src/ruby/.rubocop.yml
+++ b/src/ruby/.rubocop.yml
@@ -5,8 +5,8 @@ inherit_from: .rubocop_todo.yml
 AllCops:
   Exclude:
     - 'bin/apis/**/*'
-    - 'bin/math.rb'
-    - 'bin/math_services.rb'
+    - 'bin/math_pb.rb'
+    - 'bin/math_services_pb.rb'
     - 'pb/grpc/health/v1/*'
     - 'pb/test/**/*'
 
diff --git a/src/ruby/bin/math_client.rb b/src/ruby/bin/math_client.rb
index d7e00e4293..1f238a798b 100755
--- a/src/ruby/bin/math_client.rb
+++ b/src/ruby/bin/math_client.rb
@@ -40,7 +40,7 @@ $LOAD_PATH.unshift(lib_dir) unless $LOAD_PATH.include?(lib_dir)
 $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 
 require 'grpc'
-require 'math_services'
+require 'math_services_pb'
 require 'optparse'
 
 include GRPC::Core::TimeConsts
diff --git a/src/ruby/bin/math.rb b/src/ruby/bin/math_pb.rb
index 60429a1505..60429a1505 100755..100644
--- a/src/ruby/bin/math.rb
+++ b/src/ruby/bin/math_pb.rb
diff --git a/src/ruby/bin/math_server.rb b/src/ruby/bin/math_server.rb
index 1ee4c5632d..751a6ebcab 100755
--- a/src/ruby/bin/math_server.rb
+++ b/src/ruby/bin/math_server.rb
@@ -42,7 +42,7 @@ $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 require 'forwardable'
 require 'grpc'
 require 'logger'
-require 'math_services'
+require 'math_services_pb'
 require 'optparse'
 
 # RubyLogger defines a logger for gRPC based on the standard ruby logger.
diff --git a/src/ruby/bin/math_services.rb b/src/ruby/bin/math_services_pb.rb
index 34c36abdda..2ba1825d4f 100755..100644
--- a/src/ruby/bin/math_services.rb
+++ b/src/ruby/bin/math_services_pb.rb
@@ -32,7 +32,7 @@
 #
 
 require 'grpc'
-require 'math'
+require 'math_pb'
 
 module Math
   module Math
@@ -44,15 +44,15 @@ module Math
       self.unmarshal_class_method = :decode
       self.service_name = 'math.Math'
 
-      # Div divides args.dividend by args.divisor and returns the quotient and
-      # remainder.
+      # Div divides DivArgs.dividend by DivArgs.divisor and returns the quotient
+      # and remainder.
       rpc :Div, DivArgs, DivReply
       # DivMany accepts an arbitrary number of division args from the client stream
       # and sends back the results in the reply stream.  The stream continues until
       # the client closes its end; the server does the same after sending all the
       # replies.  The stream ends immediately if either end aborts.
       rpc :DivMany, stream(DivArgs), stream(DivReply)
-      # Fib generates numbers in the Fibonacci sequence.  If args.limit > 0, Fib
+      # Fib generates numbers in the Fibonacci sequence.  If FibArgs.limit > 0, Fib
       # generates up to limit numbers; otherwise it continues until the call is
       # canceled.  Unlike Fib above, Fib has no final FibReply.
       rpc :Fib, FibArgs, stream(Num)
diff --git a/src/ruby/ext/grpc/rb_byte_buffer.c b/src/ruby/ext/grpc/rb_byte_buffer.c
index 1172691116..61b7c30315 100644
--- a/src/ruby/ext/grpc/rb_byte_buffer.c
+++ b/src/ruby/ext/grpc/rb_byte_buffer.c
@@ -56,7 +56,10 @@ VALUE grpc_rb_byte_buffer_to_s(grpc_byte_buffer *buffer) {
     return Qnil;
   }
   rb_string = rb_str_buf_new(grpc_byte_buffer_length(buffer));
-  grpc_byte_buffer_reader_init(&reader, buffer);
+  if (!grpc_byte_buffer_reader_init(&reader, buffer)) {
+    rb_raise(rb_eRuntimeError, "Error initializing byte buffer reader.");
+    return Qnil;
+  }
   while (grpc_byte_buffer_reader_next(&reader, &next) != 0) {
     rb_str_cat(rb_string, (const char *) GPR_SLICE_START_PTR(next),
                GPR_SLICE_LENGTH(next));
diff --git a/src/ruby/ext/grpc/rb_call.c b/src/ruby/ext/grpc/rb_call.c
index b436057c16..67a42af619 100644
--- a/src/ruby/ext/grpc/rb_call.c
+++ b/src/ruby/ext/grpc/rb_call.c
@@ -38,6 +38,7 @@
 
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
+#include <grpc/impl/codegen/compression_types.h>
 
 #include "rb_byte_buffer.h"
 #include "rb_call_credentials.h"
@@ -63,27 +64,18 @@ static VALUE grpc_rb_sBatchResult;
  * grpc_metadata_array. */
 static VALUE grpc_rb_cMdAry;
 
-/* id_cq is the name of the hidden ivar that preserves a reference to a
- * completion queue */
-static ID id_cq;
-
-/* id_flags is the name of the hidden ivar that preserves the value of
- * the flags used to create metadata from a Hash */
-static ID id_flags;
-
 /* id_credentials is the name of the hidden ivar that preserves the value
  * of the credentials added to the call */
 static ID id_credentials;
 
-/* id_input_md is the name of the hidden ivar that preserves the hash used to
- * create metadata, so that references to the strings it contains last as long
- * as the call the metadata is added to. */
-static ID id_input_md;
-
 /* id_metadata is name of the attribute used to access the metadata hash
  * received by the call and subsequently saved on it. */
 static ID id_metadata;
 
+/* id_trailing_metadata is the name of the attribute used to access the trailing
+ * metadata hash received by the call and subsequently saved on it. */
+static ID id_trailing_metadata;
+
 /* id_status is name of the attribute used to access the status object
  * received by the call and subsequently saved on it. */
 static ID id_status;
@@ -101,14 +93,27 @@ static VALUE sym_message;
 static VALUE sym_status;
 static VALUE sym_cancelled;
 
+typedef struct grpc_rb_call {
+  grpc_call *wrapped;
+  grpc_completion_queue *queue;
+} grpc_rb_call;
+
+static void destroy_call(grpc_rb_call *call) {
+  /* Ensure that we only try to destroy the call once */
+  if (call->wrapped != NULL) {
+    grpc_call_destroy(call->wrapped);
+    call->wrapped = NULL;
+    grpc_rb_completion_queue_destroy(call->queue);
+    call->queue = NULL;
+  }
+}
+
 /* Destroys a Call. */
 static void grpc_rb_call_destroy(void *p) {
-  grpc_call* call = NULL;
   if (p == NULL) {
     return;
   }
-  call = (grpc_call *)p;
-  grpc_call_destroy(call);
+  destroy_call((grpc_rb_call*)p);
 }
 
 static size_t md_ary_datasize(const void *p) {
@@ -167,15 +172,15 @@ const char *grpc_call_error_detail_of(grpc_call_error err) {
 /* Called by clients to cancel an RPC on the server.
    Can be called multiple times, from any thread. */
 static VALUE grpc_rb_call_cancel(VALUE self) {
-  grpc_call *call = NULL;
+  grpc_rb_call *call = NULL;
   grpc_call_error err;
   if (RTYPEDDATA_DATA(self) == NULL) {
     //This call has been closed
     return Qnil;
   }
 
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
-  err = grpc_call_cancel(call, NULL);
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
+  err = grpc_call_cancel(call->wrapped, NULL);
   if (err != GRPC_CALL_OK) {
     rb_raise(grpc_rb_eCallError, "cancel failed: %s (code=%d)",
              grpc_call_error_detail_of(err), err);
@@ -189,10 +194,10 @@ static VALUE grpc_rb_call_cancel(VALUE self) {
    processed.
 */
 static VALUE grpc_rb_call_close(VALUE self) {
-  grpc_call *call = NULL;
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
+  grpc_rb_call *call = NULL;
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
   if(call != NULL) {
-    grpc_call_destroy(call);
+    destroy_call(call);
     RTYPEDDATA_DATA(self) = NULL;
   }
   return Qnil;
@@ -201,14 +206,14 @@ static VALUE grpc_rb_call_close(VALUE self) {
 /* Called to obtain the peer that this call is connected to. */
 static VALUE grpc_rb_call_get_peer(VALUE self) {
   VALUE res = Qnil;
-  grpc_call *call = NULL;
+  grpc_rb_call *call = NULL;
   char *peer = NULL;
   if (RTYPEDDATA_DATA(self) == NULL) {
     rb_raise(grpc_rb_eCallError, "Cannot get peer value on closed call");
     return Qnil;
   }
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
-  peer = grpc_call_get_peer(call);
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
+  peer = grpc_call_get_peer(call->wrapped);
   res = rb_str_new2(peer);
   gpr_free(peer);
 
@@ -217,16 +222,16 @@ static VALUE grpc_rb_call_get_peer(VALUE self) {
 
 /* Called to obtain the x509 cert of an authenticated peer. */
 static VALUE grpc_rb_call_get_peer_cert(VALUE self) {
-  grpc_call *call = NULL;
+  grpc_rb_call *call = NULL;
   VALUE res = Qnil;
   grpc_auth_context *ctx = NULL;
   if (RTYPEDDATA_DATA(self) == NULL) {
     rb_raise(grpc_rb_eCallError, "Cannot get peer cert on closed call");
     return Qnil;
   }
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
 
-  ctx = grpc_call_auth_context(call);
+  ctx = grpc_call_auth_context(call->wrapped);
 
   if (!ctx || !grpc_auth_context_peer_is_authenticated(ctx)) {
     return Qnil;
@@ -298,6 +303,30 @@ static VALUE grpc_rb_call_set_metadata(VALUE self, VALUE metadata) {
 
 /*
   call-seq:
+  trailing_metadata = call.trailing_metadata
+
+  Gets the trailing metadata object saved on the call */
+static VALUE grpc_rb_call_get_trailing_metadata(VALUE self) {
+  return rb_ivar_get(self, id_trailing_metadata);
+}
+
+/*
+  call-seq:
+  call.trailing_metadata = trailing_metadata
+
+  Saves the trailing metadata hash on the call. */
+static VALUE grpc_rb_call_set_trailing_metadata(VALUE self, VALUE metadata) {
+  if (!NIL_P(metadata) && TYPE(metadata) != T_HASH) {
+    rb_raise(rb_eTypeError, "bad metadata: got:<%s> want: <Hash>",
+             rb_obj_classname(metadata));
+    return Qnil;
+  }
+
+  return rb_ivar_set(self, id_trailing_metadata, metadata);
+}
+
+/*
+  call-seq:
   write_flag = call.write_flag
 
   Gets the write_flag value saved the call.  */
@@ -326,21 +355,23 @@ static VALUE grpc_rb_call_set_write_flag(VALUE self, VALUE write_flag) {
 
   Sets credentials on a call */
 static VALUE grpc_rb_call_set_credentials(VALUE self, VALUE credentials) {
-  grpc_call *call = NULL;
+  grpc_rb_call *call = NULL;
   grpc_call_credentials *creds;
   grpc_call_error err;
   if (RTYPEDDATA_DATA(self) == NULL) {
     rb_raise(grpc_rb_eCallError, "Cannot set credentials of closed call");
     return Qnil;
   }
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
   creds = grpc_rb_get_wrapped_call_credentials(credentials);
-  err = grpc_call_set_credentials(call, creds);
+  err = grpc_call_set_credentials(call->wrapped, creds);
   if (err != GRPC_CALL_OK) {
     rb_raise(grpc_rb_eCallError,
              "grpc_call_set_credentials failed with %s (code=%d)",
              grpc_call_error_detail_of(err), err);
   }
+  /* We need the credentials to be alive for as long as the call is alive,
+     but we don't care about destruction order. */
   rb_ivar_set(self, id_credentials, credentials);
   return Qnil;
 }
@@ -733,7 +764,6 @@ static VALUE grpc_run_batch_stack_build_result(run_batch_stack *st) {
 }
 
 /* call-seq:
-   cq = CompletionQueue.new
    ops = {
      GRPC::Core::CallOps::SEND_INITIAL_METADATA => <op_value>,
      GRPC::Core::CallOps::SEND_MESSAGE => <op_value>,
@@ -741,7 +771,7 @@ static VALUE grpc_run_batch_stack_build_result(run_batch_stack *st) {
    }
    tag = Object.new
    timeout = 10
-   call.start_batch(cq, tag, timeout, ops)
+   call.start_batch(tag, timeout, ops)
 
    Start a batch of operations defined in the array ops; when complete, post a
    completion of type 'tag' to the completion queue bound to the call.
@@ -750,20 +780,20 @@ static VALUE grpc_run_batch_stack_build_result(run_batch_stack *st) {
    The order of ops specified in the batch has no significance.
    Only one operation of each type can be active at once in any given
    batch */
-static VALUE grpc_rb_call_run_batch(VALUE self, VALUE cqueue, VALUE tag,
-                                    VALUE timeout, VALUE ops_hash) {
+static VALUE grpc_rb_call_run_batch(VALUE self, VALUE ops_hash) {
   run_batch_stack st;
-  grpc_call *call = NULL;
+  grpc_rb_call *call = NULL;
   grpc_event ev;
   grpc_call_error err;
   VALUE result = Qnil;
   VALUE rb_write_flag = rb_ivar_get(self, id_write_flag);
   unsigned write_flag = 0;
+  void *tag = (void*)&st;
   if (RTYPEDDATA_DATA(self) == NULL) {
     rb_raise(grpc_rb_eCallError, "Cannot run batch on closed call");
     return Qnil;
   }
-  TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
+  TypedData_Get_Struct(self, grpc_rb_call, &grpc_call_data_type, call);
 
   /* Validate the ops args, adding them to a ruby array */
   if (TYPE(ops_hash) != T_HASH) {
@@ -778,7 +808,7 @@ static VALUE grpc_rb_call_run_batch(VALUE self, VALUE cqueue, VALUE tag,
 
   /* call grpc_call_start_batch, then wait for it to complete using
    * pluck_event */
-  err = grpc_call_start_batch(call, st.ops, st.op_num, ROBJECT(tag), NULL);
+  err = grpc_call_start_batch(call->wrapped, st.ops, st.op_num, tag, NULL);
   if (err != GRPC_CALL_OK) {
     grpc_run_batch_stack_cleanup(&st);
     rb_raise(grpc_rb_eCallError,
@@ -786,13 +816,11 @@ static VALUE grpc_rb_call_run_batch(VALUE self, VALUE cqueue, VALUE tag,
              grpc_call_error_detail_of(err), err);
     return Qnil;
   }
-  ev = grpc_rb_completion_queue_pluck_event(cqueue, tag, timeout);
-  if (ev.type == GRPC_QUEUE_TIMEOUT) {
-    grpc_run_batch_stack_cleanup(&st);
-    rb_raise(grpc_rb_eOutOfTime, "grpc_call_start_batch timed out");
-    return Qnil;
+  ev = rb_completion_queue_pluck(call->queue, tag,
+                                 gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+  if (!ev.success) {
+    rb_raise(grpc_rb_eCallError, "call#run_batch failed somehow");
   }
-
   /* Build and return the BatchResult struct result,
      if there is an error, it's reflected in the status */
   result = grpc_run_batch_stack_build_result(&st);
@@ -883,6 +911,12 @@ static void Init_grpc_op_codes() {
                   UINT2NUM(GRPC_OP_RECV_CLOSE_ON_SERVER));
 }
 
+static void Init_grpc_metadata_keys() {
+  VALUE grpc_rb_mMetadataKeys = rb_define_module_under(grpc_rb_mGrpcCore, "MetadataKeys");
+  rb_define_const(grpc_rb_mMetadataKeys, "COMPRESSION_REQUEST_ALGORITHM",
+                  rb_str_new2(GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY));
+}
+
 void Init_grpc_call() {
   /* CallError inherits from Exception to signal that it is non-recoverable */
   grpc_rb_eCallError =
@@ -900,7 +934,7 @@ void Init_grpc_call() {
                    1);
 
   /* Add ruby analogues of the Call methods. */
-  rb_define_method(grpc_rb_cCall, "run_batch", grpc_rb_call_run_batch, 4);
+  rb_define_method(grpc_rb_cCall, "run_batch", grpc_rb_call_run_batch, 1);
   rb_define_method(grpc_rb_cCall, "cancel", grpc_rb_call_cancel, 0);
   rb_define_method(grpc_rb_cCall, "close", grpc_rb_call_close, 0);
   rb_define_method(grpc_rb_cCall, "peer", grpc_rb_call_get_peer, 0);
@@ -909,6 +943,10 @@ void Init_grpc_call() {
   rb_define_method(grpc_rb_cCall, "status=", grpc_rb_call_set_status, 1);
   rb_define_method(grpc_rb_cCall, "metadata", grpc_rb_call_get_metadata, 0);
   rb_define_method(grpc_rb_cCall, "metadata=", grpc_rb_call_set_metadata, 1);
+  rb_define_method(grpc_rb_cCall, "trailing_metadata",
+                   grpc_rb_call_get_trailing_metadata, 0);
+  rb_define_method(grpc_rb_cCall, "trailing_metadata=",
+                   grpc_rb_call_set_trailing_metadata, 1);
   rb_define_method(grpc_rb_cCall, "write_flag", grpc_rb_call_get_write_flag, 0);
   rb_define_method(grpc_rb_cCall, "write_flag=", grpc_rb_call_set_write_flag,
                    1);
@@ -917,13 +955,11 @@ void Init_grpc_call() {
 
   /* Ids used to support call attributes */
   id_metadata = rb_intern("metadata");
+  id_trailing_metadata = rb_intern("trailing_metadata");
   id_status = rb_intern("status");
   id_write_flag = rb_intern("write_flag");
 
   /* Ids used by the c wrapping internals. */
-  id_cq = rb_intern("__cq");
-  id_flags = rb_intern("__flags");
-  id_input_md = rb_intern("__input_md");
   id_credentials = rb_intern("__credentials");
 
   /* Ids used in constructing the batch result. */
@@ -943,19 +979,24 @@ void Init_grpc_call() {
   Init_grpc_error_codes();
   Init_grpc_op_codes();
   Init_grpc_write_flags();
+  Init_grpc_metadata_keys();
 }
 
 /* Gets the call from the ruby object */
 grpc_call *grpc_rb_get_wrapped_call(VALUE v) {
-  grpc_call *c = NULL;
-  TypedData_Get_Struct(v, grpc_call, &grpc_call_data_type, c);
-  return c;
+  grpc_rb_call *call = NULL;
+  TypedData_Get_Struct(v, grpc_rb_call, &grpc_call_data_type, call);
+  return call->wrapped;
 }
 
 /* Obtains the wrapped object for a given call */
-VALUE grpc_rb_wrap_call(grpc_call *c) {
-  if (c == NULL) {
+VALUE grpc_rb_wrap_call(grpc_call *c, grpc_completion_queue *q) {
+  grpc_rb_call *wrapper;
+  if (c == NULL || q == NULL) {
     return Qnil;
   }
-  return TypedData_Wrap_Struct(grpc_rb_cCall, &grpc_call_data_type, c);
+  wrapper = ALLOC(grpc_rb_call);
+  wrapper->wrapped = c;
+  wrapper->queue = q;
+  return TypedData_Wrap_Struct(grpc_rb_cCall, &grpc_call_data_type, wrapper);
 }
diff --git a/src/ruby/ext/grpc/rb_call.h b/src/ruby/ext/grpc/rb_call.h
index 24adb3477b..56becdc5a4 100644
--- a/src/ruby/ext/grpc/rb_call.h
+++ b/src/ruby/ext/grpc/rb_call.h
@@ -42,7 +42,7 @@
 grpc_call* grpc_rb_get_wrapped_call(VALUE v);
 
 /* Gets the VALUE corresponding to given grpc_call. */
-VALUE grpc_rb_wrap_call(grpc_call* c);
+VALUE grpc_rb_wrap_call(grpc_call *c, grpc_completion_queue *q);
 
 /* Provides the details of an call error */
 const char* grpc_call_error_detail_of(grpc_call_error err);
diff --git a/src/ruby/ext/grpc/rb_call_credentials.c b/src/ruby/ext/grpc/rb_call_credentials.c
index 79ca5b32ce..9b6675da84 100644
--- a/src/ruby/ext/grpc/rb_call_credentials.c
+++ b/src/ruby/ext/grpc/rb_call_credentials.c
@@ -211,35 +211,6 @@ VALUE grpc_rb_wrap_call_credentials(grpc_call_credentials *c, VALUE mark) {
   return rb_wrapper;
 }
 
-/* Clones CallCredentials instances.
-   Gives CallCredentials a consistent implementation of Ruby's object copy/dup
-   protocol. */
-static VALUE grpc_rb_call_credentials_init_copy(VALUE copy, VALUE orig) {
-  grpc_rb_call_credentials *orig_cred = NULL;
-  grpc_rb_call_credentials *copy_cred = NULL;
-
-  if (copy == orig) {
-    return copy;
-  }
-
-  /* Raise an error if orig is not a credentials object or a subclass. */
-  if (TYPE(orig) != T_DATA ||
-      RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_call_credentials_free) {
-    rb_raise(rb_eTypeError, "not a %s",
-             rb_obj_classname(grpc_rb_cCallCredentials));
-  }
-
-  TypedData_Get_Struct(orig, grpc_rb_call_credentials,
-                       &grpc_rb_call_credentials_data_type, orig_cred);
-  TypedData_Get_Struct(copy, grpc_rb_call_credentials,
-                       &grpc_rb_call_credentials_data_type, copy_cred);
-
-  /* use ruby's MEMCPY to make a byte-for-byte copy of the credentials
-   * wrapper object. */
-  MEMCPY(copy_cred, orig_cred, grpc_rb_call_credentials, 1);
-  return copy;
-}
-
 /* The attribute used on the mark object to hold the callback */
 static ID id_callback;
 
@@ -308,7 +279,7 @@ void Init_grpc_call_credentials() {
   rb_define_method(grpc_rb_cCallCredentials, "initialize",
                    grpc_rb_call_credentials_init, 1);
   rb_define_method(grpc_rb_cCallCredentials, "initialize_copy",
-                   grpc_rb_call_credentials_init_copy, 1);
+                   grpc_rb_cannot_init_copy, 1);
   rb_define_method(grpc_rb_cCallCredentials, "compose",
                    grpc_rb_call_credentials_compose, -1);
 
diff --git a/src/ruby/ext/grpc/rb_channel.c b/src/ruby/ext/grpc/rb_channel.c
index 6943c93d4a..e6d30a174b 100644
--- a/src/ruby/ext/grpc/rb_channel.c
+++ b/src/ruby/ext/grpc/rb_channel.c
@@ -39,6 +39,8 @@
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
 #include "rb_grpc.h"
 #include "rb_call.h"
 #include "rb_channel_args.h"
@@ -55,11 +57,6 @@ static ID id_channel;
  * GCed before the channel */
 static ID id_target;
 
-/* id_cqueue is the name of the hidden ivar that preserves a reference to the
- * completion queue used to create the call, preserved so that it does not get
- * GCed before the channel */
-static ID id_cqueue;
-
 /* id_insecure_channel is used to indicate that a channel is insecure */
 static VALUE id_insecure_channel;
 
@@ -75,6 +72,7 @@ typedef struct grpc_rb_channel {
 
   /* The actual channel */
   grpc_channel *wrapped;
+  grpc_completion_queue *queue;
 } grpc_rb_channel;
 
 /* Destroys Channel instances. */
@@ -87,6 +85,7 @@ static void grpc_rb_channel_free(void *p) {
 
   if (ch->wrapped != NULL) {
     grpc_channel_destroy(ch->wrapped);
+    grpc_rb_completion_queue_destroy(ch->queue);
   }
 
   xfree(p);
@@ -169,6 +168,7 @@ static VALUE grpc_rb_channel_init(int argc, VALUE *argv, VALUE self) {
   }
   rb_ivar_set(self, id_target, target);
   wrapper->wrapped = ch;
+  wrapper->queue = grpc_completion_queue_create(NULL);
   return self;
 }
 
@@ -207,16 +207,18 @@ static VALUE grpc_rb_channel_get_connectivity_state(int argc, VALUE *argv,
    the completion queue with success=0 */
 static VALUE grpc_rb_channel_watch_connectivity_state(VALUE self,
                                                       VALUE last_state,
-                                                      VALUE cqueue,
-                                                      VALUE deadline,
-                                                      VALUE tag) {
+                                                      VALUE deadline) {
   grpc_rb_channel *wrapper = NULL;
   grpc_channel *ch = NULL;
   grpc_completion_queue *cq = NULL;
 
-  cq = grpc_rb_get_wrapped_completion_queue(cqueue);
+  void *tag = wrapper;
+
+  grpc_event event;
+
   TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper);
   ch = wrapper->wrapped;
+  cq = wrapper->queue;
   if (ch == NULL) {
     rb_raise(rb_eRuntimeError, "closed!");
     return Qnil;
@@ -226,45 +228,23 @@ static VALUE grpc_rb_channel_watch_connectivity_state(VALUE self,
       (grpc_connectivity_state)NUM2LONG(last_state),
       grpc_rb_time_timeval(deadline, /* absolute time */ 0),
       cq,
-      ROBJECT(tag));
-
-  return Qnil;
-}
-
-/* Clones Channel instances.
+      tag);
 
-   Gives Channel a consistent implementation of Ruby's object copy/dup
-   protocol. */
-static VALUE grpc_rb_channel_init_copy(VALUE copy, VALUE orig) {
-  grpc_rb_channel *orig_ch = NULL;
-  grpc_rb_channel *copy_ch = NULL;
+  event = rb_completion_queue_pluck(cq, tag,
+                                    gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
 
-  if (copy == orig) {
-    return copy;
-  }
-
-  /* Raise an error if orig is not a channel object or a subclass. */
-  if (TYPE(orig) != T_DATA ||
-      RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_channel_free) {
-    rb_raise(rb_eTypeError, "not a %s", rb_obj_classname(grpc_rb_cChannel));
-    return Qnil;
+  if (event.success) {
+    return Qtrue;
+  } else {
+    return Qfalse;
   }
-
-  TypedData_Get_Struct(orig, grpc_rb_channel, &grpc_channel_data_type, orig_ch);
-  TypedData_Get_Struct(copy, grpc_rb_channel, &grpc_channel_data_type, copy_ch);
-
-  /* use ruby's MEMCPY to make a byte-for-byte copy of the channel wrapper
-   * object. */
-  MEMCPY(copy_ch, orig_ch, grpc_rb_channel, 1);
-  return copy;
 }
 
 /* Create a call given a grpc_channel, in order to call method. The request
    is not sent until grpc_call_invoke is called. */
-static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue,
-                                         VALUE parent, VALUE mask,
-                                         VALUE method, VALUE host,
-                                         VALUE deadline) {
+static VALUE grpc_rb_channel_create_call(VALUE self, VALUE parent,
+                                         VALUE mask, VALUE method,
+                                         VALUE host, VALUE deadline) {
   VALUE res = Qnil;
   grpc_rb_channel *wrapper = NULL;
   grpc_call *call = NULL;
@@ -284,7 +264,7 @@ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue,
     parent_call = grpc_rb_get_wrapped_call(parent);
   }
 
-  cq = grpc_rb_get_wrapped_completion_queue(cqueue);
+  cq = grpc_completion_queue_create(NULL);
   TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper);
   ch = wrapper->wrapped;
   if (ch == NULL) {
@@ -301,15 +281,11 @@ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue,
              method_chars);
     return Qnil;
   }
-  res = grpc_rb_wrap_call(call);
+  res = grpc_rb_wrap_call(call, cq);
 
   /* Make this channel an instance attribute of the call so that it is not GCed
    * before the call. */
   rb_ivar_set(res, id_channel, self);
-
-  /* Make the completion queue an instance attribute of the call so that it is
-   * not GCed before the call. */
-  rb_ivar_set(res, id_cqueue, cqueue);
   return res;
 }
 
@@ -387,7 +363,7 @@ void Init_grpc_channel() {
   /* Provides a ruby constructor and support for dup/clone. */
   rb_define_method(grpc_rb_cChannel, "initialize", grpc_rb_channel_init, -1);
   rb_define_method(grpc_rb_cChannel, "initialize_copy",
-                   grpc_rb_channel_init_copy, 1);
+                   grpc_rb_cannot_init_copy, 1);
 
   /* Add ruby analogues of the Channel methods. */
   rb_define_method(grpc_rb_cChannel, "connectivity_state",
@@ -396,13 +372,12 @@ void Init_grpc_channel() {
   rb_define_method(grpc_rb_cChannel, "watch_connectivity_state",
                    grpc_rb_channel_watch_connectivity_state, 4);
   rb_define_method(grpc_rb_cChannel, "create_call",
-                   grpc_rb_channel_create_call, 6);
+                   grpc_rb_channel_create_call, 5);
   rb_define_method(grpc_rb_cChannel, "target", grpc_rb_channel_get_target, 0);
   rb_define_method(grpc_rb_cChannel, "destroy", grpc_rb_channel_destroy, 0);
   rb_define_alias(grpc_rb_cChannel, "close", "destroy");
 
   id_channel = rb_intern("__channel");
-  id_cqueue = rb_intern("__cqueue");
   id_target = rb_intern("__target");
   rb_define_const(grpc_rb_cChannel, "SSL_TARGET",
                   ID2SYM(rb_intern(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)));
diff --git a/src/ruby/ext/grpc/rb_channel_credentials.c b/src/ruby/ext/grpc/rb_channel_credentials.c
index cbb23885aa..5b7aa3417e 100644
--- a/src/ruby/ext/grpc/rb_channel_credentials.c
+++ b/src/ruby/ext/grpc/rb_channel_credentials.c
@@ -126,36 +126,6 @@ VALUE grpc_rb_wrap_channel_credentials(grpc_channel_credentials *c, VALUE mark)
   return rb_wrapper;
 }
 
-/* Clones ChannelCredentials instances.
-   Gives ChannelCredentials a consistent implementation of Ruby's object copy/dup
-   protocol. */
-static VALUE grpc_rb_channel_credentials_init_copy(VALUE copy, VALUE orig) {
-  grpc_rb_channel_credentials *orig_cred = NULL;
-  grpc_rb_channel_credentials *copy_cred = NULL;
-
-  if (copy == orig) {
-    return copy;
-  }
-
-  /* Raise an error if orig is not a credentials object or a subclass. */
-  if (TYPE(orig) != T_DATA ||
-      RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_channel_credentials_free) {
-    rb_raise(rb_eTypeError, "not a %s",
-             rb_obj_classname(grpc_rb_cChannelCredentials));
-  }
-
-  TypedData_Get_Struct(orig, grpc_rb_channel_credentials,
-                       &grpc_rb_channel_credentials_data_type, orig_cred);
-  TypedData_Get_Struct(copy, grpc_rb_channel_credentials,
-                       &grpc_rb_channel_credentials_data_type, copy_cred);
-
-  /* use ruby's MEMCPY to make a byte-for-byte copy of the credentials
-   * wrapper object. */
-  MEMCPY(copy_cred, orig_cred, grpc_rb_channel_credentials, 1);
-  return copy;
-}
-
-
 /* The attribute used on the mark object to hold the pem_root_certs. */
 static ID id_pem_root_certs;
 
@@ -271,7 +241,7 @@ void Init_grpc_channel_credentials() {
   rb_define_method(grpc_rb_cChannelCredentials, "initialize",
                    grpc_rb_channel_credentials_init, -1);
   rb_define_method(grpc_rb_cChannelCredentials, "initialize_copy",
-                   grpc_rb_channel_credentials_init_copy, 1);
+                   grpc_rb_cannot_init_copy, 1);
   rb_define_method(grpc_rb_cChannelCredentials, "compose",
                    grpc_rb_channel_credentials_compose, -1);
   rb_define_module_function(grpc_rb_cChannelCredentials,
diff --git a/src/ruby/ext/grpc/rb_completion_queue.c b/src/ruby/ext/grpc/rb_completion_queue.c
index 9466402db0..fd75d2f691 100644
--- a/src/ruby/ext/grpc/rb_completion_queue.c
+++ b/src/ruby/ext/grpc/rb_completion_queue.c
@@ -40,12 +40,9 @@
 
 #include <grpc/grpc.h>
 #include <grpc/support/time.h>
+#include <grpc/support/log.h>
 #include "rb_grpc.h"
 
-/* grpc_rb_cCompletionQueue is the ruby class that proxies
- * grpc_completion_queue. */
-static VALUE grpc_rb_cCompletionQueue = Qnil;
-
 /* Used to allow grpc_completion_queue_next call to release the GIL */
 typedef struct next_call_stack {
   grpc_completion_queue *cq;
@@ -55,23 +52,6 @@ typedef struct next_call_stack {
   volatile int interrupted;
 } next_call_stack;
 
-/* Calls grpc_completion_queue_next without holding the ruby GIL */
-static void *grpc_rb_completion_queue_next_no_gil(void *param) {
-  next_call_stack *const next_call = (next_call_stack*)param;
-  gpr_timespec increment = gpr_time_from_millis(20, GPR_TIMESPAN);
-  gpr_timespec deadline;
-  do {
-    deadline = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), increment);
-    next_call->event = grpc_completion_queue_next(next_call->cq,
-                                                  deadline, NULL);
-    if (next_call->event.type != GRPC_QUEUE_TIMEOUT ||
-        gpr_time_cmp(deadline, next_call->timeout) > 0) {
-      break;
-    }
-  } while (!next_call->interrupted);
-  return NULL;
-}
-
 /* Calls grpc_completion_queue_pluck without holding the ruby GIL */
 static void *grpc_rb_completion_queue_pluck_no_gil(void *param) {
   next_call_stack *const next_call = (next_call_stack*)param;
@@ -90,107 +70,32 @@ static void *grpc_rb_completion_queue_pluck_no_gil(void *param) {
   return NULL;
 }
 
-/* Shuts down and drains the completion queue if necessary.
- *
- * This is done when the ruby completion queue object is about to be GCed.
- */
-static void grpc_rb_completion_queue_shutdown_drain(grpc_completion_queue *cq) {
-  next_call_stack next_call;
-  grpc_completion_type type;
-  int drained = 0;
-  MEMZERO(&next_call, next_call_stack, 1);
-
-  grpc_completion_queue_shutdown(cq);
-  next_call.cq = cq;
-  next_call.event.type = GRPC_QUEUE_TIMEOUT;
-  /* TODO: the timeout should be a module level constant that defaults
-   * to gpr_inf_future(GPR_CLOCK_REALTIME).
-   *
-   * - at the moment this does not work, it stalls.  Using a small timeout like
-   *   this one works, and leads to fast test run times; a longer timeout was
-   *   causing unnecessary delays in the test runs.
-   *
-   * - investigate further, this is probably another example of C-level cleanup
-   * not working consistently in all cases.
-   */
-  next_call.timeout = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
-                                   gpr_time_from_micros(5e3, GPR_TIMESPAN));
-  do {
-    rb_thread_call_without_gvl(grpc_rb_completion_queue_next_no_gil,
-                               (void *)&next_call, NULL, NULL);
-    type = next_call.event.type;
-    if (type == GRPC_QUEUE_TIMEOUT) break;
-    if (type != GRPC_QUEUE_SHUTDOWN) {
-      ++drained;
-      rb_warning("completion queue shutdown: %d undrained events", drained);
-    }
-  } while (type != GRPC_QUEUE_SHUTDOWN);
-}
-
 /* Helper function to free a completion queue. */
-static void grpc_rb_completion_queue_destroy(void *p) {
-  grpc_completion_queue *cq = NULL;
-  if (p == NULL) {
-    return;
-  }
-  cq = (grpc_completion_queue *)p;
-  grpc_rb_completion_queue_shutdown_drain(cq);
+void grpc_rb_completion_queue_destroy(grpc_completion_queue *cq) {
+  /* Every function that adds an event to a queue also synchronously plucks
+     that event from the queue, and holds a reference to the Ruby object that
+     holds the queue, so we only get to this point if all of those functions
+     have completed, and the queue is empty */
+  grpc_completion_queue_shutdown(cq);
   grpc_completion_queue_destroy(cq);
 }
 
-static rb_data_type_t grpc_rb_completion_queue_data_type = {
-    "grpc_completion_queue",
-    {GRPC_RB_GC_NOT_MARKED, grpc_rb_completion_queue_destroy,
-     GRPC_RB_MEMSIZE_UNAVAILABLE, {NULL, NULL}},
-    NULL, NULL,
-#ifdef RUBY_TYPED_FREE_IMMEDIATELY
-    /* cannot immediately free because grpc_rb_completion_queue_shutdown_drain
-     * calls rb_thread_call_without_gvl. */
-    0,
-#endif
-};
-
-/* Releases the c-level resources associated with a completion queue */
-static VALUE grpc_rb_completion_queue_close(VALUE self) {
-  grpc_completion_queue* cq = grpc_rb_get_wrapped_completion_queue(self);
-  grpc_rb_completion_queue_destroy(cq);
-  RTYPEDDATA_DATA(self) = NULL;
-  return Qnil;
-}
-
-/* Allocates a completion queue. */
-static VALUE grpc_rb_completion_queue_alloc(VALUE cls) {
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
-  if (cq == NULL) {
-    rb_raise(rb_eArgError, "could not create a completion queue: not sure why");
-  }
-  return TypedData_Wrap_Struct(cls, &grpc_rb_completion_queue_data_type, cq);
-}
-
 static void unblock_func(void *param) {
   next_call_stack *const next_call = (next_call_stack*)param;
   next_call->interrupted = 1;
 }
 
-/* Blocks until the next event for given tag is available, and returns the
- * event. */
-grpc_event grpc_rb_completion_queue_pluck_event(VALUE self, VALUE tag,
-                                                VALUE timeout) {
+/* Does the same thing as grpc_completion_queue_pluck, while properly releasing
+   the GVL and handling interrupts */
+grpc_event rb_completion_queue_pluck(grpc_completion_queue *queue, void *tag,
+                                     gpr_timespec deadline, void *reserved) {
   next_call_stack next_call;
   MEMZERO(&next_call, next_call_stack, 1);
-  TypedData_Get_Struct(self, grpc_completion_queue,
-                       &grpc_rb_completion_queue_data_type, next_call.cq);
-  if (TYPE(timeout) == T_NIL) {
-    next_call.timeout = gpr_inf_future(GPR_CLOCK_REALTIME);
-  } else {
-    next_call.timeout = grpc_rb_time_timeval(timeout, /* absolute time*/ 0);
-  }
-  if (TYPE(tag) == T_NIL) {
-    next_call.tag = NULL;
-  } else {
-    next_call.tag = ROBJECT(tag);
-  }
+  next_call.cq = queue;
+  next_call.timeout = deadline;
+  next_call.tag = tag;
   next_call.event.type = GRPC_QUEUE_TIMEOUT;
+  (void)reserved;
   /* Loop until we finish a pluck without an interruption. The internal
      pluck function runs either until it is interrupted or it gets an
      event, or time runs out.
@@ -210,27 +115,3 @@ grpc_event grpc_rb_completion_queue_pluck_event(VALUE self, VALUE tag,
            next_call.event.type == GRPC_QUEUE_TIMEOUT);
   return next_call.event;
 }
-
-void Init_grpc_completion_queue() {
-  grpc_rb_cCompletionQueue =
-      rb_define_class_under(grpc_rb_mGrpcCore, "CompletionQueue", rb_cObject);
-
-  /* constructor: uses an alloc func without an initializer. Using a simple
-     alloc func works here as the grpc header does not specify any args for
-     this func, so no separate initialization step is necessary. */
-  rb_define_alloc_func(grpc_rb_cCompletionQueue,
-                       grpc_rb_completion_queue_alloc);
-
-  /* close: Provides a way to close the underlying file descriptor without
-     waiting for ruby garbage collection. */
-  rb_define_method(grpc_rb_cCompletionQueue, "close",
-                   grpc_rb_completion_queue_close, 0);
-}
-
-/* Gets the wrapped completion queue from the ruby wrapper */
-grpc_completion_queue *grpc_rb_get_wrapped_completion_queue(VALUE v) {
-  grpc_completion_queue *cq = NULL;
-  TypedData_Get_Struct(v, grpc_completion_queue,
-                       &grpc_rb_completion_queue_data_type, cq);
-  return cq;
-}
diff --git a/src/ruby/ext/grpc/rb_completion_queue.h b/src/ruby/ext/grpc/rb_completion_queue.h
index 42de43c3fb..aa9dc6416a 100644
--- a/src/ruby/ext/grpc/rb_completion_queue.h
+++ b/src/ruby/ext/grpc/rb_completion_queue.h
@@ -38,18 +38,14 @@
 
 #include <grpc/grpc.h>
 
-/* Gets the wrapped completion queue from the ruby wrapper */
-grpc_completion_queue *grpc_rb_get_wrapped_completion_queue(VALUE v);
+void grpc_rb_completion_queue_destroy(grpc_completion_queue *cq);
 
 /**
  * Makes the implementation of CompletionQueue#pluck available in other files
  *
  * This avoids having code that holds the GIL repeated at multiple sites.
  */
-grpc_event grpc_rb_completion_queue_pluck_event(VALUE self, VALUE tag,
-                                                VALUE timeout);
-
-/* Initializes the CompletionQueue class. */
-void Init_grpc_completion_queue();
+grpc_event rb_completion_queue_pluck(grpc_completion_queue *queue, void *tag,
+                                     gpr_timespec deadline, void *reserved);
 
 #endif /* GRPC_RB_COMPLETION_QUEUE_H_ */
diff --git a/src/ruby/ext/grpc/rb_compression_options.c b/src/ruby/ext/grpc/rb_compression_options.c
new file mode 100644
index 0000000000..0a3a215b1c
--- /dev/null
+++ b/src/ruby/ext/grpc/rb_compression_options.c
@@ -0,0 +1,464 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <ruby/ruby.h>
+
+#include "rb_compression_options.h"
+#include "rb_grpc_imports.generated.h"
+
+#include <grpc/compression.h>
+#include <grpc/grpc.h>
+#include <grpc/impl/codegen/alloc.h>
+#include <grpc/impl/codegen/compression_types.h>
+#include <grpc/impl/codegen/grpc_types.h>
+#include <string.h>
+
+#include "rb_grpc.h"
+
+static VALUE grpc_rb_cCompressionOptions = Qnil;
+
+/* Ruby Ids for the names of valid compression levels. */
+static VALUE id_compress_level_none = Qnil;
+static VALUE id_compress_level_low = Qnil;
+static VALUE id_compress_level_medium = Qnil;
+static VALUE id_compress_level_high = Qnil;
+
+/* grpc_rb_compression_options wraps a grpc_compression_options.
+ * It can be used to get the channel argument key-values for specific
+ * compression settings. */
+
+/* Note that ruby objects of this type don't carry any state in other
+ * Ruby objects and don't have a mark for GC. */
+typedef struct grpc_rb_compression_options {
+  /* The actual compression options that's being wrapped */
+  grpc_compression_options *wrapped;
+} grpc_rb_compression_options;
+
+/* Destroys the compression options instances and free the
+ * wrapped grpc compression options. */
+static void grpc_rb_compression_options_free(void *p) {
+  grpc_rb_compression_options *wrapper = NULL;
+  if (p == NULL) {
+    return;
+  };
+  wrapper = (grpc_rb_compression_options *)p;
+
+  if (wrapper->wrapped != NULL) {
+    gpr_free(wrapper->wrapped);
+    wrapper->wrapped = NULL;
+  }
+
+  xfree(p);
+}
+
+/* Ruby recognized data type for the CompressionOptions class. */
+static rb_data_type_t grpc_rb_compression_options_data_type = {
+    "grpc_compression_options",
+    {NULL,
+     grpc_rb_compression_options_free,
+     GRPC_RB_MEMSIZE_UNAVAILABLE,
+     {NULL, NULL}},
+    NULL,
+    NULL,
+#ifdef RUBY_TYPED_FREE_IMMEDIATELY
+    RUBY_TYPED_FREE_IMMEDIATELY
+#endif
+};
+
+/* Allocates CompressionOptions instances.
+   Allocate the wrapped grpc compression options and
+   initialize it here too. */
+static VALUE grpc_rb_compression_options_alloc(VALUE cls) {
+  grpc_rb_compression_options *wrapper =
+      gpr_malloc(sizeof(grpc_rb_compression_options));
+  wrapper->wrapped = NULL;
+  wrapper->wrapped = gpr_malloc(sizeof(grpc_compression_options));
+  grpc_compression_options_init(wrapper->wrapped);
+
+  return TypedData_Wrap_Struct(cls, &grpc_rb_compression_options_data_type,
+                               wrapper);
+}
+
+/* Disables a compression algorithm, given the GRPC core internal number of a
+ * compression algorithm. */
+VALUE grpc_rb_compression_options_disable_compression_algorithm_internal(
+    VALUE self, VALUE algorithm_to_disable) {
+  grpc_compression_algorithm compression_algorithm = 0;
+  grpc_rb_compression_options *wrapper = NULL;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+  compression_algorithm =
+      (grpc_compression_algorithm)NUM2INT(algorithm_to_disable);
+
+  grpc_compression_options_disable_algorithm(wrapper->wrapped,
+                                             compression_algorithm);
+
+  return Qnil;
+}
+
+/* Gets the compression internal enum value of a compression level given its
+ * name. */
+grpc_compression_level grpc_rb_compression_options_level_name_to_value_internal(
+    VALUE level_name) {
+  Check_Type(level_name, T_SYMBOL);
+
+  /* Check the compression level of the name passed in, and see which macro
+   * from the GRPC core header files match. */
+  if (id_compress_level_none == SYM2ID(level_name)) {
+    return GRPC_COMPRESS_LEVEL_NONE;
+  } else if (id_compress_level_low == SYM2ID(level_name)) {
+    return GRPC_COMPRESS_LEVEL_LOW;
+  } else if (id_compress_level_medium == SYM2ID(level_name)) {
+    return GRPC_COMPRESS_LEVEL_MED;
+  } else if (id_compress_level_high == SYM2ID(level_name)) {
+    return GRPC_COMPRESS_LEVEL_HIGH;
+  }
+
+  rb_raise(rb_eArgError,
+           "Unrecognized compression level name."
+           "Valid compression level names are none, low, medium, and high.");
+
+  /* Dummy return statement. */
+  return GRPC_COMPRESS_LEVEL_NONE;
+}
+
+/* Sets the default compression level, given the name of a compression level.
+ * Throws an error if no algorithm matched. */
+void grpc_rb_compression_options_set_default_level(
+    grpc_compression_options *options, VALUE new_level_name) {
+  options->default_level.level =
+      grpc_rb_compression_options_level_name_to_value_internal(new_level_name);
+  options->default_level.is_set = 1;
+}
+
+/* Gets the internal value of a compression algorithm suitable as the value
+ * in a GRPC core channel arguments hash.
+ * algorithm_value is an out parameter.
+ * Raises an error if the name of the algorithm passed in is invalid. */
+void grpc_rb_compression_options_algorithm_name_to_value_internal(
+    grpc_compression_algorithm *algorithm_value, VALUE algorithm_name) {
+  char *name_str = NULL;
+  long name_len = 0;
+  VALUE algorithm_name_as_string = Qnil;
+
+  Check_Type(algorithm_name, T_SYMBOL);
+
+  /* Convert the algorithm symbol to a ruby string, so that we can get the
+   * correct C string out of it. */
+  algorithm_name_as_string = rb_funcall(algorithm_name, rb_intern("to_s"), 0);
+
+  name_str = RSTRING_PTR(algorithm_name_as_string);
+  name_len = RSTRING_LEN(algorithm_name_as_string);
+
+  /* Raise an error if the name isn't recognized as a compression algorithm by
+   * the algorithm parse function
+   * in GRPC core. */
+  if (!grpc_compression_algorithm_parse(name_str, name_len, algorithm_value)) {
+    rb_raise(rb_eNameError, "Invalid compression algorithm name: %s",
+             StringValueCStr(algorithm_name_as_string));
+  }
+}
+
+/* Indicates whether a given algorithm is enabled on this instance, given the
+ * readable algorithm name. */
+VALUE grpc_rb_compression_options_is_algorithm_enabled(VALUE self,
+                                                       VALUE algorithm_name) {
+  grpc_rb_compression_options *wrapper = NULL;
+  grpc_compression_algorithm internal_algorithm_value;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+  grpc_rb_compression_options_algorithm_name_to_value_internal(
+      &internal_algorithm_value, algorithm_name);
+
+  if (grpc_compression_options_is_algorithm_enabled(wrapper->wrapped,
+                                                    internal_algorithm_value)) {
+    return Qtrue;
+  }
+  return Qfalse;
+}
+
+/* Sets the default algorithm to the name of the algorithm passed in.
+ * Raises an error if the name is not a valid compression algorithm name. */
+void grpc_rb_compression_options_set_default_algorithm(
+    grpc_compression_options *options, VALUE algorithm_name) {
+  grpc_rb_compression_options_algorithm_name_to_value_internal(
+      &options->default_algorithm.algorithm, algorithm_name);
+  options->default_algorithm.is_set = 1;
+}
+
+/* Disables an algorithm on the current instance, given the name of an
+ * algorithm.
+ * Fails if the algorithm name is invalid. */
+void grpc_rb_compression_options_disable_algorithm(
+    grpc_compression_options *compression_options, VALUE algorithm_name) {
+  grpc_compression_algorithm internal_algorithm_value;
+
+  grpc_rb_compression_options_algorithm_name_to_value_internal(
+      &internal_algorithm_value, algorithm_name);
+  grpc_compression_options_disable_algorithm(compression_options,
+                                             internal_algorithm_value);
+}
+
+/* Provides a ruby hash of GRPC core channel argument key-values that
+ * correspond to the compression settings on this instance. */
+VALUE grpc_rb_compression_options_to_hash(VALUE self) {
+  grpc_rb_compression_options *wrapper = NULL;
+  grpc_compression_options *compression_options = NULL;
+  VALUE channel_arg_hash = rb_hash_new();
+  VALUE key = Qnil;
+  VALUE value = Qnil;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+  compression_options = wrapper->wrapped;
+
+  /* Add key-value pairs to the new Ruby hash. It can be used
+   * as GRPC core channel arguments. */
+  if (compression_options->default_level.is_set) {
+    key = rb_str_new2(GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL);
+    value = INT2NUM((int)compression_options->default_level.level);
+    rb_hash_aset(channel_arg_hash, key, value);
+  }
+
+  if (compression_options->default_algorithm.is_set) {
+    key = rb_str_new2(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM);
+    value = INT2NUM((int)compression_options->default_algorithm.algorithm);
+    rb_hash_aset(channel_arg_hash, key, value);
+  }
+
+  key = rb_str_new2(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET);
+  value = INT2NUM((int)compression_options->enabled_algorithms_bitset);
+  rb_hash_aset(channel_arg_hash, key, value);
+
+  return channel_arg_hash;
+}
+
+/* Converts an internal enum level value to a readable level name.
+ * Fails if the level value is invalid. */
+VALUE grpc_rb_compression_options_level_value_to_name_internal(
+    grpc_compression_level compression_value) {
+  switch (compression_value) {
+    case GRPC_COMPRESS_LEVEL_NONE:
+      return ID2SYM(id_compress_level_none);
+    case GRPC_COMPRESS_LEVEL_LOW:
+      return ID2SYM(id_compress_level_low);
+    case GRPC_COMPRESS_LEVEL_MED:
+      return ID2SYM(id_compress_level_medium);
+    case GRPC_COMPRESS_LEVEL_HIGH:
+      return ID2SYM(id_compress_level_high);
+    default:
+      rb_raise(
+          rb_eArgError,
+          "Failed to convert compression level value to name for value: %d",
+          (int)compression_value);
+  }
+}
+
+/* Converts an algorithm internal enum value to a readable name.
+ * Fails if the enum value is invalid. */
+VALUE grpc_rb_compression_options_algorithm_value_to_name_internal(
+    grpc_compression_algorithm internal_value) {
+  char *algorithm_name = NULL;
+
+  if (!grpc_compression_algorithm_name(internal_value, &algorithm_name)) {
+    rb_raise(rb_eArgError, "Failed to convert algorithm value to name");
+  }
+
+  return ID2SYM(rb_intern(algorithm_name));
+}
+
+/* Gets the readable name of the default algorithm if one has been set.
+ * Returns nil if no algorithm has been set. */
+VALUE grpc_rb_compression_options_get_default_algorithm(VALUE self) {
+  grpc_compression_algorithm internal_value;
+  grpc_rb_compression_options *wrapper = NULL;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+
+  if (wrapper->wrapped->default_algorithm.is_set) {
+    internal_value = wrapper->wrapped->default_algorithm.algorithm;
+    return grpc_rb_compression_options_algorithm_value_to_name_internal(
+        internal_value);
+  }
+
+  return Qnil;
+}
+
+/* Gets the internal value of the default compression level that is to be passed
+ * to the GRPC core as a channel argument value.
+ * A nil return value means that it hasn't been set. */
+VALUE grpc_rb_compression_options_get_default_level(VALUE self) {
+  grpc_compression_level internal_value;
+  grpc_rb_compression_options *wrapper = NULL;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+
+  if (wrapper->wrapped->default_level.is_set) {
+    internal_value = wrapper->wrapped->default_level.level;
+    return grpc_rb_compression_options_level_value_to_name_internal(
+        internal_value);
+  }
+
+  return Qnil;
+}
+
+/* Gets a list of the disabled algorithms as readable names.
+ * Returns an empty list if no algorithms have been disabled. */
+VALUE grpc_rb_compression_options_get_disabled_algorithms(VALUE self) {
+  VALUE disabled_algorithms = rb_ary_new();
+  grpc_compression_algorithm internal_value;
+  grpc_rb_compression_options *wrapper = NULL;
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+
+  for (internal_value = GRPC_COMPRESS_NONE;
+       internal_value < GRPC_COMPRESS_ALGORITHMS_COUNT; internal_value++) {
+    if (!grpc_compression_options_is_algorithm_enabled(wrapper->wrapped,
+                                                       internal_value)) {
+      rb_ary_push(disabled_algorithms,
+                  grpc_rb_compression_options_algorithm_value_to_name_internal(
+                      internal_value));
+    }
+  }
+  return disabled_algorithms;
+}
+
+/* Initializes the compression options wrapper.
+ * Takes an optional hash parameter.
+ *
+ * Example call-seq:
+ *   options = CompressionOptions.new(
+ *     default_level: :none,
+ *     disabled_algorithms: [:gzip]
+ *   )
+ *   channel_arg hash = Hash.new[...]
+ *   channel_arg_hash_with_compression_options = channel_arg_hash.merge(options)
+ */
+VALUE grpc_rb_compression_options_init(int argc, VALUE *argv, VALUE self) {
+  grpc_rb_compression_options *wrapper = NULL;
+  VALUE default_algorithm = Qnil;
+  VALUE default_level = Qnil;
+  VALUE disabled_algorithms = Qnil;
+  VALUE algorithm_name = Qnil;
+  VALUE hash_arg = Qnil;
+
+  rb_scan_args(argc, argv, "01", &hash_arg);
+
+  /* Check if the hash parameter was passed, or if invalid arguments were
+   * passed. */
+  if (hash_arg == Qnil) {
+    return self;
+  } else if (TYPE(hash_arg) != T_HASH || argc > 1) {
+    rb_raise(rb_eArgError,
+             "Invalid arguments. Expecting optional hash parameter");
+  }
+
+  TypedData_Get_Struct(self, grpc_rb_compression_options,
+                       &grpc_rb_compression_options_data_type, wrapper);
+
+  /* Set the default algorithm if one was chosen. */
+  default_algorithm =
+      rb_hash_aref(hash_arg, ID2SYM(rb_intern("default_algorithm")));
+  if (default_algorithm != Qnil) {
+    grpc_rb_compression_options_set_default_algorithm(wrapper->wrapped,
+                                                      default_algorithm);
+  }
+
+  /* Set the default level if one was chosen. */
+  default_level = rb_hash_aref(hash_arg, ID2SYM(rb_intern("default_level")));
+  if (default_level != Qnil) {
+    grpc_rb_compression_options_set_default_level(wrapper->wrapped,
+                                                  default_level);
+  }
+
+  /* Set the disabled algorithms if any were chosen. */
+  disabled_algorithms =
+      rb_hash_aref(hash_arg, ID2SYM(rb_intern("disabled_algorithms")));
+  if (disabled_algorithms != Qnil) {
+    Check_Type(disabled_algorithms, T_ARRAY);
+
+    for (int i = 0; i < RARRAY_LEN(disabled_algorithms); i++) {
+      algorithm_name = rb_ary_entry(disabled_algorithms, i);
+      grpc_rb_compression_options_disable_algorithm(wrapper->wrapped,
+                                                    algorithm_name);
+    }
+  }
+
+  return self;
+}
+
+void Init_grpc_compression_options() {
+  grpc_rb_cCompressionOptions = rb_define_class_under(
+      grpc_rb_mGrpcCore, "CompressionOptions", rb_cObject);
+
+  /* Allocates an object managed by the ruby runtime. */
+  rb_define_alloc_func(grpc_rb_cCompressionOptions,
+                       grpc_rb_compression_options_alloc);
+
+  /* Initializes the ruby wrapper. #new method takes an optional hash argument.
+   */
+  rb_define_method(grpc_rb_cCompressionOptions, "initialize",
+                   grpc_rb_compression_options_init, -1);
+
+  /* Methods for getting the default algorithm, default level, and disabled
+   * algorithms as readable names. */
+  rb_define_method(grpc_rb_cCompressionOptions, "default_algorithm",
+                   grpc_rb_compression_options_get_default_algorithm, 0);
+  rb_define_method(grpc_rb_cCompressionOptions, "default_level",
+                   grpc_rb_compression_options_get_default_level, 0);
+  rb_define_method(grpc_rb_cCompressionOptions, "disabled_algorithms",
+                   grpc_rb_compression_options_get_disabled_algorithms, 0);
+
+  /* Determines whether or not an algorithm is enabled, given a readable
+   * algorithm name.*/
+  rb_define_method(grpc_rb_cCompressionOptions, "algorithm_enabled?",
+                   grpc_rb_compression_options_is_algorithm_enabled, 1);
+
+  /* Provides a hash of the compression settings suitable
+   * for passing to server or channel args. */
+  rb_define_method(grpc_rb_cCompressionOptions, "to_hash",
+                   grpc_rb_compression_options_to_hash, 0);
+  rb_define_alias(grpc_rb_cCompressionOptions, "to_channel_arg_hash",
+                  "to_hash");
+
+  /* Ruby ids for the names of the different compression levels. */
+  id_compress_level_none = rb_intern("none");
+  id_compress_level_low = rb_intern("low");
+  id_compress_level_medium = rb_intern("medium");
+  id_compress_level_high = rb_intern("high");
+}
diff --git a/src/ruby/ext/grpc/rb_compression_options.h b/src/ruby/ext/grpc/rb_compression_options.h
new file mode 100644
index 0000000000..4d5a924786
--- /dev/null
+++ b/src/ruby/ext/grpc/rb_compression_options.h
@@ -0,0 +1,44 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_RB_COMPRESSION_OPTIONS_H_
+#define GRPC_RB_COMPRESSION_OPTIONS_H_
+
+#include <ruby/ruby.h>
+
+#include <grpc/grpc.h>
+
+/* Initializes the compression options ruby wrapper. */
+void Init_grpc_compression_options();
+
+#endif /* GRPC_RB_COMPRESSION_OPTIONS_H_ */
diff --git a/src/ruby/ext/grpc/rb_grpc.c b/src/ruby/ext/grpc/rb_grpc.c
index 9246893f9f..17cd165a91 100644
--- a/src/ruby/ext/grpc/rb_grpc.c
+++ b/src/ruby/ext/grpc/rb_grpc.c
@@ -46,10 +46,10 @@
 #include "rb_call_credentials.h"
 #include "rb_channel.h"
 #include "rb_channel_credentials.h"
-#include "rb_completion_queue.h"
 #include "rb_loader.h"
 #include "rb_server.h"
 #include "rb_server_credentials.h"
+#include "rb_compression_options.h"
 
 static VALUE grpc_rb_cTimeVal = Qnil;
 
@@ -85,7 +85,7 @@ VALUE grpc_rb_cannot_init(VALUE self) {
 VALUE grpc_rb_cannot_init_copy(VALUE copy, VALUE self) {
   (void)self;
   rb_raise(rb_eTypeError,
-           "initialization of %s only allowed from the gRPC native layer",
+           "Copy initialization of %s is not supported",
            rb_obj_classname(copy));
   return Qnil;
 }
@@ -221,7 +221,7 @@ static VALUE grpc_rb_time_val_to_time(VALUE self) {
                        time_const);
   real_time = gpr_convert_clock_type(*time_const, GPR_CLOCK_REALTIME);
   return rb_funcall(rb_cTime, id_at, 2, INT2NUM(real_time.tv_sec),
-                    INT2NUM(real_time.tv_nsec));
+                    INT2NUM(real_time.tv_nsec / 1000));
 }
 
 /* Invokes inspect on the ctime version of the time val. */
@@ -318,7 +318,7 @@ void Init_grpc_c() {
   grpc_rb_mGrpcCore = rb_define_module_under(grpc_rb_mGRPC, "Core");
   grpc_rb_sNewServerRpc =
       rb_struct_define("NewServerRpc", "method", "host",
-                       "deadline", "metadata", "call", "cq", NULL);
+                       "deadline", "metadata", "call", NULL);
   grpc_rb_sStatus =
       rb_struct_define("Status", "code", "details", "metadata", NULL);
   sym_code = ID2SYM(rb_intern("code"));
@@ -326,7 +326,6 @@ void Init_grpc_c() {
   sym_metadata = ID2SYM(rb_intern("metadata"));
 
   Init_grpc_channel();
-  Init_grpc_completion_queue();
   Init_grpc_call();
   Init_grpc_call_credentials();
   Init_grpc_channel_credentials();
@@ -334,4 +333,5 @@ void Init_grpc_c() {
   Init_grpc_server_credentials();
   Init_grpc_status_codes();
   Init_grpc_time_consts();
+  Init_grpc_compression_options();
 }
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c
index 4d9707638f..aee57b11aa 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c
@@ -61,15 +61,10 @@ census_trace_print_type census_trace_print_import;
 census_trace_scan_start_type census_trace_scan_start_import;
 census_get_trace_record_type census_get_trace_record_import;
 census_trace_scan_end_type census_trace_scan_end_import;
+census_define_resource_type census_define_resource_import;
+census_delete_resource_type census_delete_resource_import;
+census_resource_id_type census_resource_id_import;
 census_record_values_type census_record_values_import;
-census_view_create_type census_view_create_import;
-census_view_delete_type census_view_delete_import;
-census_view_metric_type census_view_metric_import;
-census_view_naggregations_type census_view_naggregations_import;
-census_view_tags_type census_view_tags_import;
-census_view_aggregrations_type census_view_aggregrations_import;
-census_view_get_data_type census_view_get_data_import;
-census_view_reset_type census_view_reset_import;
 grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_import;
 grpc_compression_algorithm_name_type grpc_compression_algorithm_name_import;
 grpc_compression_algorithm_for_level_type grpc_compression_algorithm_for_level_import;
@@ -128,6 +123,7 @@ grpc_is_binary_header_type grpc_is_binary_header_import;
 grpc_call_error_to_string_type grpc_call_error_to_string_import;
 grpc_insecure_channel_create_from_fd_type grpc_insecure_channel_create_from_fd_import;
 grpc_server_add_insecure_channel_from_fd_type grpc_server_add_insecure_channel_from_fd_import;
+grpc_use_signal_type grpc_use_signal_import;
 grpc_auth_property_iterator_next_type grpc_auth_property_iterator_next_import;
 grpc_auth_context_property_iterator_type grpc_auth_context_property_iterator_import;
 grpc_auth_context_peer_identity_type grpc_auth_context_peer_identity_import;
@@ -185,6 +181,7 @@ gpr_set_log_function_type gpr_set_log_function_import;
 gpr_slice_ref_type gpr_slice_ref_import;
 gpr_slice_unref_type gpr_slice_unref_import;
 gpr_slice_new_type gpr_slice_new_import;
+gpr_slice_new_with_user_data_type gpr_slice_new_with_user_data_import;
 gpr_slice_new_with_len_type gpr_slice_new_with_len_import;
 gpr_slice_malloc_type gpr_slice_malloc_import;
 gpr_slice_from_copied_string_type gpr_slice_from_copied_string_import;
@@ -332,15 +329,10 @@ void grpc_rb_load_imports(HMODULE library) {
   census_trace_scan_start_import = (census_trace_scan_start_type) GetProcAddress(library, "census_trace_scan_start");
   census_get_trace_record_import = (census_get_trace_record_type) GetProcAddress(library, "census_get_trace_record");
   census_trace_scan_end_import = (census_trace_scan_end_type) GetProcAddress(library, "census_trace_scan_end");
+  census_define_resource_import = (census_define_resource_type) GetProcAddress(library, "census_define_resource");
+  census_delete_resource_import = (census_delete_resource_type) GetProcAddress(library, "census_delete_resource");
+  census_resource_id_import = (census_resource_id_type) GetProcAddress(library, "census_resource_id");
   census_record_values_import = (census_record_values_type) GetProcAddress(library, "census_record_values");
-  census_view_create_import = (census_view_create_type) GetProcAddress(library, "census_view_create");
-  census_view_delete_import = (census_view_delete_type) GetProcAddress(library, "census_view_delete");
-  census_view_metric_import = (census_view_metric_type) GetProcAddress(library, "census_view_metric");
-  census_view_naggregations_import = (census_view_naggregations_type) GetProcAddress(library, "census_view_naggregations");
-  census_view_tags_import = (census_view_tags_type) GetProcAddress(library, "census_view_tags");
-  census_view_aggregrations_import = (census_view_aggregrations_type) GetProcAddress(library, "census_view_aggregrations");
-  census_view_get_data_import = (census_view_get_data_type) GetProcAddress(library, "census_view_get_data");
-  census_view_reset_import = (census_view_reset_type) GetProcAddress(library, "census_view_reset");
   grpc_compression_algorithm_parse_import = (grpc_compression_algorithm_parse_type) GetProcAddress(library, "grpc_compression_algorithm_parse");
   grpc_compression_algorithm_name_import = (grpc_compression_algorithm_name_type) GetProcAddress(library, "grpc_compression_algorithm_name");
   grpc_compression_algorithm_for_level_import = (grpc_compression_algorithm_for_level_type) GetProcAddress(library, "grpc_compression_algorithm_for_level");
@@ -399,6 +391,7 @@ void grpc_rb_load_imports(HMODULE library) {
   grpc_call_error_to_string_import = (grpc_call_error_to_string_type) GetProcAddress(library, "grpc_call_error_to_string");
   grpc_insecure_channel_create_from_fd_import = (grpc_insecure_channel_create_from_fd_type) GetProcAddress(library, "grpc_insecure_channel_create_from_fd");
   grpc_server_add_insecure_channel_from_fd_import = (grpc_server_add_insecure_channel_from_fd_type) GetProcAddress(library, "grpc_server_add_insecure_channel_from_fd");
+  grpc_use_signal_import = (grpc_use_signal_type) GetProcAddress(library, "grpc_use_signal");
   grpc_auth_property_iterator_next_import = (grpc_auth_property_iterator_next_type) GetProcAddress(library, "grpc_auth_property_iterator_next");
   grpc_auth_context_property_iterator_import = (grpc_auth_context_property_iterator_type) GetProcAddress(library, "grpc_auth_context_property_iterator");
   grpc_auth_context_peer_identity_import = (grpc_auth_context_peer_identity_type) GetProcAddress(library, "grpc_auth_context_peer_identity");
@@ -456,6 +449,7 @@ void grpc_rb_load_imports(HMODULE library) {
   gpr_slice_ref_import = (gpr_slice_ref_type) GetProcAddress(library, "gpr_slice_ref");
   gpr_slice_unref_import = (gpr_slice_unref_type) GetProcAddress(library, "gpr_slice_unref");
   gpr_slice_new_import = (gpr_slice_new_type) GetProcAddress(library, "gpr_slice_new");
+  gpr_slice_new_with_user_data_import = (gpr_slice_new_with_user_data_type) GetProcAddress(library, "gpr_slice_new_with_user_data");
   gpr_slice_new_with_len_import = (gpr_slice_new_with_len_type) GetProcAddress(library, "gpr_slice_new_with_len");
   gpr_slice_malloc_import = (gpr_slice_malloc_type) GetProcAddress(library, "gpr_slice_malloc");
   gpr_slice_from_copied_string_import = (gpr_slice_from_copied_string_type) GetProcAddress(library, "gpr_slice_from_copied_string");
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
index 072e29c232..3bb76fbb97 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
@@ -134,33 +134,18 @@ extern census_get_trace_record_type census_get_trace_record_import;
 typedef void(*census_trace_scan_end_type)();
 extern census_trace_scan_end_type census_trace_scan_end_import;
 #define census_trace_scan_end census_trace_scan_end_import
+typedef int32_t(*census_define_resource_type)(const uint8_t *resource_pb, size_t resource_pb_size);
+extern census_define_resource_type census_define_resource_import;
+#define census_define_resource census_define_resource_import
+typedef void(*census_delete_resource_type)(int32_t resource_id);
+extern census_delete_resource_type census_delete_resource_import;
+#define census_delete_resource census_delete_resource_import
+typedef int32_t(*census_resource_id_type)(const char *name);
+extern census_resource_id_type census_resource_id_import;
+#define census_resource_id census_resource_id_import
 typedef void(*census_record_values_type)(census_context *context, census_value *values, size_t nvalues);
 extern census_record_values_type census_record_values_import;
 #define census_record_values census_record_values_import
-typedef census_view *(*census_view_create_type)(uint32_t metric_id, const census_context *tags, const census_aggregation *aggregations, size_t naggregations);
-extern census_view_create_type census_view_create_import;
-#define census_view_create census_view_create_import
-typedef void(*census_view_delete_type)(census_view *view);
-extern census_view_delete_type census_view_delete_import;
-#define census_view_delete census_view_delete_import
-typedef size_t(*census_view_metric_type)(const census_view *view);
-extern census_view_metric_type census_view_metric_import;
-#define census_view_metric census_view_metric_import
-typedef size_t(*census_view_naggregations_type)(const census_view *view);
-extern census_view_naggregations_type census_view_naggregations_import;
-#define census_view_naggregations census_view_naggregations_import
-typedef const census_context *(*census_view_tags_type)(const census_view *view);
-extern census_view_tags_type census_view_tags_import;
-#define census_view_tags census_view_tags_import
-typedef const census_aggregation *(*census_view_aggregrations_type)(const census_view *view);
-extern census_view_aggregrations_type census_view_aggregrations_import;
-#define census_view_aggregrations census_view_aggregrations_import
-typedef const census_view_data *(*census_view_get_data_type)(const census_view *view);
-extern census_view_get_data_type census_view_get_data_import;
-#define census_view_get_data census_view_get_data_import
-typedef void(*census_view_reset_type)(census_view *view);
-extern census_view_reset_type census_view_reset_import;
-#define census_view_reset census_view_reset_import
 typedef int(*grpc_compression_algorithm_parse_type)(const char *name, size_t name_length, grpc_compression_algorithm *algorithm);
 extern grpc_compression_algorithm_parse_type grpc_compression_algorithm_parse_import;
 #define grpc_compression_algorithm_parse grpc_compression_algorithm_parse_import
@@ -335,6 +320,9 @@ extern grpc_insecure_channel_create_from_fd_type grpc_insecure_channel_create_fr
 typedef void(*grpc_server_add_insecure_channel_from_fd_type)(grpc_server *server, grpc_completion_queue *cq, int fd);
 extern grpc_server_add_insecure_channel_from_fd_type grpc_server_add_insecure_channel_from_fd_import;
 #define grpc_server_add_insecure_channel_from_fd grpc_server_add_insecure_channel_from_fd_import
+typedef void(*grpc_use_signal_type)(int signum);
+extern grpc_use_signal_type grpc_use_signal_import;
+#define grpc_use_signal grpc_use_signal_import
 typedef const grpc_auth_property *(*grpc_auth_property_iterator_next_type)(grpc_auth_property_iterator *it);
 extern grpc_auth_property_iterator_next_type grpc_auth_property_iterator_next_import;
 #define grpc_auth_property_iterator_next grpc_auth_property_iterator_next_import
@@ -467,7 +455,7 @@ extern grpc_byte_buffer_length_type grpc_byte_buffer_length_import;
 typedef void(*grpc_byte_buffer_destroy_type)(grpc_byte_buffer *byte_buffer);
 extern grpc_byte_buffer_destroy_type grpc_byte_buffer_destroy_import;
 #define grpc_byte_buffer_destroy grpc_byte_buffer_destroy_import
-typedef void(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
+typedef int(*grpc_byte_buffer_reader_init_type)(grpc_byte_buffer_reader *reader, grpc_byte_buffer *buffer);
 extern grpc_byte_buffer_reader_init_type grpc_byte_buffer_reader_init_import;
 #define grpc_byte_buffer_reader_init grpc_byte_buffer_reader_init_import
 typedef void(*grpc_byte_buffer_reader_destroy_type)(grpc_byte_buffer_reader *reader);
@@ -506,6 +494,9 @@ extern gpr_slice_unref_type gpr_slice_unref_import;
 typedef gpr_slice(*gpr_slice_new_type)(void *p, size_t len, void (*destroy)(void *));
 extern gpr_slice_new_type gpr_slice_new_import;
 #define gpr_slice_new gpr_slice_new_import
+typedef gpr_slice(*gpr_slice_new_with_user_data_type)(void *p, size_t len, void (*destroy)(void *), void *user_data);
+extern gpr_slice_new_with_user_data_type gpr_slice_new_with_user_data_import;
+#define gpr_slice_new_with_user_data gpr_slice_new_with_user_data_import
 typedef gpr_slice(*gpr_slice_new_with_len_type)(void *p, size_t len, void (*destroy)(void *, size_t));
 extern gpr_slice_new_with_len_type gpr_slice_new_with_len_import;
 #define gpr_slice_new_with_len gpr_slice_new_with_len_import
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index f108b8acfc..2a6a246e67 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -38,6 +38,7 @@
 
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
+#include <grpc/support/log.h>
 #include "rb_call.h"
 #include "rb_channel_args.h"
 #include "rb_completion_queue.h"
@@ -53,53 +54,51 @@ static ID id_at;
 /* id_insecure_server is used to indicate that a server is insecure */
 static VALUE id_insecure_server;
 
-/* grpc_rb_server wraps a grpc_server.  It provides a peer ruby object,
-  'mark' to minimize copying when a server is created from ruby. */
+/* grpc_rb_server wraps a grpc_server. */
 typedef struct grpc_rb_server {
-  /* Holder of ruby objects involved in constructing the server */
-  VALUE mark;
   /* The actual server */
   grpc_server *wrapped;
   grpc_completion_queue *queue;
 } grpc_rb_server;
 
+static void destroy_server(grpc_rb_server *server, gpr_timespec deadline) {
+  grpc_event ev;
+  if (server->wrapped != NULL) {
+    grpc_server_shutdown_and_notify(server->wrapped, server->queue, NULL);
+    ev = rb_completion_queue_pluck(server->queue, NULL, deadline, NULL);
+    if (ev.type == GRPC_QUEUE_TIMEOUT) {
+      grpc_server_cancel_all_calls(server->wrapped);
+      rb_completion_queue_pluck(server->queue, NULL,
+                                gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+    }
+    grpc_server_destroy(server->wrapped);
+    grpc_rb_completion_queue_destroy(server->queue);
+    server->wrapped = NULL;
+    server->queue = NULL;
+  }
+}
+
 /* Destroys server instances. */
 static void grpc_rb_server_free(void *p) {
   grpc_rb_server *svr = NULL;
+  gpr_timespec deadline;
   if (p == NULL) {
     return;
   };
   svr = (grpc_rb_server *)p;
 
-  /* Deletes the wrapped object if the mark object is Qnil, which indicates
-     that no other object is the actual owner. */
-  /* grpc_server_shutdown does not exist. Change this to something that does
-     or delete it */
-  if (svr->wrapped != NULL && svr->mark == Qnil) {
-    // grpc_server_shutdown(svr->wrapped);
-    // Aborting to indicate a bug
-    abort();
-    grpc_server_destroy(svr->wrapped);
-  }
+  deadline = gpr_time_add(
+      gpr_now(GPR_CLOCK_REALTIME),
+      gpr_time_from_seconds(2, GPR_TIMESPAN));
 
-  xfree(p);
-}
+  destroy_server(svr, deadline);
 
-/* Protects the mark object from GC */
-static void grpc_rb_server_mark(void *p) {
-  grpc_rb_server *server = NULL;
-  if (p == NULL) {
-    return;
-  }
-  server = (grpc_rb_server *)p;
-  if (server->mark != Qnil) {
-    rb_gc_mark(server->mark);
-  }
+  xfree(p);
 }
 
 static const rb_data_type_t grpc_rb_server_data_type = {
     "grpc_server",
-    {grpc_rb_server_mark, grpc_rb_server_free, GRPC_RB_MEMSIZE_UNAVAILABLE,
+    {GRPC_RB_GC_NOT_MARKED, grpc_rb_server_free, GRPC_RB_MEMSIZE_UNAVAILABLE,
      {NULL, NULL}},
     NULL,
     NULL,
@@ -116,23 +115,20 @@ static const rb_data_type_t grpc_rb_server_data_type = {
 static VALUE grpc_rb_server_alloc(VALUE cls) {
   grpc_rb_server *wrapper = ALLOC(grpc_rb_server);
   wrapper->wrapped = NULL;
-  wrapper->mark = Qnil;
   return TypedData_Wrap_Struct(cls, &grpc_rb_server_data_type, wrapper);
 }
 
 /*
   call-seq:
-    cq = CompletionQueue.new
-    server = Server.new(cq, {'arg1': 'value1'})
+    server = Server.new({'arg1': 'value1'})
 
   Initializes server instances. */
-static VALUE grpc_rb_server_init(VALUE self, VALUE cqueue, VALUE channel_args) {
-  grpc_completion_queue *cq = NULL;
+static VALUE grpc_rb_server_init(VALUE self, VALUE channel_args) {
+  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
   grpc_rb_server *wrapper = NULL;
   grpc_server *srv = NULL;
   grpc_channel_args args;
   MEMZERO(&args, grpc_channel_args, 1);
-  cq = grpc_rb_get_wrapped_completion_queue(cqueue);
   TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type,
                        wrapper);
   grpc_rb_hash_convert_to_channel_args(channel_args, &args);
@@ -148,41 +144,9 @@ static VALUE grpc_rb_server_init(VALUE self, VALUE cqueue, VALUE channel_args) {
   wrapper->wrapped = srv;
   wrapper->queue = cq;
 
-  /* Add the cq as the server's mark object. This ensures the ruby cq can't be
-     GCed before the server */
-  wrapper->mark = cqueue;
   return self;
 }
 
-/* Clones Server instances.
-
-   Gives Server a consistent implementation of Ruby's object copy/dup
-   protocol. */
-static VALUE grpc_rb_server_init_copy(VALUE copy, VALUE orig) {
-  grpc_rb_server *orig_srv = NULL;
-  grpc_rb_server *copy_srv = NULL;
-
-  if (copy == orig) {
-    return copy;
-  }
-
-  /* Raise an error if orig is not a server object or a subclass. */
-  if (TYPE(orig) != T_DATA ||
-      RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_server_free) {
-    rb_raise(rb_eTypeError, "not a %s", rb_obj_classname(grpc_rb_cServer));
-  }
-
-  TypedData_Get_Struct(orig, grpc_rb_server, &grpc_rb_server_data_type,
-                       orig_srv);
-  TypedData_Get_Struct(copy, grpc_rb_server, &grpc_rb_server_data_type,
-                       copy_srv);
-
-  /* use ruby's MEMCPY to make a byte-for-byte copy of the server wrapper
-     object. */
-  MEMCPY(copy_srv, orig_srv, grpc_rb_server, 1);
-  return copy;
-}
-
 /* request_call_stack holds various values used by the
  * grpc_rb_server_request_call function */
 typedef struct request_call_stack {
@@ -208,65 +172,57 @@ static void grpc_request_call_stack_cleanup(request_call_stack* st) {
 }
 
 /* call-seq:
-   cq = CompletionQueue.new
-   tag = Object.new
-   timeout = 10
-   server.request_call(cqueue, tag, timeout)
+   server.request_call
 
    Requests notification of a new call on a server. */
-static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
-                                         VALUE tag_new, VALUE timeout) {
+static VALUE grpc_rb_server_request_call(VALUE self) {
   grpc_rb_server *s = NULL;
   grpc_call *call = NULL;
   grpc_event ev;
   grpc_call_error err;
   request_call_stack st;
   VALUE result;
+  void *tag = (void*)&st;
+  grpc_completion_queue *call_queue = grpc_completion_queue_create(NULL);
   gpr_timespec deadline;
   TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
   if (s->wrapped == NULL) {
     rb_raise(rb_eRuntimeError, "destroyed!");
     return Qnil;
-  } else {
-    grpc_request_call_stack_init(&st);
-    /* call grpc_server_request_call, then wait for it to complete using
-     * pluck_event */
-    err = grpc_server_request_call(
-        s->wrapped, &call, &st.details, &st.md_ary,
-        grpc_rb_get_wrapped_completion_queue(cqueue),
-        grpc_rb_get_wrapped_completion_queue(s->mark),
-        ROBJECT(tag_new));
-    if (err != GRPC_CALL_OK) {
-      grpc_request_call_stack_cleanup(&st);
-      rb_raise(grpc_rb_eCallError,
-              "grpc_server_request_call failed: %s (code=%d)",
-               grpc_call_error_detail_of(err), err);
-      return Qnil;
-    }
-
-    ev = grpc_rb_completion_queue_pluck_event(s->mark, tag_new, timeout);
-    if (ev.type == GRPC_QUEUE_TIMEOUT) {
-      grpc_request_call_stack_cleanup(&st);
-      return Qnil;
-    }
-    if (!ev.success) {
-      grpc_request_call_stack_cleanup(&st);
-      rb_raise(grpc_rb_eCallError, "request_call completion failed");
-      return Qnil;
-    }
+  }
+  grpc_request_call_stack_init(&st);
+  /* call grpc_server_request_call, then wait for it to complete using
+   * pluck_event */
+  err = grpc_server_request_call(
+      s->wrapped, &call, &st.details, &st.md_ary,
+      call_queue, s->queue, tag);
+  if (err != GRPC_CALL_OK) {
+    grpc_request_call_stack_cleanup(&st);
+    rb_raise(grpc_rb_eCallError,
+             "grpc_server_request_call failed: %s (code=%d)",
+             grpc_call_error_detail_of(err), err);
+    return Qnil;
+  }
 
-    /* build the NewServerRpc struct result */
-    deadline = gpr_convert_clock_type(st.details.deadline, GPR_CLOCK_REALTIME);
-    result = rb_struct_new(
-        grpc_rb_sNewServerRpc, rb_str_new2(st.details.method),
-        rb_str_new2(st.details.host),
-        rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec),
-                   INT2NUM(deadline.tv_nsec)),
-        grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call), cqueue, NULL);
+  ev = rb_completion_queue_pluck(s->queue, tag,
+                                 gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
+  if (!ev.success) {
     grpc_request_call_stack_cleanup(&st);
-    return result;
+    rb_raise(grpc_rb_eCallError, "request_call completion failed");
+    return Qnil;
   }
-  return Qnil;
+
+  /* build the NewServerRpc struct result */
+  deadline = gpr_convert_clock_type(st.details.deadline, GPR_CLOCK_REALTIME);
+  result = rb_struct_new(
+      grpc_rb_sNewServerRpc, rb_str_new2(st.details.method),
+      rb_str_new2(st.details.host),
+      rb_funcall(rb_cTime, id_at, 2, INT2NUM(deadline.tv_sec),
+                 INT2NUM(deadline.tv_nsec / 1000)),
+      grpc_rb_md_ary_to_h(&st.md_ary), grpc_rb_wrap_call(call, call_queue),
+      NULL);
+  grpc_request_call_stack_cleanup(&st);
+  return result;
 }
 
 static VALUE grpc_rb_server_start(VALUE self) {
@@ -282,41 +238,33 @@ static VALUE grpc_rb_server_start(VALUE self) {
 
 /*
   call-seq:
-    cq = CompletionQueue.new
-    server = Server.new(cq, {'arg1': 'value1'})
+    server = Server.new({'arg1': 'value1'})
     ... // do stuff with server
     ...
     ... // to shutdown the server
-    server.destroy(cq)
+    server.destroy()
 
     ... // to shutdown the server with a timeout
-    server.destroy(cq, timeout)
+    server.destroy(timeout)
 
   Destroys server instances. */
 static VALUE grpc_rb_server_destroy(int argc, VALUE *argv, VALUE self) {
-  VALUE cqueue = Qnil;
   VALUE timeout = Qnil;
-  grpc_completion_queue *cq = NULL;
-  grpc_event ev;
+  gpr_timespec deadline;
   grpc_rb_server *s = NULL;
 
-  /* "11" == 1 mandatory args, 1 (timeout) is optional */
-  rb_scan_args(argc, argv, "11", &cqueue, &timeout);
-  cq = grpc_rb_get_wrapped_completion_queue(cqueue);
+  /* "01" == 0 mandatory args, 1 (timeout) is optional */
+  rb_scan_args(argc, argv, "01", &timeout);
   TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type, s);
-
-  if (s->wrapped != NULL) {
-    grpc_server_shutdown_and_notify(s->wrapped, cq, NULL);
-    ev = grpc_rb_completion_queue_pluck_event(cqueue, Qnil, timeout);
-    if (!ev.success) {
-      rb_warn("server shutdown failed, cancelling the calls, objects may leak");
-      grpc_server_cancel_all_calls(s->wrapped);
-      return Qfalse;
-    }
-    grpc_server_destroy(s->wrapped);
-    s->wrapped = NULL;
+  if (TYPE(timeout) == T_NIL) {
+    deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
+  } else {
+    deadline = grpc_rb_time_timeval(timeout, /* absolute time*/ 0);
   }
-  return Qtrue;
+
+  destroy_server(s, deadline);
+
+  return Qnil;
 }
 
 /*
@@ -376,13 +324,13 @@ void Init_grpc_server() {
   rb_define_alloc_func(grpc_rb_cServer, grpc_rb_server_alloc);
 
   /* Provides a ruby constructor and support for dup/clone. */
-  rb_define_method(grpc_rb_cServer, "initialize", grpc_rb_server_init, 2);
+  rb_define_method(grpc_rb_cServer, "initialize", grpc_rb_server_init, 1);
   rb_define_method(grpc_rb_cServer, "initialize_copy",
-                   grpc_rb_server_init_copy, 1);
+                   grpc_rb_cannot_init_copy, 1);
 
   /* Add the server methods. */
   rb_define_method(grpc_rb_cServer, "request_call",
-                   grpc_rb_server_request_call, 3);
+                   grpc_rb_server_request_call, 0);
   rb_define_method(grpc_rb_cServer, "start", grpc_rb_server_start, 0);
   rb_define_method(grpc_rb_cServer, "destroy", grpc_rb_server_destroy, -1);
   rb_define_alias(grpc_rb_cServer, "close", "destroy");
diff --git a/src/ruby/ext/grpc/rb_server_credentials.c b/src/ruby/ext/grpc/rb_server_credentials.c
index 3b0fb6c910..a44ce715ae 100644
--- a/src/ruby/ext/grpc/rb_server_credentials.c
+++ b/src/ruby/ext/grpc/rb_server_credentials.c
@@ -38,6 +38,7 @@
 
 #include <grpc/grpc.h>
 #include <grpc/grpc_security.h>
+#include <grpc/support/log.h>
 
 #include "rb_grpc.h"
 
@@ -46,8 +47,8 @@
 static VALUE grpc_rb_cServerCredentials = Qnil;
 
 /* grpc_rb_server_credentials wraps a grpc_server_credentials.  It provides a
-   peer ruby object, 'mark' to minimize copying when a server credential is
-   created from ruby. */
+   peer ruby object, 'mark' to hold references to objects involved in
+   constructing the server credentials. */
 typedef struct grpc_rb_server_credentials {
   /* Holder of ruby objects involved in constructing the server credentials */
   VALUE mark;
@@ -111,36 +112,6 @@ static VALUE grpc_rb_server_credentials_alloc(VALUE cls) {
                                wrapper);
 }
 
-/* Clones ServerCredentials instances.
-
-   Gives ServerCredentials a consistent implementation of Ruby's object copy/dup
-   protocol. */
-static VALUE grpc_rb_server_credentials_init_copy(VALUE copy, VALUE orig) {
-  grpc_rb_server_credentials *orig_ch = NULL;
-  grpc_rb_server_credentials *copy_ch = NULL;
-
-  if (copy == orig) {
-    return copy;
-  }
-
-  /* Raise an error if orig is not a server_credentials object or a subclass. */
-  if (TYPE(orig) != T_DATA ||
-      RDATA(orig)->dfree != (RUBY_DATA_FUNC)grpc_rb_server_credentials_free) {
-    rb_raise(rb_eTypeError, "not a %s",
-             rb_obj_classname(grpc_rb_cServerCredentials));
-  }
-
-  TypedData_Get_Struct(orig, grpc_rb_server_credentials,
-                       &grpc_rb_server_credentials_data_type, orig_ch);
-  TypedData_Get_Struct(copy, grpc_rb_server_credentials,
-                       &grpc_rb_server_credentials_data_type, copy_ch);
-
-  /* use ruby's MEMCPY to make a byte-for-byte copy of the server_credentials
-     wrapper object. */
-  MEMCPY(copy_ch, orig_ch, grpc_rb_server_credentials, 1);
-  return copy;
-}
-
 /* The attribute used on the mark object to preserve the pem_root_certs. */
 static ID id_pem_root_certs;
 
@@ -270,7 +241,7 @@ void Init_grpc_server_credentials() {
   rb_define_method(grpc_rb_cServerCredentials, "initialize",
                    grpc_rb_server_credentials_init, 3);
   rb_define_method(grpc_rb_cServerCredentials, "initialize_copy",
-                   grpc_rb_server_credentials_init_copy, 1);
+                   grpc_rb_cannot_init_copy, 1);
 
   id_pem_key_certs = rb_intern("__pem_key_certs");
   id_pem_root_certs = rb_intern("__pem_root_certs");
diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb
index b03ddbc193..23688dc924 100644
--- a/src/ruby/lib/grpc/generic/active_call.rb
+++ b/src/ruby/lib/grpc/generic/active_call.rb
@@ -43,8 +43,7 @@ class Struct
         GRPC.logger.debug("Failing with status #{status}")
         # raise BadStatus, propagating the metadata if present.
         md = status.metadata
-        with_sym_keys = Hash[md.each_pair.collect { |x, y| [x.to_sym, y] }]
-        fail GRPC::BadStatus.new(status.code, status.details, with_sym_keys)
+        fail GRPC::BadStatus.new(status.code, status.details, md)
       end
       status
     end
@@ -59,9 +58,9 @@ module GRPC
     include Core::TimeConsts
     include Core::CallOps
     extend Forwardable
-    attr_reader(:deadline)
+    attr_reader :deadline, :metadata_sent, :metadata_to_send
     def_delegators :@call, :cancel, :metadata, :write_flag, :write_flag=,
-                   :peer, :peer_cert
+                   :peer, :peer_cert, :trailing_metadata
 
     # client_invoke begins a client invocation.
     #
@@ -75,17 +74,10 @@ module GRPC
     # if a keyword value is a list, multiple metadata for it's key are sent
     #
     # @param call [Call] a call on which to start and invocation
-    # @param q [CompletionQueue] the completion queue
     # @param metadata [Hash] the metadata
-    def self.client_invoke(call, q, metadata = {})
+    def self.client_invoke(call, metadata = {})
       fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
-      unless q.is_a? Core::CompletionQueue
-        fail(TypeError, '!Core::CompletionQueue')
-      end
-      metadata_tag = Object.new
-      call.run_batch(q, metadata_tag, INFINITE_FUTURE,
-                     SEND_INITIAL_METADATA => metadata)
-      metadata_tag
+      call.run_batch(SEND_INITIAL_METADATA => metadata)
     end
 
     # Creates an ActiveCall.
@@ -102,27 +94,36 @@ module GRPC
     # deadline is the absolute deadline for the call.
     #
     # @param call [Call] the call used by the ActiveCall
-    # @param q [CompletionQueue] the completion queue used to accept
-    #          the call.  This queue will be closed on call completion.
     # @param marshal [Function] f(obj)->string that marshal requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param deadline [Fixnum] the deadline for the call to complete
-    # @param metadata_tag [Object] the object use obtain metadata for clients
-    # @param started [true|false] indicates if the call has begun
-    def initialize(call, q, marshal, unmarshal, deadline, started: true,
-                   metadata_tag: nil)
+    # @param started [true|false] indicates that metadata was sent
+    # @param metadata_received [true|false] indicates if metadata has already
+    #     been received. Should always be true for server calls
+    def initialize(call, marshal, unmarshal, deadline, started: true,
+                   metadata_received: false, metadata_to_send: nil)
       fail(TypeError, '!Core::Call') unless call.is_a? Core::Call
-      unless q.is_a? Core::CompletionQueue
-        fail(TypeError, '!Core::CompletionQueue')
-      end
       @call = call
-      @cq = q
       @deadline = deadline
       @marshal = marshal
-      @started = started
       @unmarshal = unmarshal
-      @metadata_tag = metadata_tag
+      @metadata_received = metadata_received
+      @metadata_sent = started
       @op_notifier = nil
+
+      fail(ArgumentError, 'Already sent md') if started && metadata_to_send
+      @metadata_to_send = metadata_to_send || {} unless started
+      @send_initial_md_mutex = Mutex.new
+    end
+
+    # Sends the initial metadata that has yet to be sent.
+    # Does nothing if metadata has already been sent for this call.
+    def send_initial_metadata
+      @send_initial_md_mutex.synchronize do
+        return if @metadata_sent
+        @metadata_tag = ActiveCall.client_invoke(@call, @metadata_to_send)
+        @metadata_sent = true
+      end
     end
 
     # output_metadata are provides access to hash that can be used to
@@ -132,7 +133,7 @@ module GRPC
     end
 
     # cancelled indicates if the call was cancelled
-    def cancelled
+    def cancelled?
       !@call.status.nil? && @call.status.code == Core::StatusCodes::CANCELLED
     end
 
@@ -168,8 +169,11 @@ module GRPC
         SEND_CLOSE_FROM_CLIENT => nil
       }
       ops[RECV_STATUS_ON_CLIENT] = nil if assert_finished
-      batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
+      batch_result = @call.run_batch(ops)
       return unless assert_finished
+      unless batch_result.status.nil?
+        @call.trailing_metadata = batch_result.status.metadata
+      end
       @call.status = batch_result.status
       op_is_done
       batch_result.check_status
@@ -179,20 +183,14 @@ module GRPC
     #
     # It blocks until the remote endpoint acknowledges by sending a status.
     def finished
-      batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE,
-                                     RECV_STATUS_ON_CLIENT => nil)
+      batch_result = @call.run_batch(RECV_STATUS_ON_CLIENT => nil)
       unless batch_result.status.nil?
-        if @call.metadata.nil?
-          @call.metadata = batch_result.status.metadata
-        else
-          @call.metadata.merge!(batch_result.status.metadata)
-        end
+        @call.trailing_metadata = batch_result.status.metadata
       end
       @call.status = batch_result.status
       op_is_done
       batch_result.check_status
       @call.close
-      @cq.close
     end
 
     # remote_send sends a request to the remote endpoint.
@@ -203,9 +201,10 @@ module GRPC
     # @param marshalled [false, true] indicates if the object is already
     # marshalled.
     def remote_send(req, marshalled = false)
+      send_initial_metadata
       GRPC.logger.debug("sending #{req}, marshalled? #{marshalled}")
       payload = marshalled ? req : @marshal.call(req)
-      @call.run_batch(@cq, self, INFINITE_FUTURE, SEND_MESSAGE => payload)
+      @call.run_batch(SEND_MESSAGE => payload)
     end
 
     # send_status sends a status to the remote endpoint.
@@ -218,11 +217,12 @@ module GRPC
     # list, mulitple metadata for its key are sent
     def send_status(code = OK, details = '', assert_finished = false,
                     metadata: {})
+      send_initial_metadata
       ops = {
         SEND_STATUS_FROM_SERVER => Struct::Status.new(code, details, metadata)
       }
       ops[RECV_CLOSE_ON_SERVER] = nil if assert_finished
-      @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
+      @call.run_batch(ops)
       nil
     end
 
@@ -234,11 +234,11 @@ module GRPC
     # raising BadStatus
     def remote_read
       ops = { RECV_MESSAGE => nil }
-      ops[RECV_INITIAL_METADATA] = nil unless @metadata_tag.nil?
-      batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
-      unless @metadata_tag.nil?
+      ops[RECV_INITIAL_METADATA] = nil unless @metadata_received
+      batch_result = @call.run_batch(ops)
+      unless @metadata_received
         @call.metadata = batch_result.metadata
-        @metadata_tag = nil
+        @metadata_received = true
       end
       GRPC.logger.debug("received req: #{batch_result}")
       unless batch_result.nil? || batch_result.message.nil?
@@ -318,7 +318,7 @@ module GRPC
     # a list, multiple metadata for its key are sent
     # @return [Object] the response received from the server
     def request_response(req, metadata: {})
-      start_call(metadata) unless @started
+      merge_metadata_to_send(metadata) && send_initial_metadata
       remote_send(req)
       writes_done(false)
       response = remote_read
@@ -342,7 +342,7 @@ module GRPC
     # a list, multiple metadata for its key are sent
     # @return [Object] the response received from the server
     def client_streamer(requests, metadata: {})
-      start_call(metadata) unless @started
+      merge_metadata_to_send(metadata) && send_initial_metadata
       requests.each { |r| remote_send(r) }
       writes_done(false)
       response = remote_read
@@ -368,7 +368,7 @@ module GRPC
     # a list, multiple metadata for its key are sent
     # @return [Enumerator|nil] a response Enumerator
     def server_streamer(req, metadata: {})
-      start_call(metadata) unless @started
+      merge_metadata_to_send(metadata) && send_initial_metadata
       remote_send(req)
       writes_done(false)
       replies = enum_for(:each_remote_read_then_finish)
@@ -407,10 +407,12 @@ module GRPC
     # a list, multiple metadata for its key are sent
     # @return [Enumerator, nil] a response Enumerator
     def bidi_streamer(requests, metadata: {}, &blk)
-      start_call(metadata) unless @started
-      bd = BidiCall.new(@call, @cq, @marshal, @unmarshal,
-                        metadata_tag: @metadata_tag)
-      @metadata_tag = nil  # run_on_client ensures metadata is read
+      merge_metadata_to_send(metadata) && send_initial_metadata
+      bd = BidiCall.new(@call,
+                        @marshal,
+                        @unmarshal,
+                        metadata_received: @metadata_received)
+
       bd.run_on_client(requests, @op_notifier, &blk)
     end
 
@@ -426,7 +428,12 @@ module GRPC
     #
     # @param gen_each_reply [Proc] generates the BiDi stream replies
     def run_server_bidi(gen_each_reply)
-      bd = BidiCall.new(@call, @cq, @marshal, @unmarshal)
+      bd = BidiCall.new(@call,
+                        @marshal,
+                        @unmarshal,
+                        metadata_received: @metadata_received,
+                        req_view: MultiReqView.new(self))
+
       bd.run_on_server(gen_each_reply)
     end
 
@@ -443,15 +450,23 @@ module GRPC
       @op_notifier.notify(self)
     end
 
+    # Add to the metadata that will be sent from the server.
+    # Fails if metadata has already been sent.
+    # Unused by client calls.
+    def merge_metadata_to_send(new_metadata = {})
+      @send_initial_md_mutex.synchronize do
+        fail('cant change metadata after already sent') if @metadata_sent
+        @metadata_to_send.merge!(new_metadata)
+      end
+    end
+
     private
 
     # Starts the call if not already started
     # @param metadata [Hash] metadata to be sent to the server. If a value is
     # a list, multiple metadata for its key are sent
     def start_call(metadata = {})
-      return if @started
-      @metadata_tag = ActiveCall.client_invoke(@call, @cq, metadata)
-      @started = true
+      merge_metadata_to_send(metadata) && send_initial_metadata
     end
 
     def self.view_class(*visible_methods)
@@ -468,18 +483,26 @@ module GRPC
 
     # SingleReqView limits access to an ActiveCall's methods for use in server
     # handlers that receive just one request.
-    SingleReqView = view_class(:cancelled, :deadline, :metadata,
-                               :output_metadata, :peer, :peer_cert)
+    SingleReqView = view_class(:cancelled?, :deadline, :metadata,
+                               :output_metadata, :peer, :peer_cert,
+                               :send_initial_metadata,
+                               :metadata_to_send,
+                               :merge_metadata_to_send,
+                               :metadata_sent)
 
     # MultiReqView limits access to an ActiveCall's methods for use in
     # server client_streamer handlers.
-    MultiReqView = view_class(:cancelled, :deadline, :each_queued_msg,
-                              :each_remote_read, :metadata, :output_metadata)
+    MultiReqView = view_class(:cancelled?, :deadline, :each_queued_msg,
+                              :each_remote_read, :metadata, :output_metadata,
+                              :send_initial_metadata,
+                              :metadata_to_send,
+                              :merge_metadata_to_send,
+                              :metadata_sent)
 
     # Operation limits access to an ActiveCall's methods for use as
     # a Operation on the client.
-    Operation = view_class(:cancel, :cancelled, :deadline, :execute,
+    Operation = view_class(:cancel, :cancelled?, :deadline, :execute,
                            :metadata, :status, :start_call, :wait, :write_flag,
-                           :write_flag=)
+                           :write_flag=, :trailing_metadata)
   end
 end
diff --git a/src/ruby/lib/grpc/generic/bidi_call.rb b/src/ruby/lib/grpc/generic/bidi_call.rb
index 238f409a1d..d7cd9e6df2 100644
--- a/src/ruby/lib/grpc/generic/bidi_call.rb
+++ b/src/ruby/lib/grpc/generic/bidi_call.rb
@@ -52,27 +52,23 @@ module GRPC
     # deadline is the absolute deadline for the call.
     #
     # @param call [Call] the call used by the ActiveCall
-    # @param q [CompletionQueue] the completion queue used to accept
-    #          the call
     # @param marshal [Function] f(obj)->string that marshal requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
-    # @param metadata_tag [Object] tag object used to collect metadata
-    def initialize(call, q, marshal, unmarshal, metadata_tag: nil)
+    # @param metadata_received [true|false] indicates if metadata has already
+    #     been received. Should always be true for server calls
+    def initialize(call, marshal, unmarshal, metadata_received: false,
+                   req_view: nil)
       fail(ArgumentError, 'not a call') unless call.is_a? Core::Call
-      unless q.is_a? Core::CompletionQueue
-        fail(ArgumentError, 'not a CompletionQueue')
-      end
       @call = call
-      @cq = q
       @marshal = marshal
       @op_notifier = nil  # signals completion on clients
-      @readq = Queue.new
       @unmarshal = unmarshal
-      @metadata_tag = metadata_tag
+      @metadata_received = metadata_received
       @reads_complete = false
       @writes_complete = false
       @complete = false
       @done_mutex = Mutex.new
+      @req_view = req_view
     end
 
     # Begins orchestration of the Bidi stream for a client sending requests.
@@ -81,13 +77,12 @@ module GRPC
     # block that can be invoked with each response.
     #
     # @param requests the Enumerable of requests to send
-    # @op_notifier a Notifier used to signal completion
+    # @param op_notifier a Notifier used to signal completion
     # @return an Enumerator of requests to yield
     def run_on_client(requests, op_notifier, &blk)
       @op_notifier = op_notifier
       @enq_th = Thread.new { write_loop(requests) }
-      @loop_th = start_read_loop
-      each_queued_msg(&blk)
+      read_loop(&blk)
     end
 
     # Begins orchestration of the Bidi stream for a server generating replies.
@@ -102,8 +97,15 @@ module GRPC
     #
     # @param gen_each_reply [Proc] generates the BiDi stream replies.
     def run_on_server(gen_each_reply)
-      replys = gen_each_reply.call(each_queued_msg)
-      @loop_th = start_read_loop(is_client: false)
+      # Pass in the optional call object parameter if possible
+      if gen_each_reply.arity == 1
+        replys = gen_each_reply.call(read_loop(is_client: false))
+      elsif gen_each_reply.arity == 2
+        replys = gen_each_reply.call(read_loop(is_client: false), @req_view)
+      else
+        fail 'Illegal arity of reply generator'
+      end
+
       write_loop(replys, is_client: false)
     end
 
@@ -124,7 +126,6 @@ module GRPC
       @done_mutex.synchronize do
         return unless @reads_complete && @writes_complete && !@complete
         @call.close
-        @cq.close
         @complete = true
       end
     end
@@ -132,49 +133,38 @@ module GRPC
     # performs a read using @call.run_batch, ensures metadata is set up
     def read_using_run_batch
       ops = { RECV_MESSAGE => nil }
-      ops[RECV_INITIAL_METADATA] = nil unless @metadata_tag.nil?
-      batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
-      unless @metadata_tag.nil?
+      ops[RECV_INITIAL_METADATA] = nil unless @metadata_received
+      batch_result = @call.run_batch(ops)
+      unless @metadata_received
         @call.metadata = batch_result.metadata
-        @metadata_tag = nil
+        @metadata_received = true
       end
       batch_result
     end
 
-    # each_queued_msg yields each message on this instances readq
-    #
-    # - messages are added to the readq by #read_loop
-    # - iteration ends when the instance itself is added
-    def each_queued_msg
-      return enum_for(:each_queued_msg) unless block_given?
-      count = 0
-      loop do
-        GRPC.logger.debug("each_queued_msg: waiting##{count}")
-        count += 1
-        req = @readq.pop
-        GRPC.logger.debug("each_queued_msg: req = #{req}")
-        fail req if req.is_a? StandardError
-        break if req.equal?(END_OF_READS)
-        yield req
-      end
-    end
-
     def write_loop(requests, is_client: true)
       GRPC.logger.debug('bidi-write-loop: starting')
-      write_tag = Object.new
       count = 0
       requests.each do |req|
         GRPC.logger.debug("bidi-write-loop: #{count}")
         count += 1
         payload = @marshal.call(req)
-        @call.run_batch(@cq, write_tag, INFINITE_FUTURE,
-                        SEND_MESSAGE => payload)
+        # Fails if status already received
+        begin
+          @req_view.send_initial_metadata unless @req_view.nil?
+          @call.run_batch(SEND_MESSAGE => payload)
+        rescue GRPC::Core::CallError => e
+          # This is almost definitely caused by a status arriving while still
+          # writing. Don't re-throw the error
+          GRPC.logger.warn('bidi-write-loop: ended with error')
+          GRPC.logger.warn(e)
+          break
+        end
       end
       GRPC.logger.debug("bidi-write-loop: #{count} writes done")
       if is_client
         GRPC.logger.debug("bidi-write-loop: client sent #{count}, waiting")
-        @call.run_batch(@cq, write_tag, INFINITE_FUTURE,
-                        SEND_CLOSE_FROM_CLIENT => nil)
+        @call.run_batch(SEND_CLOSE_FROM_CLIENT => nil)
         GRPC.logger.debug('bidi-write-loop: done')
         notify_done
         @writes_complete = true
@@ -190,49 +180,45 @@ module GRPC
       raise e
     end
 
-    # starts the read loop
-    def start_read_loop(is_client: true)
-      Thread.new do
-        GRPC.logger.debug('bidi-read-loop: starting')
-        begin
-          read_tag = Object.new
-          count = 0
-          # queue the initial read before beginning the loop
-          loop do
-            GRPC.logger.debug("bidi-read-loop: #{count}")
-            count += 1
-            batch_result = read_using_run_batch
-
-            # handle the next message
-            if batch_result.message.nil?
-              GRPC.logger.debug("bidi-read-loop: null batch #{batch_result}")
-
-              if is_client
-                batch_result = @call.run_batch(@cq, read_tag, INFINITE_FUTURE,
-                                               RECV_STATUS_ON_CLIENT => nil)
-                @call.status = batch_result.status
-                batch_result.check_status
-                GRPC.logger.debug("bidi-read-loop: done status #{@call.status}")
-              end
-
-              @readq.push(END_OF_READS)
-              GRPC.logger.debug('bidi-read-loop: done reading!')
-              break
+    # Provides an enumerator that yields results of remote reads
+    def read_loop(is_client: true)
+      return enum_for(:read_loop,
+                      is_client: is_client) unless block_given?
+      GRPC.logger.debug('bidi-read-loop: starting')
+      begin
+        count = 0
+        # queue the initial read before beginning the loop
+        loop do
+          GRPC.logger.debug("bidi-read-loop: #{count}")
+          count += 1
+          batch_result = read_using_run_batch
+
+          # handle the next message
+          if batch_result.message.nil?
+            GRPC.logger.debug("bidi-read-loop: null batch #{batch_result}")
+
+            if is_client
+              batch_result = @call.run_batch(RECV_STATUS_ON_CLIENT => nil)
+              @call.status = batch_result.status
+              batch_result.check_status
+              GRPC.logger.debug("bidi-read-loop: done status #{@call.status}")
             end
 
-            # push the latest read onto the queue and continue reading
-            res = @unmarshal.call(batch_result.message)
-            @readq.push(res)
+            GRPC.logger.debug('bidi-read-loop: done reading!')
+            break
           end
-        rescue StandardError => e
-          GRPC.logger.warn('bidi: read-loop failed')
-          GRPC.logger.warn(e)
-          @readq.push(e)  # let each_queued_msg terminate with this error
+
+          res = @unmarshal.call(batch_result.message)
+          yield res
         end
-        GRPC.logger.debug('bidi-read-loop: finished')
-        @reads_complete = true
-        finished
+      rescue StandardError => e
+        GRPC.logger.warn('bidi: read-loop failed')
+        GRPC.logger.warn(e)
+        raise e
       end
+      GRPC.logger.debug('bidi-read-loop: finished')
+      @reads_complete = true
+      finished
     end
   end
 end
diff --git a/src/ruby/lib/grpc/generic/client_stub.rb b/src/ruby/lib/grpc/generic/client_stub.rb
index cddca13d17..0d7c1f7805 100644
--- a/src/ruby/lib/grpc/generic/client_stub.rb
+++ b/src/ruby/lib/grpc/generic/client_stub.rb
@@ -34,7 +34,8 @@ require_relative '../version'
 module GRPC
   # rubocop:disable Metrics/ParameterLists
 
-  # ClientStub represents an endpoint used to send requests to GRPC servers.
+  # ClientStub represents a client connection to a gRPC server, and can be used
+  # to send requests.
   class ClientStub
     include Core::StatusCodes
     include Core::TimeConsts
@@ -75,8 +76,9 @@ module GRPC
     # my_stub = ClientStub.new(example.host.com:50505,
     #                          :this_channel_is_insecure)
     #
-    # Any arbitrary keyword arguments are treated as channel arguments used to
-    # configure the RPC connection to the host.
+    # If a channel_override argument is passed, it will be used as the
+    # underlying channel. Otherwise, the channel_args argument will be used
+    # to construct a new underlying channel.
     #
     # There are some specific keyword args that are not used to configure the
     # channel:
@@ -90,19 +92,23 @@ module GRPC
     # when present, this is the default timeout used for calls
     #
     # @param host [String] the host the stub connects to
-    # @param q [Core::CompletionQueue] used to wait for events - now deprecated
-    #        since each new active call gets its own separately
     # @param creds [Core::ChannelCredentials|Symbol] the channel credentials, or
-    #     :this_channel_is_insecure
+    #     :this_channel_is_insecure, which explicitly indicates that the client
+    #     should be created with an insecure connection. Note: this argument is
+    #     ignored if the channel_override argument is provided.
     # @param channel_override [Core::Channel] a pre-created channel
     # @param timeout [Number] the default timeout to use in requests
-    # @param channel_args [Hash] the channel arguments
-    def initialize(host, q, creds,
+    # @param propagate_mask [Number] A bitwise combination of flags in
+    #     GRPC::Core::PropagateMasks. Indicates how data should be propagated
+    #     from parent server calls to child client calls if this client is being
+    #     used within a gRPC server.
+    # @param channel_args [Hash] the channel arguments. Note: this argument is
+    #     ignored if the channel_override argument is provided.
+    def initialize(host, creds,
                    channel_override: nil,
                    timeout: nil,
                    propagate_mask: nil,
                    channel_args: {})
-      fail(TypeError, '!CompletionQueue') unless q.is_a?(Core::CompletionQueue)
       @ch = ClientStub.setup_channel(channel_override, host, creds,
                                      channel_args)
       alt_host = channel_args[Core::Channel::SSL_TARGET]
@@ -392,11 +398,11 @@ module GRPC
     # @param marshal [Function] f(obj)->string that marshals requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param deadline [Time] (optional) the time the request should complete
+    # @param return_op [true|false] return an Operation if true
     # @param parent [Core::Call] a prior call whose reserved metadata
     #   will be propagated by this one.
     # @param credentials [Core::CallCredentials] credentials to use when making
     #   the call
-    # @param return_op [true|false] return an Operation if true
     # @param metadata [Hash] metadata to be sent to the server
     # @param blk [Block] when provided, is executed for each response
     # @return [Enumerator|nil|Operation] as discussed above
@@ -433,7 +439,8 @@ module GRPC
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param parent [Grpc::Call] a parent call, available when calls are
     #   made from server
-    # @param timeout [TimeConst]
+    # @param credentials [Core::CallCredentials] credentials to use when making
+    #   the call
     def new_active_call(method, marshal, unmarshal,
                         deadline: nil,
                         parent: nil,
@@ -441,15 +448,13 @@ module GRPC
 
       deadline = from_relative_time(@timeout) if deadline.nil?
       # Provide each new client call with its own completion queue
-      call_queue = Core::CompletionQueue.new
-      call = @ch.create_call(call_queue,
-                             parent, # parent call
+      call = @ch.create_call(parent, # parent call
                              @propagate_mask, # propagation options
                              method,
                              nil, # host use nil,
                              deadline)
       call.set_credentials! credentials unless credentials.nil?
-      ActiveCall.new(call, call_queue, marshal, unmarshal, deadline,
+      ActiveCall.new(call, marshal, unmarshal, deadline,
                      started: false)
     end
   end
diff --git a/src/ruby/lib/grpc/generic/rpc_desc.rb b/src/ruby/lib/grpc/generic/rpc_desc.rb
index 913f55d0d3..584fe78169 100644
--- a/src/ruby/lib/grpc/generic/rpc_desc.rb
+++ b/src/ruby/lib/grpc/generic/rpc_desc.rb
@@ -104,7 +104,14 @@ module GRPC
     end
 
     def assert_arity_matches(mth)
-      if request_response? || server_streamer?
+      # A bidi handler function can optionally be passed a second
+      # call object parameter for access to metadata, cancelling, etc.
+      if bidi_streamer?
+        if mth.arity != 2 && mth.arity != 1
+          fail arity_error(mth, 2, "should be #{mth.name}(req, call) or " \
+            "#{mth.name}(req)")
+        end
+      elsif request_response? || server_streamer?
         if mth.arity != 2
           fail arity_error(mth, 2, "should be #{mth.name}(req, call)")
         end
diff --git a/src/ruby/lib/grpc/generic/rpc_server.rb b/src/ruby/lib/grpc/generic/rpc_server.rb
index ab7333d133..8ea798dce0 100644
--- a/src/ruby/lib/grpc/generic/rpc_server.rb
+++ b/src/ruby/lib/grpc/generic/rpc_server.rb
@@ -159,16 +159,6 @@ module GRPC
     # Signal check period is 0.25s
     SIGNAL_CHECK_PERIOD = 0.25
 
-    # setup_cq is used by #initialize to constuct a Core::CompletionQueue from
-    # its arguments.
-    def self.setup_cq(alt_cq)
-      return Core::CompletionQueue.new if alt_cq.nil?
-      unless alt_cq.is_a? Core::CompletionQueue
-        fail(TypeError, '!CompletionQueue')
-      end
-      alt_cq
-    end
-
     # setup_connect_md_proc is used by #initialize to validate the
     # connect_md_proc.
     def self.setup_connect_md_proc(a_proc)
@@ -182,26 +172,18 @@ module GRPC
     # The RPC server is configured using keyword arguments.
     #
     # There are some specific keyword args used to configure the RpcServer
-    # instance, however other arbitrary are allowed and when present are used
-    # to configure the listeninng connection set up by the RpcServer.
-    #
-    # * poll_period: when present, the server polls for new events with this
-    # period
+    # instance.
     #
     # * pool_size: the size of the thread pool the server uses to run its
     # threads
     #
-    # * completion_queue_override: when supplied, this will be used as the
-    # completion_queue that the server uses to receive network events,
-    # otherwise its creates a new instance itself
-    #
-    # * creds: [GRPC::Core::ServerCredentials]
-    # the credentials used to secure the server
-    #
     # * max_waiting_requests: the maximum number of requests that are not
     # being handled to allow. When this limit is exceeded, the server responds
     # with not available to new requests
     #
+    # * poll_period: when present, the server polls for new events with this
+    # period
+    #
     # * connect_md_proc:
     # when non-nil is a proc for determining metadata to to send back the client
     # on receiving an invocation req.  The proc signature is:
@@ -212,11 +194,9 @@ module GRPC
     def initialize(pool_size:DEFAULT_POOL_SIZE,
                    max_waiting_requests:DEFAULT_MAX_WAITING_REQUESTS,
                    poll_period:DEFAULT_POLL_PERIOD,
-                   completion_queue_override:nil,
                    connect_md_proc:nil,
                    server_args:{})
       @connect_md_proc = RpcServer.setup_connect_md_proc(connect_md_proc)
-      @cq = RpcServer.setup_cq(completion_queue_override)
       @max_waiting_requests = max_waiting_requests
       @poll_period = poll_period
       @pool_size = pool_size
@@ -226,7 +206,7 @@ module GRPC
       # running_state can take 4 values: :not_started, :running, :stopping, and
       # :stopped. State transitions can only proceed in that order.
       @running_state = :not_started
-      @server = Core::Server.new(@cq, server_args)
+      @server = Core::Server.new(server_args)
     end
 
     # stops a running server
@@ -240,7 +220,7 @@ module GRPC
         transition_running_state(:stopping)
       end
       deadline = from_relative_time(@poll_period)
-      @server.close(@cq, deadline)
+      @server.close(deadline)
       @pool.stop
     end
 
@@ -355,7 +335,11 @@ module GRPC
       return an_rpc if @pool.jobs_waiting <= @max_waiting_requests
       GRPC.logger.warn("NOT AVAILABLE: too many jobs_waiting: #{an_rpc}")
       noop = proc { |x| x }
-      c = ActiveCall.new(an_rpc.call, an_rpc.cq, noop, noop, an_rpc.deadline)
+
+      # Create a new active call that knows that metadata hasn't been
+      # sent yet
+      c = ActiveCall.new(an_rpc.call, noop, noop, an_rpc.deadline,
+                         metadata_received: true, started: false)
       c.send_status(GRPC::Core::StatusCodes::RESOURCE_EXHAUSTED, '')
       nil
     end
@@ -366,7 +350,11 @@ module GRPC
       return an_rpc if rpc_descs.key?(mth)
       GRPC.logger.warn("UNIMPLEMENTED: #{an_rpc}")
       noop = proc { |x| x }
-      c = ActiveCall.new(an_rpc.call, an_rpc.cq, noop, noop, an_rpc.deadline)
+
+      # Create a new active call that knows that
+      # metadata hasn't been sent yet
+      c = ActiveCall.new(an_rpc.call, noop, noop, an_rpc.deadline,
+                         metadata_received: true, started: false)
       c.send_status(GRPC::Core::StatusCodes::UNIMPLEMENTED, '')
       nil
     end
@@ -374,11 +362,9 @@ module GRPC
     # handles calls to the server
     def loop_handle_server_calls
       fail 'not started' if running_state == :not_started
-      loop_tag = Object.new
       while running_state == :running
         begin
-          comp_queue = Core::CompletionQueue.new
-          an_rpc = @server.request_call(comp_queue, loop_tag, INFINITE_FUTURE)
+          an_rpc = @server.request_call
           break if (!an_rpc.nil?) && an_rpc.call.nil?
           active_call = new_active_server_call(an_rpc)
           unless active_call.nil?
@@ -410,25 +396,26 @@ module GRPC
       return nil if an_rpc.nil? || an_rpc.call.nil?
 
       # allow the metadata to be accessed from the call
-      handle_call_tag = Object.new
       an_rpc.call.metadata = an_rpc.metadata  # attaches md to call for handlers
       GRPC.logger.debug("call md is #{an_rpc.metadata}")
       connect_md = nil
       unless @connect_md_proc.nil?
         connect_md = @connect_md_proc.call(an_rpc.method, an_rpc.metadata)
       end
-      an_rpc.call.run_batch(an_rpc.cq, handle_call_tag, INFINITE_FUTURE,
-                            SEND_INITIAL_METADATA => connect_md)
 
       return nil unless available?(an_rpc)
       return nil unless implemented?(an_rpc)
 
-      # Create the ActiveCall
+      # Create the ActiveCall. Indicate that metadata hasnt been sent yet.
       GRPC.logger.info("deadline is #{an_rpc.deadline}; (now=#{Time.now})")
       rpc_desc = rpc_descs[an_rpc.method.to_sym]
-      c = ActiveCall.new(an_rpc.call, an_rpc.cq,
-                         rpc_desc.marshal_proc, rpc_desc.unmarshal_proc(:input),
-                         an_rpc.deadline)
+      c = ActiveCall.new(an_rpc.call,
+                         rpc_desc.marshal_proc,
+                         rpc_desc.unmarshal_proc(:input),
+                         an_rpc.deadline,
+                         metadata_received: true,
+                         started: false,
+                         metadata_to_send: connect_md)
       mth = an_rpc.method.to_sym
       [c, mth]
     end
diff --git a/src/ruby/lib/grpc/generic/service.rb b/src/ruby/lib/grpc/generic/service.rb
index f30242ee80..7cb9f1cc99 100644
--- a/src/ruby/lib/grpc/generic/service.rb
+++ b/src/ruby/lib/grpc/generic/service.rb
@@ -168,7 +168,7 @@ module GRPC
           # @param kw [KeywordArgs] the channel arguments, plus any optional
           #                         args for configuring the client's channel
           def initialize(host, creds, **kw)
-            super(host, Core::CompletionQueue.new, creds, **kw)
+            super(host, creds, **kw)
           end
 
           # Used define_method to add a method for each rpc_desc.  Each method
diff --git a/src/ruby/lib/grpc/version.rb b/src/ruby/lib/grpc/version.rb
index 01c8c5ac8f..6e62af94d4 100644
--- a/src/ruby/lib/grpc/version.rb
+++ b/src/ruby/lib/grpc/version.rb
@@ -29,5 +29,5 @@
 
 # GRPC contains the General RPC module.
 module GRPC
-  VERSION = '0.15.0.dev'
+  VERSION = '1.1.0.dev'
 end
diff --git a/src/ruby/pb/grpc/health/checker.rb b/src/ruby/pb/grpc/health/checker.rb
index f7310d9289..4bce1744c4 100644
--- a/src/ruby/pb/grpc/health/checker.rb
+++ b/src/ruby/pb/grpc/health/checker.rb
@@ -28,7 +28,7 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 require 'grpc'
-require 'grpc/health/v1/health_services'
+require 'grpc/health/v1/health_services_pb'
 require 'thread'
 
 module Grpc
diff --git a/src/ruby/pb/grpc/health/v1/health.rb b/src/ruby/pb/grpc/health/v1/health_pb.rb
index aa87a93918..aa87a93918 100644
--- a/src/ruby/pb/grpc/health/v1/health.rb
+++ b/src/ruby/pb/grpc/health/v1/health_pb.rb
diff --git a/src/ruby/pb/grpc/health/v1/health_services.rb b/src/ruby/pb/grpc/health/v1/health_services_pb.rb
index 68a3956f54..8cc01e91dc 100644
--- a/src/ruby/pb/grpc/health/v1/health_services.rb
+++ b/src/ruby/pb/grpc/health/v1/health_services_pb.rb
@@ -32,7 +32,7 @@
 #
 
 require 'grpc'
-require 'grpc/health/v1/health'
+require 'grpc/health/v1/health_pb'
 
 module Grpc
   module Health
diff --git a/src/ruby/pb/grpc/testing/duplicate/echo_duplicate_services.rb b/src/ruby/pb/grpc/testing/duplicate/echo_duplicate_services_pb.rb
index eb523ffa6f..e51c2f087a 100644
--- a/src/ruby/pb/grpc/testing/duplicate/echo_duplicate_services.rb
+++ b/src/ruby/pb/grpc/testing/duplicate/echo_duplicate_services_pb.rb
@@ -1,5 +1,5 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
-# Source: grpc/testing/duplicate/echo_duplicate.proto for package 'grpc.testing.duplicate'
+# Source: src/proto/grpc/testing/duplicate/echo_duplicate.proto for package 'grpc.testing.duplicate'
 # Original file comments:
 # Copyright 2015, Google Inc.
 # All rights reserved.
@@ -34,7 +34,7 @@
 #
 
 require 'grpc'
-require 'grpc/testing/duplicate/echo_duplicate'
+require 'src/proto/grpc/testing/duplicate/echo_duplicate_pb'
 
 module Grpc
   module Testing
diff --git a/src/ruby/pb/grpc/testing/metrics.rb b/src/ruby/pb/grpc/testing/metrics_pb.rb
index 3b3c8cd61b..77b6c90970 100644
--- a/src/ruby/pb/grpc/testing/metrics.rb
+++ b/src/ruby/pb/grpc/testing/metrics_pb.rb
@@ -1,5 +1,5 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
-# source: grpc/testing/metrics.proto
+# source: src/proto/grpc/testing/metrics.proto
 
 require 'google/protobuf'
 
diff --git a/src/ruby/pb/grpc/testing/metrics_services.rb b/src/ruby/pb/grpc/testing/metrics_services_pb.rb
index 467b7b3ee5..e46366b1fb 100644
--- a/src/ruby/pb/grpc/testing/metrics_services.rb
+++ b/src/ruby/pb/grpc/testing/metrics_services_pb.rb
@@ -1,5 +1,5 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
-# Source: grpc/testing/metrics.proto for package 'grpc.testing'
+# Source: src/proto/grpc/testing/metrics.proto for package 'grpc.testing'
 # Original file comments:
 # Copyright 2015-2016, Google Inc.
 # All rights reserved.
@@ -38,7 +38,7 @@
 # service.
 
 require 'grpc'
-require 'grpc/testing/metrics'
+require 'src/proto/grpc/testing/metrics_pb'
 
 module Grpc
   module Testing
diff --git a/src/ruby/pb/src/proto/grpc/testing/empty.rb b/src/ruby/pb/src/proto/grpc/testing/empty_pb.rb
index 9c2568d605..9c2568d605 100644
--- a/src/ruby/pb/src/proto/grpc/testing/empty.rb
+++ b/src/ruby/pb/src/proto/grpc/testing/empty_pb.rb
diff --git a/src/ruby/pb/src/proto/grpc/testing/messages.rb b/src/ruby/pb/src/proto/grpc/testing/messages_pb.rb
index 2bdfe0eade..e27ccd0dc0 100644
--- a/src/ruby/pb/src/proto/grpc/testing/messages.rb
+++ b/src/ruby/pb/src/proto/grpc/testing/messages_pb.rb
@@ -4,6 +4,9 @@
 require 'google/protobuf'
 
 Google::Protobuf::DescriptorPool.generated_pool.build do
+  add_message "grpc.testing.BoolValue" do
+    optional :value, :bool, 1
+  end
   add_message "grpc.testing.Payload" do
     optional :type, :enum, 1, "grpc.testing.PayloadType"
     optional :body, :bytes, 2
@@ -18,8 +21,9 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
     optional :payload, :message, 3, "grpc.testing.Payload"
     optional :fill_username, :bool, 4
     optional :fill_oauth_scope, :bool, 5
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
+    optional :response_compressed, :message, 6, "grpc.testing.BoolValue"
     optional :response_status, :message, 7, "grpc.testing.EchoStatus"
+    optional :expect_compressed, :message, 8, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.SimpleResponse" do
     optional :payload, :message, 1, "grpc.testing.Payload"
@@ -28,6 +32,7 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   end
   add_message "grpc.testing.StreamingInputCallRequest" do
     optional :payload, :message, 1, "grpc.testing.Payload"
+    optional :expect_compressed, :message, 2, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.StreamingInputCallResponse" do
     optional :aggregated_payload_size, :int32, 1
@@ -35,12 +40,12 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   add_message "grpc.testing.ResponseParameters" do
     optional :size, :int32, 1
     optional :interval_us, :int32, 2
+    optional :compressed, :message, 3, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.StreamingOutputCallRequest" do
     optional :response_type, :enum, 1, "grpc.testing.PayloadType"
     repeated :response_parameters, :message, 2, "grpc.testing.ResponseParameters"
     optional :payload, :message, 3, "grpc.testing.Payload"
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
     optional :response_status, :message, 7, "grpc.testing.EchoStatus"
   end
   add_message "grpc.testing.StreamingOutputCallResponse" do
@@ -55,18 +60,12 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   end
   add_enum "grpc.testing.PayloadType" do
     value :COMPRESSABLE, 0
-    value :UNCOMPRESSABLE, 1
-    value :RANDOM, 2
-  end
-  add_enum "grpc.testing.CompressionType" do
-    value :NONE, 0
-    value :GZIP, 1
-    value :DEFLATE, 2
   end
 end
 
 module Grpc
   module Testing
+    BoolValue = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.BoolValue").msgclass
     Payload = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.Payload").msgclass
     EchoStatus = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.EchoStatus").msgclass
     SimpleRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.SimpleRequest").msgclass
@@ -79,6 +78,5 @@ module Grpc
     ReconnectParams = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ReconnectParams").msgclass
     ReconnectInfo = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ReconnectInfo").msgclass
     PayloadType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.PayloadType").enummodule
-    CompressionType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.CompressionType").enummodule
   end
 end
diff --git a/src/ruby/pb/src/proto/grpc/testing/test.rb b/src/ruby/pb/src/proto/grpc/testing/test_pb.rb
index 245b5ce00c..2cc9863031 100644
--- a/src/ruby/pb/src/proto/grpc/testing/test.rb
+++ b/src/ruby/pb/src/proto/grpc/testing/test_pb.rb
@@ -3,8 +3,8 @@
 
 require 'google/protobuf'
 
-require 'src/proto/grpc/testing/empty'
-require 'src/proto/grpc/testing/messages'
+require 'src/proto/grpc/testing/empty_pb'
+require 'src/proto/grpc/testing/messages_pb'
 Google::Protobuf::DescriptorPool.generated_pool.build do
 end
 
diff --git a/src/ruby/pb/src/proto/grpc/testing/test_services.rb b/src/ruby/pb/src/proto/grpc/testing/test_services_pb.rb
index 2652de5e6d..fde328e4c5 100644
--- a/src/ruby/pb/src/proto/grpc/testing/test_services.rb
+++ b/src/ruby/pb/src/proto/grpc/testing/test_services_pb.rb
@@ -35,7 +35,7 @@
 #
 
 require 'grpc'
-require 'src/proto/grpc/testing/test'
+require 'src/proto/grpc/testing/test_pb'
 
 module Grpc
   module Testing
diff --git a/src/ruby/pb/test/client.rb b/src/ruby/pb/test/client.rb
index b6695482a2..1e3ae65630 100755
--- a/src/ruby/pb/test/client.rb
+++ b/src/ruby/pb/test/client.rb
@@ -52,9 +52,9 @@ require_relative '../../lib/grpc'
 require 'googleauth'
 require 'google/protobuf'
 
-require_relative 'proto/empty'
-require_relative 'proto/messages'
-require_relative 'proto/test_services'
+require_relative '../src/proto/grpc/testing/empty_pb'
+require_relative '../src/proto/grpc/testing/messages_pb'
+require_relative '../src/proto/grpc/testing/test_services_pb'
 
 AUTH_ENV = Google::Auth::CredentialsLoader::ENV_VAR
 
@@ -111,6 +111,18 @@ end
 # creates a test stub that accesses host:port securely.
 def create_stub(opts)
   address = "#{opts.host}:#{opts.port}"
+
+  # Provide channel args that request compression by default
+  # for compression interop tests
+  if ['client_compressed_unary',
+      'client_compressed_streaming'].include?(opts.test_case)
+    compression_options =
+      GRPC::Core::CompressionOptions.new(default_algorithm: :gzip)
+    compression_channel_args = compression_options.to_channel_arg_hash
+  else
+    compression_channel_args = {}
+  end
+
   if opts.secure
     creds = ssl_creds(opts.use_test_ca)
     stub_opts = {
@@ -145,10 +157,15 @@ def create_stub(opts)
     end
 
     GRPC.logger.info("... connecting securely to #{address}")
+    stub_opts[:channel_args].merge!(compression_channel_args)
     Grpc::Testing::TestService::Stub.new(address, creds, **stub_opts)
   else
     GRPC.logger.info("... connecting insecurely to #{address}")
-    Grpc::Testing::TestService::Stub.new(address, :this_channel_is_insecure)
+    Grpc::Testing::TestService::Stub.new(
+      address,
+      :this_channel_is_insecure,
+      channel_args: compression_channel_args
+    )
   end
 end
 
@@ -197,10 +214,47 @@ class PingPongPlayer
   end
 end
 
+class BlockingEnumerator
+  include Grpc::Testing
+  include Grpc::Testing::PayloadType
+
+  def initialize(req_size, sleep_time)
+    @req_size = req_size
+    @sleep_time = sleep_time
+  end
+
+  def each_item
+    return enum_for(:each_item) unless block_given?
+    req_cls = StreamingOutputCallRequest
+    req = req_cls.new(payload: Payload.new(body: nulls(@req_size)))
+    yield req
+    # Sleep until after the deadline should have passed
+    sleep(@sleep_time)
+  end
+end
+
+# Intended to be used to wrap a call_op, and to adjust
+# the write flag of the call_op in between messages yielded to it.
+class WriteFlagSettingStreamingInputEnumerable
+  attr_accessor :call_op
+
+  def initialize(requests_and_write_flags)
+    @requests_and_write_flags = requests_and_write_flags
+  end
+
+  def each
+    @requests_and_write_flags.each do |request_and_flag|
+      @call_op.write_flag = request_and_flag[:write_flag]
+      yield request_and_flag[:request]
+    end
+  end
+end
+
 # defines methods corresponding to each interop test case.
 class NamedTests
   include Grpc::Testing
   include Grpc::Testing::PayloadType
+  include GRPC::Core::MetadataKeys
 
   def initialize(stub, args)
     @stub = stub
@@ -216,6 +270,48 @@ class NamedTests
     perform_large_unary
   end
 
+  def client_compressed_unary
+    # first request used also for the probe
+    req_size, wanted_response_size = 271_828, 314_159
+    expect_compressed = BoolValue.new(value: true)
+    payload = Payload.new(type: :COMPRESSABLE, body: nulls(req_size))
+    req = SimpleRequest.new(response_type: :COMPRESSABLE,
+                            response_size: wanted_response_size,
+                            payload: payload,
+                            expect_compressed: expect_compressed)
+
+    # send a probe to see if CompressedResponse is supported on the server
+    send_probe_for_compressed_request_support do
+      request_uncompressed_args = {
+        COMPRESSION_REQUEST_ALGORITHM => 'identity'
+      }
+      @stub.unary_call(req, metadata: request_uncompressed_args)
+    end
+
+    # make a call with a compressed message
+    resp = @stub.unary_call(req)
+    assert('Expected second unary call with compression to work') do
+      resp.payload.body.length == wanted_response_size
+    end
+
+    # make a call with an uncompressed message
+    stub_options = {
+      COMPRESSION_REQUEST_ALGORITHM => 'identity'
+    }
+
+    req = SimpleRequest.new(
+      response_type: :COMPRESSABLE,
+      response_size: wanted_response_size,
+      payload: payload,
+      expect_compressed: BoolValue.new(value: false)
+    )
+
+    resp = @stub.unary_call(req, metadata: stub_options)
+    assert('Expected second unary call with compression to work') do
+      resp.payload.body.length == wanted_response_size
+    end
+  end
+
   def service_account_creds
     # ignore this test if the oauth options are not set
     if @args.oauth_scope.nil?
@@ -290,6 +386,50 @@ class NamedTests
     end
   end
 
+  def client_compressed_streaming
+    # first request used also by the probe
+    first_request = StreamingInputCallRequest.new(
+      payload: Payload.new(type: :COMPRESSABLE, body: nulls(27_182)),
+      expect_compressed: BoolValue.new(value: true)
+    )
+
+    # send a probe to see if CompressedResponse is supported on the server
+    send_probe_for_compressed_request_support do
+      request_uncompressed_args = {
+        COMPRESSION_REQUEST_ALGORITHM => 'identity'
+      }
+      @stub.streaming_input_call([first_request],
+                                 metadata: request_uncompressed_args)
+    end
+
+    second_request = StreamingInputCallRequest.new(
+      payload: Payload.new(type: :COMPRESSABLE, body: nulls(45_904)),
+      expect_compressed: BoolValue.new(value: false)
+    )
+
+    # Create the requests messages and the corresponding write flags
+    # for each message
+    requests = WriteFlagSettingStreamingInputEnumerable.new([
+      { request: first_request,
+        write_flag: 0 },
+      { request: second_request,
+        write_flag: GRPC::Core::WriteFlags::NO_COMPRESS }
+    ])
+
+    # Create the call_op, pass it to the requests enumerable, and
+    # run the call
+    call_op = @stub.streaming_input_call(requests,
+                                         return_op: true)
+    requests.call_op = call_op
+    resp = call_op.execute
+
+    wanted_aggregate_size = 73_086
+
+    assert("#{__callee__}: aggregate payload size is incorrect") do
+      wanted_aggregate_size == resp.aggregated_payload_size
+    end
+  end
+
   def server_streaming
     msg_sizes = [31_415, 9, 2653, 58_979]
     response_spec = msg_sizes.map { |s| ResponseParameters.new(size: s) }
@@ -315,11 +455,10 @@ class NamedTests
   end
 
   def timeout_on_sleeping_server
-    msg_sizes = [[27_182, 31_415]]
-    ppp = PingPongPlayer.new(msg_sizes)
-    deadline = GRPC::Core::TimeConsts::from_relative_time(0.001)
-    resps = @stub.full_duplex_call(ppp.each_item, deadline: deadline)
-    resps.each { |r| ppp.queue.push(r) }
+    enum = BlockingEnumerator.new(27_182, 2)
+    deadline = GRPC::Core::TimeConsts::from_relative_time(1)
+    resps = @stub.full_duplex_call(enum.each_item, deadline: deadline)
+    resps.each { } # wait to receive each request (or timeout)
     fail 'Should have raised GRPC::BadStatus(DEADLINE_EXCEEDED)'
   rescue GRPC::BadStatus => e
     assert("#{__callee__}: status was wrong") do
@@ -351,7 +490,7 @@ class NamedTests
     op.execute
     fail 'Should have raised GRPC:Cancelled'
   rescue GRPC::Cancelled
-    assert("#{__callee__}: call operation should be CANCELLED") { op.cancelled }
+    assert("#{__callee__}: call operation should be CANCELLED") { op.cancelled? }
   end
 
   def cancel_after_first_response
@@ -362,7 +501,7 @@ class NamedTests
     op.execute.each { |r| ppp.queue.push(r) }
     fail 'Should have raised GRPC:Cancelled'
   rescue GRPC::Cancelled
-    assert("#{__callee__}: call operation should be CANCELLED") { op.cancelled }
+    assert("#{__callee__}: call operation should be CANCELLED") { op.cancelled? }
     op.wait
   end
 
@@ -397,6 +536,29 @@ class NamedTests
     end
     resp
   end
+
+  # Send probing message for compressed request on the server, to see
+  # if it's implemented.
+  def send_probe_for_compressed_request_support(&send_probe)
+    bad_status_occured = false
+
+    begin
+      send_probe.call
+    rescue GRPC::BadStatus => e
+      if e.code == GRPC::Core::StatusCodes::INVALID_ARGUMENT
+        bad_status_occured = true
+      else
+        fail AssertionError, "Bad status received but code is #{e.code}"
+      end
+    rescue Exception => e
+      fail AssertionError, "Expected BadStatus. Received: #{e.inspect}"
+    end
+
+    assert('CompressedRequest probe failed') do
+      bad_status_occured
+    end
+  end
+
 end
 
 # Args is used to hold the command line info.
diff --git a/src/ruby/pb/test/proto/empty.rb b/src/ruby/pb/test/proto/empty.rb
deleted file mode 100644
index 559adcc85e..0000000000
--- a/src/ruby/pb/test/proto/empty.rb
+++ /dev/null
@@ -1,15 +0,0 @@
-# Generated by the protocol buffer compiler.  DO NOT EDIT!
-# source: test/proto/empty.proto
-
-require 'google/protobuf'
-
-Google::Protobuf::DescriptorPool.generated_pool.build do
-  add_message "grpc.testing.Empty" do
-  end
-end
-
-module Grpc
-  module Testing
-    Empty = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.Empty").msgclass
-  end
-end
diff --git a/src/ruby/pb/test/proto/messages.rb b/src/ruby/pb/test/proto/messages.rb
deleted file mode 100644
index 5222c9824a..0000000000
--- a/src/ruby/pb/test/proto/messages.rb
+++ /dev/null
@@ -1,80 +0,0 @@
-# Generated by the protocol buffer compiler.  DO NOT EDIT!
-# source: test/proto/messages.proto
-
-require 'google/protobuf'
-
-Google::Protobuf::DescriptorPool.generated_pool.build do
-  add_message "grpc.testing.Payload" do
-    optional :type, :enum, 1, "grpc.testing.PayloadType"
-    optional :body, :bytes, 2
-  end
-  add_message "grpc.testing.EchoStatus" do
-    optional :code, :int32, 1
-    optional :message, :string, 2
-  end
-  add_message "grpc.testing.SimpleRequest" do
-    optional :response_type, :enum, 1, "grpc.testing.PayloadType"
-    optional :response_size, :int32, 2
-    optional :payload, :message, 3, "grpc.testing.Payload"
-    optional :fill_username, :bool, 4
-    optional :fill_oauth_scope, :bool, 5
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
-    optional :response_status, :message, 7, "grpc.testing.EchoStatus"
-  end
-  add_message "grpc.testing.SimpleResponse" do
-    optional :payload, :message, 1, "grpc.testing.Payload"
-    optional :username, :string, 2
-    optional :oauth_scope, :string, 3
-  end
-  add_message "grpc.testing.StreamingInputCallRequest" do
-    optional :payload, :message, 1, "grpc.testing.Payload"
-  end
-  add_message "grpc.testing.StreamingInputCallResponse" do
-    optional :aggregated_payload_size, :int32, 1
-  end
-  add_message "grpc.testing.ResponseParameters" do
-    optional :size, :int32, 1
-    optional :interval_us, :int32, 2
-  end
-  add_message "grpc.testing.StreamingOutputCallRequest" do
-    optional :response_type, :enum, 1, "grpc.testing.PayloadType"
-    repeated :response_parameters, :message, 2, "grpc.testing.ResponseParameters"
-    optional :payload, :message, 3, "grpc.testing.Payload"
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
-    optional :response_status, :message, 7, "grpc.testing.EchoStatus"
-  end
-  add_message "grpc.testing.StreamingOutputCallResponse" do
-    optional :payload, :message, 1, "grpc.testing.Payload"
-  end
-  add_message "grpc.testing.ReconnectInfo" do
-    optional :passed, :bool, 1
-    repeated :backoff_ms, :int32, 2
-  end
-  add_enum "grpc.testing.PayloadType" do
-    value :COMPRESSABLE, 0
-    value :UNCOMPRESSABLE, 1
-    value :RANDOM, 2
-  end
-  add_enum "grpc.testing.CompressionType" do
-    value :NONE, 0
-    value :GZIP, 1
-    value :DEFLATE, 2
-  end
-end
-
-module Grpc
-  module Testing
-    Payload = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.Payload").msgclass
-    EchoStatus = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.EchoStatus").msgclass
-    SimpleRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.SimpleRequest").msgclass
-    SimpleResponse = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.SimpleResponse").msgclass
-    StreamingInputCallRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.StreamingInputCallRequest").msgclass
-    StreamingInputCallResponse = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.StreamingInputCallResponse").msgclass
-    ResponseParameters = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ResponseParameters").msgclass
-    StreamingOutputCallRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.StreamingOutputCallRequest").msgclass
-    StreamingOutputCallResponse = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.StreamingOutputCallResponse").msgclass
-    ReconnectInfo = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ReconnectInfo").msgclass
-    PayloadType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.PayloadType").enummodule
-    CompressionType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.CompressionType").enummodule
-  end
-end
diff --git a/src/ruby/pb/test/proto/test.rb b/src/ruby/pb/test/proto/test.rb
deleted file mode 100644
index 100eb6505c..0000000000
--- a/src/ruby/pb/test/proto/test.rb
+++ /dev/null
@@ -1,14 +0,0 @@
-# Generated by the protocol buffer compiler.  DO NOT EDIT!
-# source: test/proto/test.proto
-
-require 'google/protobuf'
-
-require 'test/proto/empty'
-require 'test/proto/messages'
-Google::Protobuf::DescriptorPool.generated_pool.build do
-end
-
-module Grpc
-  module Testing
-  end
-end
diff --git a/src/ruby/pb/test/proto/test_services.rb b/src/ruby/pb/test/proto/test_services.rb
deleted file mode 100644
index 9df9cc5860..0000000000
--- a/src/ruby/pb/test/proto/test_services.rb
+++ /dev/null
@@ -1,64 +0,0 @@
-# Generated by the protocol buffer compiler.  DO NOT EDIT!
-# Source: test/proto/test.proto for package 'grpc.testing'
-
-require 'grpc'
-require 'test/proto/test'
-
-module Grpc
-  module Testing
-    module TestService
-
-      # TODO: add proto service documentation here
-      class Service
-
-        include GRPC::GenericService
-
-        self.marshal_class_method = :encode
-        self.unmarshal_class_method = :decode
-        self.service_name = 'grpc.testing.TestService'
-
-        rpc :EmptyCall, Empty, Empty
-        rpc :UnaryCall, SimpleRequest, SimpleResponse
-        rpc :StreamingOutputCall, StreamingOutputCallRequest, stream(StreamingOutputCallResponse)
-        rpc :StreamingInputCall, stream(StreamingInputCallRequest), StreamingInputCallResponse
-        rpc :FullDuplexCall, stream(StreamingOutputCallRequest), stream(StreamingOutputCallResponse)
-        rpc :HalfDuplexCall, stream(StreamingOutputCallRequest), stream(StreamingOutputCallResponse)
-      end
-
-      Stub = Service.rpc_stub_class
-    end
-    module UnimplementedService
-
-      # TODO: add proto service documentation here
-      class Service
-
-        include GRPC::GenericService
-
-        self.marshal_class_method = :encode
-        self.unmarshal_class_method = :decode
-        self.service_name = 'grpc.testing.UnimplementedService'
-
-        rpc :UnimplementedCall, Empty, Empty
-      end
-
-      Stub = Service.rpc_stub_class
-    end
-    module ReconnectService
-
-      # TODO: add proto service documentation here
-      class Service
-
-        include GRPC::GenericService
-
-        self.marshal_class_method = :encode
-        self.unmarshal_class_method = :decode
-        self.service_name = 'grpc.testing.ReconnectService'
-
-        rpc :Start, Empty, Empty
-        rpc :Stop, Empty, ReconnectInfo
-      end
-
-      Stub = Service.rpc_stub_class
-    end
-  end
-end
diff --git a/src/ruby/pb/test/server.rb b/src/ruby/pb/test/server.rb
index 914c7cc79d..0808121661 100755
--- a/src/ruby/pb/test/server.rb
+++ b/src/ruby/pb/test/server.rb
@@ -50,9 +50,9 @@ require 'optparse'
 
 require 'grpc'
 
-require 'test/proto/empty'
-require 'test/proto/messages'
-require 'test/proto/test_services'
+require_relative '../src/proto/grpc/testing/empty_pb'
+require_relative '../src/proto/grpc/testing/messages_pb'
+require_relative '../src/proto/grpc/testing/test_services_pb'
 
 # DebugIsTruncated extends the default Logger to truncate debug messages
 class DebugIsTruncated < Logger
@@ -188,11 +188,13 @@ class TestTarget < Grpc::Testing::TestService::Service
       begin
         GRPC.logger.info('interop-server: started receiving')
         reqs.each do |req|
-          resp_size = req.response_parameters[0].size
-          GRPC.logger.info("read a req, response size is #{resp_size}")
-          resp = cls.new(payload: Payload.new(type: req.response_type,
-                                              body: nulls(resp_size)))
-          q.push(resp)
+          req.response_parameters.each do |params|
+            resp_size = params.size
+            GRPC.logger.info("read a req, response size is #{resp_size}")
+            resp = cls.new(payload: Payload.new(type: req.response_type,
+                                                body: nulls(resp_size)))
+            q.push(resp)
+          end
         end
         GRPC.logger.info('interop-server: finished receiving')
         q.push(self)
diff --git a/src/ruby/qps/client.rb b/src/ruby/qps/client.rb
index 917b01271e..7ed648acef 100644
--- a/src/ruby/qps/client.rb
+++ b/src/ruby/qps/client.rb
@@ -38,7 +38,7 @@ $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 
 require 'grpc'
 require 'histogram'
-require 'src/proto/grpc/testing/services_services'
+require 'src/proto/grpc/testing/services_services_pb'
 
 class Poisson
   def interarrival
diff --git a/src/ruby/qps/server.rb b/src/ruby/qps/server.rb
index 52a89ce847..cd98ee1fd9 100644
--- a/src/ruby/qps/server.rb
+++ b/src/ruby/qps/server.rb
@@ -38,9 +38,9 @@ $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 
 require 'grpc'
 require 'qps-common'
-require 'src/proto/grpc/testing/messages'
-require 'src/proto/grpc/testing/services_services'
-require 'src/proto/grpc/testing/stats'
+require 'src/proto/grpc/testing/messages_pb'
+require 'src/proto/grpc/testing/services_services_pb'
+require 'src/proto/grpc/testing/stats_pb'
 
 class BenchmarkServiceImpl < Grpc::Testing::BenchmarkService::Service
   def unary_call(req, _call)
diff --git a/src/ruby/qps/src/proto/grpc/testing/control.rb b/src/ruby/qps/src/proto/grpc/testing/control_pb.rb
index 958fca320b..02207a2b5d 100644
--- a/src/ruby/qps/src/proto/grpc/testing/control.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/control_pb.rb
@@ -3,8 +3,8 @@
 
 require 'google/protobuf'
 
-require 'src/proto/grpc/testing/payloads'
-require 'src/proto/grpc/testing/stats'
+require 'src/proto/grpc/testing/payloads_pb'
+require 'src/proto/grpc/testing/stats_pb'
 Google::Protobuf::DescriptorPool.generated_pool.build do
   add_message "grpc.testing.PoissonParams" do
     optional :offered_load, :double, 1
@@ -109,6 +109,8 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
     repeated :server_stats, :message, 4, "grpc.testing.ServerStats"
     repeated :server_cores, :int32, 5
     optional :summary, :message, 6, "grpc.testing.ScenarioResultSummary"
+    repeated :client_success, :bool, 7
+    repeated :server_success, :bool, 8
   end
   add_enum "grpc.testing.ClientType" do
     value :SYNC_CLIENT, 0
diff --git a/src/ruby/qps/src/proto/grpc/testing/messages.rb b/src/ruby/qps/src/proto/grpc/testing/messages_pb.rb
index 2bdfe0eade..e27ccd0dc0 100644
--- a/src/ruby/qps/src/proto/grpc/testing/messages.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/messages_pb.rb
@@ -4,6 +4,9 @@
 require 'google/protobuf'
 
 Google::Protobuf::DescriptorPool.generated_pool.build do
+  add_message "grpc.testing.BoolValue" do
+    optional :value, :bool, 1
+  end
   add_message "grpc.testing.Payload" do
     optional :type, :enum, 1, "grpc.testing.PayloadType"
     optional :body, :bytes, 2
@@ -18,8 +21,9 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
     optional :payload, :message, 3, "grpc.testing.Payload"
     optional :fill_username, :bool, 4
     optional :fill_oauth_scope, :bool, 5
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
+    optional :response_compressed, :message, 6, "grpc.testing.BoolValue"
     optional :response_status, :message, 7, "grpc.testing.EchoStatus"
+    optional :expect_compressed, :message, 8, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.SimpleResponse" do
     optional :payload, :message, 1, "grpc.testing.Payload"
@@ -28,6 +32,7 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   end
   add_message "grpc.testing.StreamingInputCallRequest" do
     optional :payload, :message, 1, "grpc.testing.Payload"
+    optional :expect_compressed, :message, 2, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.StreamingInputCallResponse" do
     optional :aggregated_payload_size, :int32, 1
@@ -35,12 +40,12 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   add_message "grpc.testing.ResponseParameters" do
     optional :size, :int32, 1
     optional :interval_us, :int32, 2
+    optional :compressed, :message, 3, "grpc.testing.BoolValue"
   end
   add_message "grpc.testing.StreamingOutputCallRequest" do
     optional :response_type, :enum, 1, "grpc.testing.PayloadType"
     repeated :response_parameters, :message, 2, "grpc.testing.ResponseParameters"
     optional :payload, :message, 3, "grpc.testing.Payload"
-    optional :response_compression, :enum, 6, "grpc.testing.CompressionType"
     optional :response_status, :message, 7, "grpc.testing.EchoStatus"
   end
   add_message "grpc.testing.StreamingOutputCallResponse" do
@@ -55,18 +60,12 @@ Google::Protobuf::DescriptorPool.generated_pool.build do
   end
   add_enum "grpc.testing.PayloadType" do
     value :COMPRESSABLE, 0
-    value :UNCOMPRESSABLE, 1
-    value :RANDOM, 2
-  end
-  add_enum "grpc.testing.CompressionType" do
-    value :NONE, 0
-    value :GZIP, 1
-    value :DEFLATE, 2
   end
 end
 
 module Grpc
   module Testing
+    BoolValue = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.BoolValue").msgclass
     Payload = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.Payload").msgclass
     EchoStatus = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.EchoStatus").msgclass
     SimpleRequest = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.SimpleRequest").msgclass
@@ -79,6 +78,5 @@ module Grpc
     ReconnectParams = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ReconnectParams").msgclass
     ReconnectInfo = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.ReconnectInfo").msgclass
     PayloadType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.PayloadType").enummodule
-    CompressionType = Google::Protobuf::DescriptorPool.generated_pool.lookup("grpc.testing.CompressionType").enummodule
   end
 end
diff --git a/src/ruby/qps/src/proto/grpc/testing/payloads.rb b/src/ruby/qps/src/proto/grpc/testing/payloads_pb.rb
index ae8855f685..ae8855f685 100644
--- a/src/ruby/qps/src/proto/grpc/testing/payloads.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/payloads_pb.rb
diff --git a/src/ruby/qps/src/proto/grpc/testing/services.rb b/src/ruby/qps/src/proto/grpc/testing/services_pb.rb
index b2675c2afe..5ce13bf8b0 100644
--- a/src/ruby/qps/src/proto/grpc/testing/services.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/services_pb.rb
@@ -3,8 +3,8 @@
 
 require 'google/protobuf'
 
-require 'src/proto/grpc/testing/messages'
-require 'src/proto/grpc/testing/control'
+require 'src/proto/grpc/testing/messages_pb'
+require 'src/proto/grpc/testing/control_pb'
 Google::Protobuf::DescriptorPool.generated_pool.build do
 end
 
diff --git a/src/ruby/qps/src/proto/grpc/testing/services_services.rb b/src/ruby/qps/src/proto/grpc/testing/services_services_pb.rb
index 94b9a1e164..bdbb9c86d0 100644
--- a/src/ruby/qps/src/proto/grpc/testing/services_services.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/services_services_pb.rb
@@ -34,7 +34,7 @@
 # of unary/streaming requests/responses.
 
 require 'grpc'
-require 'src/proto/grpc/testing/services'
+require 'src/proto/grpc/testing/services_pb'
 
 module Grpc
   module Testing
diff --git a/src/ruby/qps/src/proto/grpc/testing/stats.rb b/src/ruby/qps/src/proto/grpc/testing/stats_pb.rb
index 41f75bedf0..41f75bedf0 100644
--- a/src/ruby/qps/src/proto/grpc/testing/stats.rb
+++ b/src/ruby/qps/src/proto/grpc/testing/stats_pb.rb
diff --git a/src/ruby/qps/worker.rb b/src/ruby/qps/worker.rb
index 665fb86352..12b8087ca0 100755
--- a/src/ruby/qps/worker.rb
+++ b/src/ruby/qps/worker.rb
@@ -44,7 +44,7 @@ require 'facter'
 require 'client'
 require 'qps-common'
 require 'server'
-require 'src/proto/grpc/testing/services_services'
+require 'src/proto/grpc/testing/services_services_pb'
 
 class WorkerServiceImpl < Grpc::Testing::WorkerService::Service
   def cpu_cores
diff --git a/src/ruby/spec/call_spec.rb b/src/ruby/spec/call_spec.rb
index ae3ce0748a..1c44b333de 100644
--- a/src/ruby/spec/call_spec.rb
+++ b/src/ruby/spec/call_spec.rb
@@ -96,7 +96,6 @@ describe GRPC::Core::CallOps do
 end
 
 describe GRPC::Core::Call do
-  let(:client_queue) { GRPC::Core::CompletionQueue.new }
   let(:test_tag)  { Object.new }
   let(:fake_host) { 'localhost:10101' }
 
@@ -154,7 +153,7 @@ describe GRPC::Core::Call do
   end
 
   def make_test_call
-    @ch.create_call(client_queue, nil, nil, 'dummy_method', nil, deadline)
+    @ch.create_call(nil, nil, 'dummy_method', nil, deadline)
   end
 
   def deadline
diff --git a/src/ruby/spec/channel_spec.rb b/src/ruby/spec/channel_spec.rb
index 355f95c9d7..740eac631a 100644
--- a/src/ruby/spec/channel_spec.rb
+++ b/src/ruby/spec/channel_spec.rb
@@ -37,7 +37,6 @@ end
 
 describe GRPC::Core::Channel do
   let(:fake_host) { 'localhost:0' }
-  let(:cq) { GRPC::Core::CompletionQueue.new }
 
   def create_test_cert
     GRPC::Core::ChannelCredentials.new(load_test_certs[0])
@@ -122,7 +121,7 @@ describe GRPC::Core::Channel do
       deadline = Time.now + 5
 
       blk = proc do
-        ch.create_call(cq, nil, nil, 'dummy_method', nil, deadline)
+        ch.create_call(nil, nil, 'dummy_method', nil, deadline)
       end
       expect(&blk).to_not raise_error
     end
@@ -133,7 +132,7 @@ describe GRPC::Core::Channel do
 
       deadline = Time.now + 5
       blk = proc do
-        ch.create_call(cq, nil, nil, 'dummy_method', nil, deadline)
+        ch.create_call(nil, nil, 'dummy_method', nil, deadline)
       end
       expect(&blk).to raise_error(RuntimeError)
     end
diff --git a/src/ruby/spec/client_server_spec.rb b/src/ruby/spec/client_server_spec.rb
index aedeca272d..d9df0b9ae2 100644
--- a/src/ruby/spec/client_server_spec.rb
+++ b/src/ruby/spec/client_server_spec.rb
@@ -34,27 +34,23 @@ include GRPC::Core
 shared_context 'setup: tags' do
   let(:sent_message) { 'sent message' }
   let(:reply_text) { 'the reply' }
-  before(:example) do
-    @client_tag = Object.new
-    @server_tag = Object.new
-  end
 
   def deadline
     Time.now + 5
   end
 
-  def server_allows_client_to_proceed
-    recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+  def server_allows_client_to_proceed(metadata = {})
+    recvd_rpc = @server.request_call
     expect(recvd_rpc).to_not eq nil
     server_call = recvd_rpc.call
-    ops = { CallOps::SEND_INITIAL_METADATA => {} }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline, ops)
+    ops = { CallOps::SEND_INITIAL_METADATA => metadata }
+    svr_batch = server_call.run_batch(ops)
     expect(svr_batch.send_metadata).to be true
     server_call
   end
 
   def new_client_call
-    @ch.create_call(@client_queue, nil, nil, '/method', nil, deadline)
+    @ch.create_call(nil, nil, '/method', nil, deadline)
   end
 end
 
@@ -91,8 +87,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::SEND_INITIAL_METADATA => {},
       CallOps::SEND_MESSAGE => sent_message
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
     expect(batch_result.send_message).to be true
 
@@ -101,8 +96,7 @@ shared_examples 'basic GRPC message delivery is OK' do
     server_ops = {
       CallOps::RECV_MESSAGE => nil
     }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline,
-                                      server_ops)
+    svr_batch = server_call.run_batch(server_ops)
     expect(svr_batch.message).to eq(sent_message)
   end
 
@@ -118,8 +112,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::SEND_INITIAL_METADATA => {},
       CallOps::SEND_MESSAGE => sent_message
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
     expect(batch_result.send_message).to be true
 
@@ -129,12 +122,50 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::RECV_MESSAGE => nil,
       CallOps::SEND_MESSAGE => reply_text
     }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline,
-                                      server_ops)
+    svr_batch = server_call.run_batch(server_ops)
     expect(svr_batch.message).to eq(sent_message)
     expect(svr_batch.send_message).to be true
   end
 
+  it 'compressed messages can be sent and received' do
+    call = new_client_call
+    server_call = nil
+    long_request_str = '0' * 2000
+    long_response_str = '1' * 2000
+    md = { 'grpc-internal-encoding-request' => 'gzip' }
+
+    server_thread = Thread.new do
+      server_call = server_allows_client_to_proceed(md)
+    end
+
+    client_ops = {
+      CallOps::SEND_INITIAL_METADATA => md,
+      CallOps::SEND_MESSAGE => long_request_str
+    }
+    batch_result = call.run_batch(client_ops)
+    expect(batch_result.send_metadata).to be true
+    expect(batch_result.send_message).to be true
+
+    # confirm the server can read the inbound message
+    server_thread.join
+    server_ops = {
+      CallOps::RECV_MESSAGE => nil,
+      CallOps::SEND_MESSAGE => long_response_str
+    }
+    svr_batch = server_call.run_batch(server_ops)
+    expect(svr_batch.message).to eq(long_request_str)
+    expect(svr_batch.send_message).to be true
+
+    client_ops = {
+      CallOps::SEND_CLOSE_FROM_CLIENT => nil,
+      CallOps::RECV_INITIAL_METADATA => nil,
+      CallOps::RECV_MESSAGE => nil
+    }
+    batch_result = call.run_batch(client_ops)
+    expect(batch_result.send_close).to be true
+    expect(batch_result.message).to eq long_response_str
+  end
+
   it 'servers can ignore a client write and send a status' do
     call = new_client_call
     server_call = nil
@@ -147,8 +178,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::SEND_INITIAL_METADATA => {},
       CallOps::SEND_MESSAGE => sent_message
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
     expect(batch_result.send_message).to be true
 
@@ -158,8 +188,7 @@ shared_examples 'basic GRPC message delivery is OK' do
     server_ops = {
       CallOps::SEND_STATUS_FROM_SERVER => the_status
     }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline,
-                                      server_ops)
+    svr_batch = server_call.run_batch(server_ops)
     expect(svr_batch.message).to eq nil
     expect(svr_batch.send_status).to be true
   end
@@ -176,8 +205,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::SEND_INITIAL_METADATA => {},
       CallOps::SEND_MESSAGE => sent_message
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
     expect(batch_result.send_message).to be true
 
@@ -189,8 +217,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::SEND_MESSAGE => reply_text,
       CallOps::SEND_STATUS_FROM_SERVER => the_status
     }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline,
-                                      server_ops)
+    svr_batch = server_call.run_batch(server_ops)
     expect(svr_batch.message).to eq sent_message
     expect(svr_batch.send_status).to be true
     expect(svr_batch.send_message).to be true
@@ -202,8 +229,7 @@ shared_examples 'basic GRPC message delivery is OK' do
       CallOps::RECV_MESSAGE => nil,
       CallOps::RECV_STATUS_ON_CLIENT => nil
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_close).to be true
     expect(batch_result.message).to eq reply_text
     expect(batch_result.status).to eq the_status
@@ -212,8 +238,7 @@ shared_examples 'basic GRPC message delivery is OK' do
     server_ops = {
       CallOps::RECV_CLOSE_ON_SERVER => nil
     }
-    svr_batch = server_call.run_batch(@server_queue, @server_tag, deadline,
-                                      server_ops)
+    svr_batch = server_call.run_batch(server_ops)
     expect(svr_batch.send_close).to be true
   end
 end
@@ -244,8 +269,7 @@ shared_examples 'GRPC metadata delivery works OK' do
           CallOps::SEND_INITIAL_METADATA => md
         }
         blk = proc do
-          call.run_batch(@client_queue, @client_tag, deadline,
-                         client_ops)
+          call.run_batch(client_ops)
         end
         expect(&blk).to raise_error
       end
@@ -255,15 +279,14 @@ shared_examples 'GRPC metadata delivery works OK' do
       @valid_metadata.each do |md|
         recvd_rpc = nil
         rcv_thread = Thread.new do
-          recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+          recvd_rpc = @server.request_call
         end
 
         call = new_client_call
         client_ops = {
           CallOps::SEND_INITIAL_METADATA => md
         }
-        batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                      client_ops)
+        batch_result = call.run_batch(client_ops)
         expect(batch_result.send_metadata).to be true
 
         # confirm the server can receive the client metadata
@@ -296,7 +319,7 @@ shared_examples 'GRPC metadata delivery works OK' do
       @bad_keys.each do |md|
         recvd_rpc = nil
         rcv_thread = Thread.new do
-          recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+          recvd_rpc = @server.request_call
         end
 
         call = new_client_call
@@ -305,7 +328,7 @@ shared_examples 'GRPC metadata delivery works OK' do
         client_ops = {
           CallOps::SEND_INITIAL_METADATA => nil
         }
-        call.run_batch(@client_queue, @client_tag, deadline, client_ops)
+        call.run_batch(client_ops)
 
         # server gets the invocation
         rcv_thread.join
@@ -314,8 +337,7 @@ shared_examples 'GRPC metadata delivery works OK' do
           CallOps::SEND_INITIAL_METADATA => md
         }
         blk = proc do
-          recvd_rpc.call.run_batch(@server_queue, @server_tag, deadline,
-                                   server_ops)
+          recvd_rpc.call.run_batch(server_ops)
         end
         expect(&blk).to raise_error
       end
@@ -324,7 +346,7 @@ shared_examples 'GRPC metadata delivery works OK' do
     it 'sends an empty hash if no metadata is added' do
       recvd_rpc = nil
       rcv_thread = Thread.new do
-        recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+        recvd_rpc = @server.request_call
       end
 
       call = new_client_call
@@ -333,7 +355,7 @@ shared_examples 'GRPC metadata delivery works OK' do
       client_ops = {
         CallOps::SEND_INITIAL_METADATA => nil
       }
-      call.run_batch(@client_queue, @client_tag, deadline, client_ops)
+      call.run_batch(client_ops)
 
       # server gets the invocation but sends no metadata back
       rcv_thread.join
@@ -342,14 +364,13 @@ shared_examples 'GRPC metadata delivery works OK' do
       server_ops = {
         CallOps::SEND_INITIAL_METADATA => nil
       }
-      server_call.run_batch(@server_queue, @server_tag, deadline, server_ops)
+      server_call.run_batch(server_ops)
 
       # client receives nothing as expected
       client_ops = {
         CallOps::RECV_INITIAL_METADATA => nil
       }
-      batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                    client_ops)
+      batch_result = call.run_batch(client_ops)
       expect(batch_result.metadata).to eq({})
     end
 
@@ -357,7 +378,7 @@ shared_examples 'GRPC metadata delivery works OK' do
       @valid_metadata.each do |md|
         recvd_rpc = nil
         rcv_thread = Thread.new do
-          recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+          recvd_rpc = @server.request_call
         end
 
         call = new_client_call
@@ -366,7 +387,7 @@ shared_examples 'GRPC metadata delivery works OK' do
         client_ops = {
           CallOps::SEND_INITIAL_METADATA => nil
         }
-        call.run_batch(@client_queue, @client_tag, deadline, client_ops)
+        call.run_batch(client_ops)
 
         # server gets the invocation but sends no metadata back
         rcv_thread.join
@@ -375,14 +396,13 @@ shared_examples 'GRPC metadata delivery works OK' do
         server_ops = {
           CallOps::SEND_INITIAL_METADATA => md
         }
-        server_call.run_batch(@server_queue, @server_tag, deadline, server_ops)
+        server_call.run_batch(server_ops)
 
         # client receives nothing as expected
         client_ops = {
           CallOps::RECV_INITIAL_METADATA => nil
         }
-        batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                      client_ops)
+        batch_result = call.run_batch(client_ops)
         replace_symbols = Hash[md.each_pair.collect { |x, y| [x.to_s, y] }]
         expect(batch_result.metadata).to eq(replace_symbols)
       end
@@ -393,9 +413,7 @@ end
 describe 'the http client/server' do
   before(:example) do
     server_host = '0.0.0.0:0'
-    @client_queue = GRPC::Core::CompletionQueue.new
-    @server_queue = GRPC::Core::CompletionQueue.new
-    @server = GRPC::Core::Server.new(@server_queue, nil)
+    @server = GRPC::Core::Server.new(nil)
     server_port = @server.add_http2_port(server_host, :this_port_is_insecure)
     @server.start
     @ch = Channel.new("0.0.0.0:#{server_port}", nil, :this_channel_is_insecure)
@@ -403,7 +421,7 @@ describe 'the http client/server' do
 
   after(:example) do
     @ch.close
-    @server.close(@server_queue, deadline)
+    @server.close(deadline)
   end
 
   it_behaves_like 'basic GRPC message delivery is OK' do
@@ -425,11 +443,9 @@ describe 'the secure http client/server' do
   before(:example) do
     certs = load_test_certs
     server_host = '0.0.0.0:0'
-    @client_queue = GRPC::Core::CompletionQueue.new
-    @server_queue = GRPC::Core::CompletionQueue.new
     server_creds = GRPC::Core::ServerCredentials.new(
       nil, [{ private_key: certs[1], cert_chain: certs[2] }], false)
-    @server = GRPC::Core::Server.new(@server_queue, nil)
+    @server = GRPC::Core::Server.new(nil)
     server_port = @server.add_http2_port(server_host, server_creds)
     @server.start
     args = { Channel::SSL_TARGET => 'foo.test.google.fr' }
@@ -438,7 +454,7 @@ describe 'the secure http client/server' do
   end
 
   after(:example) do
-    @server.close(@server_queue, deadline)
+    @server.close(deadline)
   end
 
   it_behaves_like 'basic GRPC message delivery is OK' do
@@ -454,7 +470,7 @@ describe 'the secure http client/server' do
     expected_md = { 'k1' => 'updated-v1', 'k2' => 'v2' }
     recvd_rpc = nil
     rcv_thread = Thread.new do
-      recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+      recvd_rpc = @server.request_call
     end
 
     call = new_client_call
@@ -462,8 +478,7 @@ describe 'the secure http client/server' do
     client_ops = {
       CallOps::SEND_INITIAL_METADATA => md
     }
-    batch_result = call.run_batch(@client_queue, @client_tag, deadline,
-                                  client_ops)
+    batch_result = call.run_batch(client_ops)
     expect(batch_result.send_metadata).to be true
 
     # confirm the server can receive the client metadata
diff --git a/src/ruby/spec/completion_queue_spec.rb b/src/ruby/spec/completion_queue_spec.rb
deleted file mode 100644
index 886a7f263b..0000000000
--- a/src/ruby/spec/completion_queue_spec.rb
+++ /dev/null
@@ -1,42 +0,0 @@
-# Copyright 2015, Google Inc.
-# All rights reserved.
-#
-# Redistribution and use in source and binary forms, with or without
-# modification, are permitted provided that the following conditions are
-# met:
-#
-#     * Redistributions of source code must retain the above copyright
-# notice, this list of conditions and the following disclaimer.
-#     * Redistributions in binary form must reproduce the above
-# copyright notice, this list of conditions and the following disclaimer
-# in the documentation and/or other materials provided with the
-# distribution.
-#     * Neither the name of Google Inc. nor the names of its
-# contributors may be used to endorse or promote products derived from
-# this software without specific prior written permission.
-#
-# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
-# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
-# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
-# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
-# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
-# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
-require 'grpc'
-
-describe GRPC::Core::CompletionQueue do
-  before(:example) do
-    @cq = GRPC::Core::CompletionQueue.new
-  end
-
-  describe '#new' do
-    it 'is constructed successufully' do
-      expect { GRPC::Core::CompletionQueue.new }.not_to raise_error
-    end
-  end
-end
diff --git a/src/ruby/spec/compression_options_spec.rb b/src/ruby/spec/compression_options_spec.rb
new file mode 100644
index 0000000000..dbd7e59294
--- /dev/null
+++ b/src/ruby/spec/compression_options_spec.rb
@@ -0,0 +1,164 @@
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+require 'grpc'
+
+describe GRPC::Core::CompressionOptions do
+  # Note these constants should be updated
+  # according to what the core lib provides.
+
+  # Names of supported compression algorithms
+  ALGORITHMS = [:identity, :deflate, :gzip]
+
+  # Names of valid supported compression levels
+  COMPRESS_LEVELS = [:none, :low, :medium, :high]
+
+  it 'implements to_s' do
+    expect { GRPC::Core::CompressionOptions.new.to_s }.to_not raise_error
+  end
+
+  it '#to_channel_arg_hash gives the same result as #to_hash' do
+    options = GRPC::Core::CompressionOptions.new
+    expect(options.to_channel_arg_hash).to eq(options.to_hash)
+  end
+
+  # Test the normal call sequence of creating an instance
+  # and then obtaining the resulting channel-arg hash that
+  # corresponds to the compression settings of the instance
+  describe 'creating, reading, and converting to channel args hash' do
+    it 'works when no optional args were provided' do
+      options = GRPC::Core::CompressionOptions.new
+
+      ALGORITHMS.each do |algorithm|
+        expect(options.algorithm_enabled?(algorithm)).to be true
+      end
+
+      expect(options.disabled_algorithms).to be_empty
+      expect(options.default_algorithm).to be nil
+      expect(options.default_level).to be nil
+      expect(options.to_hash).to be_instance_of(Hash)
+    end
+
+    it 'works when disabling multiple algorithms' do
+      options = GRPC::Core::CompressionOptions.new(
+        default_algorithm: :identity,
+        default_level: :none,
+        disabled_algorithms: [:gzip, :deflate]
+      )
+
+      [:gzip, :deflate].each do |algorithm|
+        expect(options.algorithm_enabled?(algorithm)).to be false
+        expect(options.disabled_algorithms.include?(algorithm)).to be true
+      end
+
+      expect(options.default_algorithm).to be(:identity)
+      expect(options.default_level).to be(:none)
+      expect(options.to_hash).to be_instance_of(Hash)
+    end
+
+    it 'works when all optional args have been set' do
+      options = GRPC::Core::CompressionOptions.new(
+        default_algorithm: :gzip,
+        default_level: :low,
+        disabled_algorithms: [:deflate]
+      )
+
+      expect(options.algorithm_enabled?(:deflate)).to be false
+      expect(options.algorithm_enabled?(:gzip)).to be true
+      expect(options.disabled_algorithms).to eq([:deflate])
+
+      expect(options.default_algorithm).to be(:gzip)
+      expect(options.default_level).to be(:low)
+      expect(options.to_hash).to be_instance_of(Hash)
+    end
+
+    it 'doesnt fail when no algorithms are disabled' do
+      options = GRPC::Core::CompressionOptions.new(
+        default_algorithm: :identity,
+        default_level: :high
+      )
+
+      ALGORITHMS.each do |algorithm|
+        expect(options.algorithm_enabled?(algorithm)).to be(true)
+      end
+
+      expect(options.disabled_algorithms).to be_empty
+      expect(options.default_algorithm).to be(:identity)
+      expect(options.default_level).to be(:high)
+      expect(options.to_hash).to be_instance_of(Hash)
+    end
+  end
+
+  describe '#new with bad parameters' do
+    it 'should fail with more than one parameter' do
+      blk = proc { GRPC::Core::CompressionOptions.new(:gzip, :none) }
+      expect { blk.call }.to raise_error
+    end
+
+    it 'should fail with a non-hash parameter' do
+      blk = proc { GRPC::Core::CompressionOptions.new(:gzip) }
+      expect { blk.call }.to raise_error
+    end
+  end
+
+  describe '#default_algorithm' do
+    it 'returns nil if unset' do
+      options = GRPC::Core::CompressionOptions.new
+      expect(options.default_algorithm).to be(nil)
+    end
+  end
+
+  describe '#default_level' do
+    it 'returns nil if unset' do
+      options = GRPC::Core::CompressionOptions.new
+      expect(options.default_level).to be(nil)
+    end
+  end
+
+  describe '#disabled_algorithms' do
+    it 'returns an empty list if no algorithms were disabled' do
+      options = GRPC::Core::CompressionOptions.new
+      expect(options.disabled_algorithms).to be_empty
+    end
+  end
+
+  describe '#algorithm_enabled?' do
+    [:none, :any, 'gzip', Object.new, 1].each do |name|
+      it "should fail for parameter ${name} of class #{name.class}" do
+        options = GRPC::Core::CompressionOptions.new(
+          disabled_algorithms: [:gzip])
+
+        blk = proc do
+          options.algorithm_enabled?(name)
+        end
+        expect { blk.call }.to raise_error
+      end
+    end
+  end
+end
diff --git a/src/ruby/spec/generic/active_call_spec.rb b/src/ruby/spec/generic/active_call_spec.rb
index d9c9780c93..48bc61e494 100644
--- a/src/ruby/spec/generic/active_call_spec.rb
+++ b/src/ruby/spec/generic/active_call_spec.rb
@@ -39,13 +39,8 @@ describe GRPC::ActiveCall do
 
   before(:each) do
     @pass_through = proc { |x| x }
-    @server_tag = Object.new
-    @tag = Object.new
-
-    @client_queue = GRPC::Core::CompletionQueue.new
-    @server_queue = GRPC::Core::CompletionQueue.new
     host = '0.0.0.0:0'
-    @server = GRPC::Core::Server.new(@server_queue, nil)
+    @server = GRPC::Core::Server.new(nil)
     server_port = @server.add_http2_port(host, :this_port_is_insecure)
     @server.start
     @ch = GRPC::Core::Channel.new("0.0.0.0:#{server_port}", nil,
@@ -53,21 +48,22 @@ describe GRPC::ActiveCall do
   end
 
   after(:each) do
-    @server.close(@server_queue, deadline)
+    @server.close(deadline)
   end
 
   describe 'restricted view methods' do
     before(:each) do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      @client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                    @pass_through, deadline,
-                                    metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      @client_call = ActiveCall.new(call, @pass_through,
+                                    @pass_through, deadline)
     end
 
     describe '#multi_req_view' do
-      it 'exposes a fixed subset of the ActiveCall methods' do
-        want = %w(cancelled, deadline, each_remote_read, metadata, shutdown)
+      it 'exposes a fixed subset of the ActiveCall.methods' do
+        want = %w(cancelled?, deadline, each_remote_read, metadata, \
+                  shutdown, peer, peer_cert, send_initial_metadata, \
+                  initial_metadata_sent)
         v = @client_call.multi_req_view
         want.each do |w|
           expect(v.methods.include?(w))
@@ -76,8 +72,10 @@ describe GRPC::ActiveCall do
     end
 
     describe '#single_req_view' do
-      it 'exposes a fixed subset of the ActiveCall methods' do
-        want = %w(cancelled, deadline, metadata, shutdown)
+      it 'exposes a fixed subset of the ActiveCall.methods' do
+        want = %w(cancelled?, deadline, metadata, shutdown, \
+                  send_initial_metadata, metadata_to_send, \
+                  merge_metadata_to_send, initial_metadata_sent)
         v = @client_call.single_req_view
         want.each do |w|
           expect(v.methods.include?(w))
@@ -89,46 +87,42 @@ describe GRPC::ActiveCall do
   describe '#remote_send' do
     it 'allows a client to send a payload to the server' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      @client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                    @pass_through, deadline,
-                                    metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      @client_call = ActiveCall.new(call, @pass_through,
+                                    @pass_through, deadline)
       msg = 'message is a string'
       @client_call.remote_send(msg)
 
       # check that server rpc new was received
-      recvd_rpc = @server.request_call(@server_queue, @server_tag, deadline)
+      recvd_rpc = @server.request_call
       expect(recvd_rpc).to_not eq nil
       recvd_call = recvd_rpc.call
 
       # Accept the call, and verify that the server reads the response ok.
-      server_ops = {
-        CallOps::SEND_INITIAL_METADATA => {}
-      }
-      recvd_call.run_batch(@server_queue, @server_tag, deadline, server_ops)
-      server_call = ActiveCall.new(recvd_call, @server_queue, @pass_through,
-                                   @pass_through, deadline)
+      server_call = ActiveCall.new(recvd_call, @pass_through,
+                                   @pass_through, deadline,
+                                   metadata_received: true)
       expect(server_call.remote_read).to eq(msg)
     end
 
     it 'marshals the payload using the marshal func' do
       call = make_test_call
-      ActiveCall.client_invoke(call, @client_queue)
+      ActiveCall.client_invoke(call)
       marshal = proc { |x| 'marshalled:' + x }
-      client_call = ActiveCall.new(call, @client_queue, marshal,
-                                   @pass_through, deadline)
+      client_call = ActiveCall.new(call, marshal, @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
 
       # confirm that the message was marshalled
-      recvd_rpc =  @server.request_call(@server_queue, @server_tag, deadline)
+      recvd_rpc =  @server.request_call
       recvd_call = recvd_rpc.call
       server_ops = {
         CallOps::SEND_INITIAL_METADATA => nil
       }
-      recvd_call.run_batch(@server_queue, @server_tag, deadline, server_ops)
-      server_call = ActiveCall.new(recvd_call, @server_queue, @pass_through,
-                                   @pass_through, deadline)
+      recvd_call.run_batch(server_ops)
+      server_call = ActiveCall.new(recvd_call, @pass_through,
+                                   @pass_through, deadline,
+                                   metadata_received: true)
       expect(server_call.remote_read).to eq('marshalled:' + msg)
     end
 
@@ -136,34 +130,175 @@ describe GRPC::ActiveCall do
     TEST_WRITE_FLAGS.each do |f|
       it "successfully makes calls with write_flag set to #{f}" do
         call = make_test_call
-        ActiveCall.client_invoke(call, @client_queue)
+        ActiveCall.client_invoke(call)
         marshal = proc { |x| 'marshalled:' + x }
-        client_call = ActiveCall.new(call, @client_queue, marshal,
+        client_call = ActiveCall.new(call, marshal,
                                      @pass_through, deadline)
         msg = 'message is a string'
         client_call.write_flag = f
         client_call.remote_send(msg)
 
         # confirm that the message was marshalled
-        recvd_rpc =  @server.request_call(@server_queue, @server_tag, deadline)
+        recvd_rpc =  @server.request_call
         recvd_call = recvd_rpc.call
         server_ops = {
           CallOps::SEND_INITIAL_METADATA => nil
         }
-        recvd_call.run_batch(@server_queue, @server_tag, deadline, server_ops)
-        server_call = ActiveCall.new(recvd_call, @server_queue, @pass_through,
-                                     @pass_through, deadline)
+        recvd_call.run_batch(server_ops)
+        server_call = ActiveCall.new(recvd_call, @pass_through,
+                                     @pass_through, deadline,
+                                     metadata_received: true)
         expect(server_call.remote_read).to eq('marshalled:' + msg)
       end
     end
   end
 
+  describe 'sending initial metadata', send_initial_metadata: true do
+    it 'sends metadata before sending a message if it hasnt been sent yet' do
+      call = make_test_call
+      @client_call = ActiveCall.new(
+        call,
+        @pass_through,
+        @pass_through,
+        deadline,
+        started: false)
+
+      metadata = { key: 'dummy_val', other: 'other_val' }
+      expect(@client_call.metadata_sent).to eq(false)
+      @client_call.merge_metadata_to_send(metadata)
+
+      message = 'dummy message'
+
+      expect(call).to(
+        receive(:run_batch)
+          .with(
+            hash_including(
+              CallOps::SEND_INITIAL_METADATA => metadata)).once)
+
+      expect(call).to(
+        receive(:run_batch).with(hash_including(
+                                   CallOps::SEND_MESSAGE => message)).once)
+      @client_call.remote_send(message)
+
+      expect(@client_call.metadata_sent).to eq(true)
+    end
+
+    it 'doesnt send metadata if it thinks its already been sent' do
+      call = make_test_call
+
+      @client_call = ActiveCall.new(call,
+                                    @pass_through,
+                                    @pass_through,
+                                    deadline)
+
+      expect(@client_call.metadata_sent).to eql(true)
+      expect(call).to(
+        receive(:run_batch).with(hash_including(
+                                   CallOps::SEND_INITIAL_METADATA)).never)
+
+      @client_call.remote_send('test message')
+    end
+
+    it 'sends metadata if it is explicitly sent and ok to do so' do
+      call = make_test_call
+
+      @client_call = ActiveCall.new(call,
+                                    @pass_through,
+                                    @pass_through,
+                                    deadline,
+                                    started: false)
+
+      expect(@client_call.metadata_sent).to eql(false)
+
+      metadata = { test_key: 'val' }
+      @client_call.merge_metadata_to_send(metadata)
+      expect(@client_call.metadata_to_send).to eq(metadata)
+
+      expect(call).to(
+        receive(:run_batch).with(hash_including(
+                                   CallOps::SEND_INITIAL_METADATA =>
+                                     metadata)).once)
+      @client_call.send_initial_metadata
+    end
+
+    it 'explicit sending does nothing if metadata has already been sent' do
+      call = make_test_call
+
+      @client_call = ActiveCall.new(call,
+                                    @pass_through,
+                                    @pass_through,
+                                    deadline)
+
+      expect(@client_call.metadata_sent).to eql(true)
+
+      blk = proc do
+        @client_call.send_initial_metadata
+      end
+
+      expect { blk.call }.to_not raise_error
+    end
+  end
+
+  describe '#merge_metadata_to_send', merge_metadata_to_send: true do
+    it 'adds to existing metadata when there is existing metadata to send' do
+      call = make_test_call
+      starting_metadata = {
+        k1: 'key1_val',
+        k2: 'key2_val',
+        k3: 'key3_val'
+      }
+
+      @client_call = ActiveCall.new(
+        call,
+        @pass_through, @pass_through,
+        deadline,
+        started: false,
+        metadata_to_send: starting_metadata)
+
+      expect(@client_call.metadata_to_send).to eq(starting_metadata)
+
+      @client_call.merge_metadata_to_send(
+        k3: 'key3_new_val',
+        k4: 'key4_val')
+
+      expected_md_to_send = {
+        k1: 'key1_val',
+        k2: 'key2_val',
+        k3: 'key3_new_val',
+        k4: 'key4_val' }
+
+      expect(@client_call.metadata_to_send).to eq(expected_md_to_send)
+
+      @client_call.merge_metadata_to_send(k5: 'key5_val')
+      expected_md_to_send.merge!(k5: 'key5_val')
+      expect(@client_call.metadata_to_send).to eq(expected_md_to_send)
+    end
+
+    it 'fails when initial metadata has already been sent' do
+      call = make_test_call
+      @client_call = ActiveCall.new(
+        call,
+        @pass_through,
+        @pass_through,
+        deadline,
+        started: true)
+
+      expect(@client_call.metadata_sent).to eq(true)
+
+      blk = proc do
+        @client_call.merge_metadata_to_send(k1: 'key1_val')
+      end
+
+      expect { blk.call }.to raise_error
+    end
+  end
+
   describe '#client_invoke' do
     it 'sends metadata to the server when present' do
       call = make_test_call
       metadata = { k1: 'v1', k2: 'v2' }
-      ActiveCall.client_invoke(call, @client_queue, metadata)
-      recvd_rpc =  @server.request_call(@server_queue, @server_tag, deadline)
+      ActiveCall.client_invoke(call, metadata)
+      recvd_rpc =  @server.request_call
       recvd_call = recvd_rpc.call
       expect(recvd_call).to_not be_nil
       expect(recvd_rpc.metadata).to_not be_nil
@@ -172,13 +307,31 @@ describe GRPC::ActiveCall do
     end
   end
 
-  describe '#remote_read' do
+  describe '#send_status', send_status: true do
+    it 'works when no metadata or messages have been sent yet' do
+      call = make_test_call
+      ActiveCall.client_invoke(call)
+
+      recvd_rpc = @server.request_call
+      server_call = ActiveCall.new(
+        recvd_rpc.call,
+        @pass_through,
+        @pass_through,
+        deadline,
+        started: false)
+
+      expect(server_call.metadata_sent).to eq(false)
+      blk = proc { server_call.send_status(OK) }
+      expect { blk.call }.to_not raise_error
+    end
+  end
+
+  describe '#remote_read', remote_read: true do
     it 'reads the response sent by a server' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       server_call = expect_server_to_receive(msg)
@@ -188,10 +341,9 @@ describe GRPC::ActiveCall do
 
     it 'saves no metadata when the server adds no metadata' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       server_call = expect_server_to_receive(msg)
@@ -203,10 +355,9 @@ describe GRPC::ActiveCall do
 
     it 'saves metadata add by the server' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       server_call = expect_server_to_receive(msg, k1: 'v1', k2: 'v2')
@@ -217,12 +368,36 @@ describe GRPC::ActiveCall do
       expect(client_call.metadata).to eq(expected)
     end
 
+    it 'get a status from server when nothing else sent from server' do
+      client_call = make_test_call
+      ActiveCall.client_invoke(client_call)
+
+      recvd_rpc = @server.request_call
+      recvd_call = recvd_rpc.call
+
+      server_call = ActiveCall.new(
+        recvd_call,
+        @pass_through,
+        @pass_through,
+        deadline,
+        started: false)
+
+      server_call.send_status(OK, 'OK')
+
+      # Check that we can receive initial metadata and a status
+      client_call.run_batch(
+        CallOps::RECV_INITIAL_METADATA => nil)
+      batch_result = client_call.run_batch(
+        CallOps::RECV_STATUS_ON_CLIENT => nil)
+
+      expect(batch_result.status.code).to eq(OK)
+    end
+
     it 'get a nil msg before a status when an OK status is sent' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       client_call.writes_done(false)
@@ -236,11 +411,10 @@ describe GRPC::ActiveCall do
 
     it 'unmarshals the response using the unmarshal func' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
+      ActiveCall.client_invoke(call)
       unmarshal = proc { |x| 'unmarshalled:' + x }
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   unmarshal, deadline,
-                                   metadata_tag: md_tag)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   unmarshal, deadline)
 
       # confirm the client receives the unmarshalled message
       msg = 'message is a string'
@@ -254,17 +428,16 @@ describe GRPC::ActiveCall do
   describe '#each_remote_read' do
     it 'creates an Enumerator' do
       call = make_test_call
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
+      client_call = ActiveCall.new(call, @pass_through,
                                    @pass_through, deadline)
       expect(client_call.each_remote_read).to be_a(Enumerator)
     end
 
     it 'the returns an enumerator that can read n responses' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       reply = 'server_response'
       client_call.remote_send(msg)
@@ -279,10 +452,9 @@ describe GRPC::ActiveCall do
 
     it 'the returns an enumerator that stops after an OK Status' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       reply = 'server_response'
       client_call.remote_send(msg)
@@ -302,10 +474,9 @@ describe GRPC::ActiveCall do
   describe '#writes_done' do
     it 'finishes ok if the server sends a status response' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       expect { client_call.writes_done(false) }.to_not raise_error
@@ -318,10 +489,9 @@ describe GRPC::ActiveCall do
 
     it 'finishes ok if the server sends an early status response' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       server_call = expect_server_to_receive(msg)
@@ -334,10 +504,9 @@ describe GRPC::ActiveCall do
 
     it 'finishes ok if writes_done is true' do
       call = make_test_call
-      md_tag = ActiveCall.client_invoke(call, @client_queue)
-      client_call = ActiveCall.new(call, @client_queue, @pass_through,
-                                   @pass_through, deadline,
-                                   metadata_tag: md_tag)
+      ActiveCall.client_invoke(call)
+      client_call = ActiveCall.new(call, @pass_through,
+                                   @pass_through, deadline)
       msg = 'message is a string'
       client_call.remote_send(msg)
       server_call = expect_server_to_receive(msg)
@@ -348,6 +517,86 @@ describe GRPC::ActiveCall do
     end
   end
 
+  # Test sending of the initial metadata in #run_server_bidi
+  # from the server handler both implicitly and explicitly.
+  describe '#run_server_bidi metadata sending tests', run_server_bidi: true do
+    before(:each) do
+      @requests = ['first message', 'second message']
+      @server_to_client_metadata = { 'test_key' => 'test_val' }
+      @server_status = OK
+
+      @client_call = make_test_call
+      @client_call.run_batch(CallOps::SEND_INITIAL_METADATA => {})
+
+      recvd_rpc = @server.request_call
+      recvd_call = recvd_rpc.call
+      @server_call = ActiveCall.new(
+        recvd_call,
+        @pass_through,
+        @pass_through,
+        deadline,
+        metadata_received: true,
+        started: false,
+        metadata_to_send: @server_to_client_metadata)
+    end
+
+    after(:each) do
+      # Send the requests and send a close so the server can send a status
+      @requests.each do |message|
+        @client_call.run_batch(CallOps::SEND_MESSAGE => message)
+      end
+      @client_call.run_batch(CallOps::SEND_CLOSE_FROM_CLIENT => nil)
+
+      @server_thread.join
+
+      # Expect that initial metadata was sent,
+      # the requests were echoed, and a status was sent
+      batch_result = @client_call.run_batch(
+        CallOps::RECV_INITIAL_METADATA => nil)
+      expect(batch_result.metadata).to eq(@server_to_client_metadata)
+
+      @requests.each do |message|
+        batch_result = @client_call.run_batch(
+          CallOps::RECV_MESSAGE => nil)
+        expect(batch_result.message).to eq(message)
+      end
+
+      batch_result = @client_call.run_batch(
+        CallOps::RECV_STATUS_ON_CLIENT => nil)
+      expect(batch_result.status.code).to eq(@server_status)
+    end
+
+    it 'sends the initial metadata implicitly if not already sent' do
+      # Server handler that doesn't have access to a "call"
+      # It echoes the requests
+      fake_gen_each_reply_with_no_call_param = proc do |msgs|
+        msgs
+      end
+
+      @server_thread = Thread.new do
+        @server_call.run_server_bidi(
+          fake_gen_each_reply_with_no_call_param)
+        @server_call.send_status(@server_status)
+      end
+    end
+
+    it 'sends the metadata when sent explicitly and not already sent' do
+      # Fake server handler that has access to a "call" object and
+      # uses it to explicitly update and send the initial metadata
+      fake_gen_each_reply_with_call_param = proc do |msgs, call_param|
+        call_param.merge_metadata_to_send(@server_to_client_metadata)
+        call_param.send_initial_metadata
+        msgs
+      end
+
+      @server_thread = Thread.new do
+        @server_call.run_server_bidi(
+          fake_gen_each_reply_with_call_param)
+        @server_call.send_status(@server_status)
+      end
+    end
+  end
+
   def expect_server_to_receive(sent_text, **kw)
     c = expect_server_to_be_invoked(**kw)
     expect(c.remote_read).to eq(sent_text)
@@ -355,17 +604,16 @@ describe GRPC::ActiveCall do
   end
 
   def expect_server_to_be_invoked(**kw)
-    recvd_rpc =  @server.request_call(@server_queue, @server_tag, deadline)
+    recvd_rpc =  @server.request_call
     expect(recvd_rpc).to_not eq nil
     recvd_call = recvd_rpc.call
-    recvd_call.run_batch(@server_queue, @server_tag, deadline,
-                         CallOps::SEND_INITIAL_METADATA => kw)
-    ActiveCall.new(recvd_call, @server_queue, @pass_through,
-                   @pass_through, deadline)
+    recvd_call.run_batch(CallOps::SEND_INITIAL_METADATA => kw)
+    ActiveCall.new(recvd_call, @pass_through, @pass_through, deadline,
+                   metadata_received: true, started: true)
   end
 
   def make_test_call
-    @ch.create_call(@client_queue, nil, nil, '/method', nil, deadline)
+    @ch.create_call(nil, nil, '/method', nil, deadline)
   end
 
   def deadline
diff --git a/src/ruby/spec/generic/client_stub_spec.rb b/src/ruby/spec/generic/client_stub_spec.rb
index 168e7fb791..6034b5419c 100644
--- a/src/ruby/spec/generic/client_stub_spec.rb
+++ b/src/ruby/spec/generic/client_stub_spec.rb
@@ -29,11 +29,14 @@
 
 require 'grpc'
 
+Thread.abort_on_exception = true
+
 def wakey_thread(&blk)
   n = GRPC::Notifier.new
   t = Thread.new do
     blk.call(n)
   end
+  t.abort_on_exception = true
   n.wait
   t
 end
@@ -54,15 +57,13 @@ describe 'ClientStub' do
   before(:each) do
     Thread.abort_on_exception = true
     @server = nil
-    @server_queue = nil
     @method = 'an_rpc_method'
     @pass = OK
     @fail = INTERNAL
-    @cq = GRPC::Core::CompletionQueue.new
   end
 
   after(:each) do
-    @server.close(@server_queue) unless @server_queue.nil?
+    @server.close(from_relative_time(2)) unless @server.nil?
   end
 
   describe '#new' do
@@ -70,7 +71,7 @@ describe 'ClientStub' do
     it 'can be created from a host and args' do
       opts = { channel_args: { a_channel_arg: 'an_arg' } }
       blk = proc do
-        GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts)
+        GRPC::ClientStub.new(fake_host, :this_channel_is_insecure, **opts)
       end
       expect(&blk).not_to raise_error
     end
@@ -81,7 +82,7 @@ describe 'ClientStub' do
         channel_override: @ch
       }
       blk = proc do
-        GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts)
+        GRPC::ClientStub.new(fake_host, :this_channel_is_insecure, **opts)
       end
       expect(&blk).not_to raise_error
     end
@@ -92,7 +93,7 @@ describe 'ClientStub' do
           channel_args: { a_channel_arg: 'an_arg' },
           channel_override: Object.new
         }
-        GRPC::ClientStub.new(fake_host, @cq, :this_channel_is_insecure, **opts)
+        GRPC::ClientStub.new(fake_host, :this_channel_is_insecure, **opts)
       end
       expect(&blk).to raise_error
     end
@@ -100,7 +101,7 @@ describe 'ClientStub' do
     it 'cannot be created with bad credentials' do
       blk = proc do
         opts = { channel_args: { a_channel_arg: 'an_arg' } }
-        GRPC::ClientStub.new(fake_host, @cq, Object.new, **opts)
+        GRPC::ClientStub.new(fake_host, Object.new, **opts)
       end
       expect(&blk).to raise_error
     end
@@ -115,7 +116,7 @@ describe 'ClientStub' do
           }
         }
         creds = GRPC::Core::ChannelCredentials.new(certs[0], nil, nil)
-        GRPC::ClientStub.new(fake_host, @cq, creds,  **opts)
+        GRPC::ClientStub.new(fake_host, creds,  **opts)
       end
       expect(&blk).to_not raise_error
     end
@@ -130,7 +131,7 @@ describe 'ClientStub' do
       it 'should send a request to/receive a reply from a server' do
         server_port = create_test_server
         th = run_request_response(@sent_msg, @resp, @pass)
-        stub = GRPC::ClientStub.new("localhost:#{server_port}", @cq,
+        stub = GRPC::ClientStub.new("localhost:#{server_port}",
                                     :this_channel_is_insecure)
         expect(get_response(stub)).to eq(@resp)
         th.join
@@ -141,7 +142,7 @@ describe 'ClientStub' do
         host = "localhost:#{server_port}"
         th = run_request_response(@sent_msg, @resp, @pass,
                                   k1: 'v1', k2: 'v2')
-        stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         expect(get_response(stub)).to eq(@resp)
         th.join
       end
@@ -151,7 +152,7 @@ describe 'ClientStub' do
         alt_host = "localhost:#{server_port}"
         th = run_request_response(@sent_msg, @resp, @pass)
         ch = GRPC::Core::Channel.new(alt_host, nil, :this_channel_is_insecure)
-        stub = GRPC::ClientStub.new('ignored-host', @cq,
+        stub = GRPC::ClientStub.new('ignored-host',
                                     :this_channel_is_insecure,
                                     channel_override: ch)
         expect(get_response(stub)).to eq(@resp)
@@ -162,7 +163,7 @@ describe 'ClientStub' do
         server_port = create_test_server
         host = "localhost:#{server_port}"
         th = run_request_response(@sent_msg, @resp, @fail)
-        stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         blk = proc { get_response(stub) }
         expect(&blk).to raise_error(GRPC::BadStatus)
         th.join
@@ -182,7 +183,8 @@ describe 'ClientStub' do
       def get_response(stub)
         op = stub.request_response(@method, @sent_msg, noop, noop,
                                    return_op: true,
-                                   metadata: { k1: 'v1', k2: 'v2' })
+                                   metadata: { k1: 'v1', k2: 'v2' },
+                                   deadline: from_relative_time(2))
         expect(op).to be_a(GRPC::ActiveCall::Operation)
         op.execute
       end
@@ -196,7 +198,7 @@ describe 'ClientStub' do
       before(:each) do
         server_port = create_test_server
         host = "localhost:#{server_port}"
-        @stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        @stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         @metadata = { k1: 'v1', k2: 'v2' }
         @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s }
         @resp = 'a_reply'
@@ -262,7 +264,7 @@ describe 'ClientStub' do
         server_port = create_test_server
         host = "localhost:#{server_port}"
         th = run_server_streamer(@sent_msg, @replys, @pass)
-        stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         expect(get_responses(stub).collect { |r| r }).to eq(@replys)
         th.join
       end
@@ -271,7 +273,7 @@ describe 'ClientStub' do
         server_port = create_test_server
         host = "localhost:#{server_port}"
         th = run_server_streamer(@sent_msg, @replys, @fail)
-        stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         e = get_responses(stub)
         expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus)
         th.join
@@ -282,7 +284,7 @@ describe 'ClientStub' do
         host = "localhost:#{server_port}"
         th = run_server_streamer(@sent_msg, @replys, @fail,
                                  k1: 'v1', k2: 'v2')
-        stub = GRPC::ClientStub.new(host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(host, :this_channel_is_insecure)
         e = get_responses(stub)
         expect { e.collect { |r| r } }.to raise_error(GRPC::BadStatus)
         th.join
@@ -327,7 +329,7 @@ describe 'ClientStub' do
       it 'supports sending all the requests first', bidi: true do
         th = run_bidi_streamer_handle_inputs_first(@sent_msgs, @replys,
                                                    @pass)
-        stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(@host, :this_channel_is_insecure)
         e = get_responses(stub)
         expect(e.collect { |r| r }).to eq(@replys)
         th.join
@@ -335,7 +337,7 @@ describe 'ClientStub' do
 
       it 'supports client-initiated ping pong', bidi: true do
         th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, true)
-        stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(@host, :this_channel_is_insecure)
         e = get_responses(stub)
         expect(e.collect { |r| r }).to eq(@sent_msgs)
         th.join
@@ -343,7 +345,7 @@ describe 'ClientStub' do
 
       it 'supports a server-initiated ping pong', bidi: true do
         th = run_bidi_streamer_echo_ping_pong(@sent_msgs, @pass, false)
-        stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure)
+        stub = GRPC::ClientStub.new(@host, :this_channel_is_insecure)
         e = get_responses(stub)
         expect(e.collect { |r| r }).to eq(@sent_msgs)
         th.join
@@ -372,26 +374,6 @@ describe 'ClientStub' do
 
       it_behaves_like 'bidi streaming'
     end
-
-    describe 'without enough time to run' do
-      before(:each) do
-        @sent_msgs = Array.new(3) { |i| 'msg_' + (i + 1).to_s }
-        @replys = Array.new(3) { |i| 'reply_' + (i + 1).to_s }
-        server_port = create_test_server
-        @host = "localhost:#{server_port}"
-      end
-
-      it 'should fail with DeadlineExceeded', bidi: true do
-        @server.start
-        stub = GRPC::ClientStub.new(@host, @cq, :this_channel_is_insecure)
-        blk = proc do
-          e = stub.bidi_streamer(@method, @sent_msgs, noop, noop,
-                                 deadline: from_relative_time(0.001))
-          e.collect { |r| r }
-        end
-        expect(&blk).to raise_error GRPC::BadStatus, /Deadline Exceeded/
-      end
-    end
   end
 
   def run_server_streamer(expected_input, replys, status, **kw)
@@ -460,21 +442,18 @@ describe 'ClientStub' do
   end
 
   def create_test_server
-    @server_queue = GRPC::Core::CompletionQueue.new
-    @server = GRPC::Core::Server.new(@server_queue, nil)
+    @server = GRPC::Core::Server.new(nil)
     @server.add_http2_port('0.0.0.0:0', :this_port_is_insecure)
   end
 
   def expect_server_to_be_invoked(notifier)
     @server.start
     notifier.notify(nil)
-    server_tag = Object.new
-    recvd_rpc = @server.request_call(@server_queue, server_tag,
-                                     INFINITE_FUTURE)
+    recvd_rpc = @server.request_call
     recvd_call = recvd_rpc.call
     recvd_call.metadata = recvd_rpc.metadata
-    recvd_call.run_batch(@server_queue, server_tag, Time.now + 2,
-                         SEND_INITIAL_METADATA => nil)
-    GRPC::ActiveCall.new(recvd_call, @server_queue, noop, noop, INFINITE_FUTURE)
+    recvd_call.run_batch(SEND_INITIAL_METADATA => nil)
+    GRPC::ActiveCall.new(recvd_call, noop, noop, INFINITE_FUTURE,
+                         metadata_received: true)
   end
 end
diff --git a/src/ruby/spec/generic/rpc_desc_spec.rb b/src/ruby/spec/generic/rpc_desc_spec.rb
index d2080b7ca2..1a895005bc 100644
--- a/src/ruby/spec/generic/rpc_desc_spec.rb
+++ b/src/ruby/spec/generic/rpc_desc_spec.rb
@@ -196,6 +196,9 @@ describe GRPC::RpcDesc do
     def fake_svstream(_arg1, _arg2)
     end
 
+    def fake_three_args(_arg1, _arg2, _arg3)
+    end
+
     it 'raises when a request_response does not have 2 args' do
       [:fake_clstream, :no_arg].each do |mth|
         blk = proc do
@@ -244,8 +247,8 @@ describe GRPC::RpcDesc do
       expect(&blk).to_not raise_error
     end
 
-    it 'raises when a bidi streamer does not have 1 arg' do
-      [:fake_svstream, :no_arg].each do |mth|
+    it 'raises when a bidi streamer does not have 1 or 2 args' do
+      [:fake_three_args, :no_arg].each do |mth|
         blk = proc do
           @bidi_streamer.assert_arity_matches(method(mth))
         end
@@ -259,6 +262,13 @@ describe GRPC::RpcDesc do
       end
       expect(&blk).to_not raise_error
     end
+
+    it 'passes when a bidi streamer has 2 args' do
+      blk = proc do
+        @bidi_streamer.assert_arity_matches(method(:fake_svstream))
+      end
+      expect(&blk).to_not raise_error
+    end
   end
 
   describe '#request_response?' do
diff --git a/src/ruby/spec/generic/rpc_server_spec.rb b/src/ruby/spec/generic/rpc_server_spec.rb
index 943502cea2..31157cf161 100644
--- a/src/ruby/spec/generic/rpc_server_spec.rb
+++ b/src/ruby/spec/generic/rpc_server_spec.rb
@@ -95,7 +95,7 @@ class FailingService
   def initialize(_default_var = 'ignored')
     @details = 'app error'
     @code = 101
-    @md = { failed_method: 'an_rpc' }
+    @md = { 'failed_method' => 'an_rpc' }
   end
 
   def an_rpc(_req, _call)
@@ -135,8 +135,6 @@ describe GRPC::RpcServer do
     @pass = 0
     @fail = 1
     @noop = proc { |x| x }
-
-    @server_queue = GRPC::Core::CompletionQueue.new
   end
 
   describe '#new' do
@@ -148,28 +146,6 @@ describe GRPC::RpcServer do
       expect(&blk).not_to raise_error
     end
 
-    it 'can be created with a completion queue override' do
-      opts = {
-        server_args: { a_channel_arg: 'an_arg' },
-        completion_queue_override: @server_queue
-      }
-      blk = proc do
-        RpcServer.new(**opts)
-      end
-      expect(&blk).not_to raise_error
-    end
-
-    it 'cannot be created with a bad completion queue override' do
-      blk = proc do
-        opts = {
-          server_args: { a_channel_arg: 'an_arg' },
-          completion_queue_override: Object.new
-        }
-        RpcServer.new(**opts)
-      end
-      expect(&blk).to raise_error
-    end
-
     it 'cannot be created with invalid ServerCredentials' do
       blk = proc do
         opts = {
@@ -294,7 +270,6 @@ describe GRPC::RpcServer do
     context 'with no connect_metadata' do
       before(:each) do
         server_opts = {
-          completion_queue_override: @server_queue,
           poll_period: 1
         }
         @srv = RpcServer.new(**server_opts)
@@ -309,8 +284,7 @@ describe GRPC::RpcServer do
         @srv.wait_till_running
         req = EchoMsg.new
         blk = proc do
-          cq = GRPC::Core::CompletionQueue.new
-          stub = GRPC::ClientStub.new(@host, cq, :this_channel_is_insecure,
+          stub = GRPC::ClientStub.new(@host, :this_channel_is_insecure,
                                       **client_opts)
           stub.request_response('/unknown', req, marshal, unmarshal)
         end
@@ -325,8 +299,7 @@ describe GRPC::RpcServer do
         @srv.wait_till_running
         req = EchoMsg.new
         blk = proc do
-          cq = GRPC::Core::CompletionQueue.new
-          stub = GRPC::ClientStub.new(@host, cq, :this_channel_is_insecure,
+          stub = GRPC::ClientStub.new(@host, :this_channel_is_insecure,
                                       **client_opts)
           stub.request_response('/an_rpc', req, marshal, unmarshal)
         end
@@ -422,7 +395,6 @@ describe GRPC::RpcServer do
       it 'should return RESOURCE_EXHAUSTED on too many jobs', server: true do
         opts = {
           server_args: { a_channel_arg: 'an_arg' },
-          completion_queue_override: @server_queue,
           pool_size: 1,
           poll_period: 1,
           max_waiting_requests: 0
@@ -466,7 +438,6 @@ describe GRPC::RpcServer do
       end
       before(:each) do
         server_opts = {
-          completion_queue_override: @server_queue,
           poll_period: 1,
           connect_md_proc: test_md_proc
         }
@@ -502,7 +473,6 @@ describe GRPC::RpcServer do
     context 'with trailing metadata' do
       before(:each) do
         server_opts = {
-          completion_queue_override: @server_queue,
           poll_period: 1
         }
         @srv = RpcServer.new(**server_opts)
@@ -545,7 +515,7 @@ describe GRPC::RpcServer do
         op = stub.an_rpc(req, return_op: true, metadata: { k1: 'v1', k2: 'v2' })
         expect(op.metadata).to be nil
         expect(op.execute).to be_a(EchoMsg)
-        expect(op.metadata).to eq(wanted_trailers)
+        expect(op.trailing_metadata).to eq(wanted_trailers)
         @srv.stop
         t.join
       end
diff --git a/src/ruby/spec/pb/duplicate/codegen_spec.rb b/src/ruby/spec/pb/duplicate/codegen_spec.rb
index 54c136c510..ea0240965c 100644
--- a/src/ruby/spec/pb/duplicate/codegen_spec.rb
+++ b/src/ruby/spec/pb/duplicate/codegen_spec.rb
@@ -44,7 +44,7 @@ describe 'Ping protobuf code generation' do
       # Get the current content
       service_path = File.join(root_dir, 'src', 'ruby', 'pb', 'grpc',
                                'testing', 'duplicate',
-                               'echo_duplicate_services.rb')
+                               'echo_duplicate_services_pb.rb')
       want = nil
       File.open(service_path) { |f| want = f.read }
 
@@ -54,7 +54,7 @@ describe 'Ping protobuf code generation' do
       got = nil
       Dir.mktmpdir do |tmp_dir|
         gen_out = File.join(tmp_dir, 'src', 'proto', 'grpc', 'testing',
-                            'duplicate', 'echo_duplicate_services.rb')
+                            'duplicate', 'echo_duplicate_services_pb.rb')
         pid = spawn(
           'protoc',
           '-I.',
diff --git a/src/ruby/spec/pb/health/checker_spec.rb b/src/ruby/spec/pb/health/checker_spec.rb
index f3d121a31e..1b2fa96827 100644
--- a/src/ruby/spec/pb/health/checker_spec.rb
+++ b/src/ruby/spec/pb/health/checker_spec.rb
@@ -28,7 +28,7 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 require 'grpc'
-require 'grpc/health/v1/health'
+require 'grpc/health/v1/health_pb'
 require 'grpc/health/checker'
 require 'open3'
 require 'tmpdir'
@@ -43,7 +43,7 @@ describe 'Health protobuf code generation' do
       skip 'protoc || grpc_ruby_plugin missing, cannot verify health code-gen'
     else
       it 'should already be loaded indirectly i.e, used by the other specs' do
-        expect(require('grpc/health/v1/health_services')).to be(false)
+        expect(require('grpc/health/v1/health_services_pb')).to be(false)
       end
 
       it 'should have the same content as created by code generation' do
@@ -52,7 +52,7 @@ describe 'Health protobuf code generation' do
 
         # Get the current content
         service_path = File.join(root_dir, 'ruby', 'pb', 'grpc',
-                                 'health', 'v1', 'health_services.rb')
+                                 'health', 'v1', 'health_services_pb.rb')
         want = nil
         File.open(service_path) { |f| want = f.read }
 
@@ -62,7 +62,7 @@ describe 'Health protobuf code generation' do
         got = nil
         Dir.mktmpdir do |tmp_dir|
           gen_out = File.join(tmp_dir, 'grpc', 'health', 'v1',
-                              'health_services.rb')
+                              'health_services_pb.rb')
           pid = spawn(
             'protoc',
             '-I.',
@@ -168,11 +168,9 @@ describe Grpc::Health::Checker do
     CheckerStub = Grpc::Health::Checker.rpc_stub_class
 
     before(:each) do
-      @server_queue = GRPC::Core::CompletionQueue.new
       server_host = '0.0.0.0:0'
       @client_opts = { channel_override: @ch }
       server_opts = {
-        completion_queue_override: @server_queue,
         poll_period: 1
       }
       @srv = RpcServer.new(**server_opts)
diff --git a/src/ruby/spec/server_spec.rb b/src/ruby/spec/server_spec.rb
index 439b19fb8d..003d8f69d5 100644
--- a/src/ruby/spec/server_spec.rb
+++ b/src/ruby/spec/server_spec.rb
@@ -43,19 +43,15 @@ describe Server do
     GRPC::Core::ServerCredentials.new(*load_test_certs)
   end
 
-  before(:each) do
-    @cq = GRPC::Core::CompletionQueue.new
-  end
-
   describe '#start' do
     it 'runs without failing' do
-      blk = proc { Server.new(@cq, nil).start }
+      blk = proc { Server.new(nil).start }
       expect(&blk).to_not raise_error
     end
 
     it 'fails if the server is closed' do
-      s = Server.new(@cq, nil)
-      s.close(@cq)
+      s = Server.new(nil)
+      s.close
       expect { s.start }.to raise_error(RuntimeError)
     end
   end
@@ -63,19 +59,19 @@ describe Server do
   describe '#destroy' do
     it 'destroys a server ok' do
       s = start_a_server
-      blk = proc { s.destroy(@cq) }
+      blk = proc { s.destroy }
       expect(&blk).to_not raise_error
     end
 
     it 'can be called more than once without error' do
       s = start_a_server
       begin
-        blk = proc { s.destroy(@cq) }
+        blk = proc { s.destroy }
         expect(&blk).to_not raise_error
         blk.call
         expect(&blk).to_not raise_error
       ensure
-        s.close(@cq)
+        s.close
       end
     end
   end
@@ -84,7 +80,7 @@ describe Server do
     it 'closes a server ok' do
       s = start_a_server
       begin
-        blk = proc { s.close(@cq) }
+        blk = proc { s.close }
         expect(&blk).to_not raise_error
       ensure
         s.close(@cq)
@@ -93,7 +89,7 @@ describe Server do
 
     it 'can be called more than once without error' do
       s = start_a_server
-      blk = proc { s.close(@cq) }
+      blk = proc { s.close }
       expect(&blk).to_not raise_error
       blk.call
       expect(&blk).to_not raise_error
@@ -104,16 +100,16 @@ describe Server do
     describe 'for insecure servers' do
       it 'runs without failing' do
         blk = proc do
-          s = Server.new(@cq, nil)
+          s = Server.new(nil)
           s.add_http2_port('localhost:0', :this_port_is_insecure)
-          s.close(@cq)
+          s.close
         end
         expect(&blk).to_not raise_error
       end
 
       it 'fails if the server is closed' do
-        s = Server.new(@cq, nil)
-        s.close(@cq)
+        s = Server.new(nil)
+        s.close
         blk = proc do
           s.add_http2_port('localhost:0', :this_port_is_insecure)
         end
@@ -125,16 +121,16 @@ describe Server do
       let(:cert) { create_test_cert }
       it 'runs without failing' do
         blk = proc do
-          s = Server.new(@cq, nil)
+          s = Server.new(nil)
           s.add_http2_port('localhost:0', cert)
-          s.close(@cq)
+          s.close
         end
         expect(&blk).to_not raise_error
       end
 
       it 'fails if the server is closed' do
-        s = Server.new(@cq, nil)
-        s.close(@cq)
+        s = Server.new(nil)
+        s.close
         blk = proc { s.add_http2_port('localhost:0', cert) }
         expect(&blk).to raise_error(RuntimeError)
       end
@@ -142,8 +138,8 @@ describe Server do
   end
 
   shared_examples '#new' do
-    it 'takes a completion queue with nil channel args' do
-      expect { Server.new(@cq, nil) }.to_not raise_error
+    it 'takes nil channel args' do
+      expect { Server.new(nil) }.to_not raise_error
     end
 
     it 'does not take a hash with bad keys as channel args' do
@@ -194,14 +190,14 @@ describe Server do
 
   describe '#new with an insecure channel' do
     def construct_with_args(a)
-      proc { Server.new(@cq, a) }
+      proc { Server.new(a) }
     end
 
     it_behaves_like '#new'
   end
 
   def start_a_server
-    s = Server.new(@cq, nil)
+    s = Server.new(nil)
     s.add_http2_port('0.0.0.0:0', :this_port_is_insecure)
     s.start
     s
diff --git a/src/ruby/stress/metrics_server.rb b/src/ruby/stress/metrics_server.rb
index 13638c4d21..2b7f78577d 100644
--- a/src/ruby/stress/metrics_server.rb
+++ b/src/ruby/stress/metrics_server.rb
@@ -27,8 +27,8 @@
 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-require_relative '../pb/grpc/testing/metrics.rb'
-require_relative '../pb/grpc/testing/metrics_services.rb'
+require_relative '../pb/grpc/testing/metrics_pb.rb'
+require_relative '../pb/grpc/testing/metrics_services_pb.rb'
 
 class Gauge
   def get_name
diff --git a/src/ruby/tools/bin/grpc_tools_ruby_protoc.rb b/src/ruby/tools/bin/grpc_tools_ruby_protoc
index 3a2a5b8dc9..dab06e7958 100755
--- a/src/ruby/tools/bin/grpc_tools_ruby_protoc.rb
+++ b/src/ruby/tools/bin/grpc_tools_ruby_protoc
@@ -32,10 +32,17 @@ require 'rbconfig'
 
 require_relative '../os_check'
 
-protoc_name = 'protoc' + RbConfig::CONFIG['EXEEXT']
+ext = RbConfig::CONFIG['EXEEXT']
 
-protoc_path = File.join(File.dirname(__FILE__),
-                        RbConfig::CONFIG['host_cpu'] + '-' + OS.os_name,
-                        protoc_name)
+protoc_name = 'protoc' + ext
 
-exec([ protoc_path, protoc_path ], *ARGV)
+plugin_name = 'grpc_ruby_plugin' + ext
+
+protoc_dir = File.join(File.dirname(__FILE__),
+                       RbConfig::CONFIG['host_cpu'] + '-' + OS.os_name)
+
+protoc_path = File.join(protoc_dir, protoc_name)
+
+plugin_path = File.join(protoc_dir, plugin_name)
+
+exec([ protoc_path, protoc_path ], "--plugin=protoc-gen-grpc=#{plugin_path}", *ARGV)
diff --git a/src/ruby/tools/bin/grpc_tools_ruby_protoc_plugin.rb b/src/ruby/tools/bin/grpc_tools_ruby_protoc_plugin
index 4b296dedc7..4b296dedc7 100755
--- a/src/ruby/tools/bin/grpc_tools_ruby_protoc_plugin.rb
+++ b/src/ruby/tools/bin/grpc_tools_ruby_protoc_plugin
diff --git a/src/ruby/tools/grpc-tools.gemspec b/src/ruby/tools/grpc-tools.gemspec
index 9fa4b66392..68e2a7a113 100644
--- a/src/ruby/tools/grpc-tools.gemspec
+++ b/src/ruby/tools/grpc-tools.gemspec
@@ -18,5 +18,5 @@ Gem::Specification.new do |s|
 
   s.platform = Gem::Platform::RUBY
 
-  s.executables = %w( grpc_tools_ruby_protoc.rb grpc_tools_ruby_protoc_plugin.rb )
+  s.executables = %w( grpc_tools_ruby_protoc grpc_tools_ruby_protoc_plugin )
 end
diff --git a/src/ruby/tools/version.rb b/src/ruby/tools/version.rb
index dca7fd7e72..e457ec09dd 100644
--- a/src/ruby/tools/version.rb
+++ b/src/ruby/tools/version.rb
@@ -29,6 +29,6 @@
 
 module GRPC
   module Tools
-    VERSION = '0.15.0.dev'
+    VERSION = '1.1.0.dev'
   end
 end