Merge branch 'plucking-hell' into the-test-be-sleepy

20 files changed, 422 insertions, 29 deletions

diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c
index 91125cb149..48c623d359 100644
--- a/src/core/channel/http_client_filter.c
+++ b/src/core/channel/http_client_filter.c
@@ -40,10 +40,12 @@
 typedef struct call_data {
   grpc_linked_mdelem method;
   grpc_linked_mdelem scheme;
+  grpc_linked_mdelem authority;
   grpc_linked_mdelem te_trailers;
   grpc_linked_mdelem content_type;
   grpc_linked_mdelem user_agent;
   int sent_initial_metadata;
+  int sent_authority;
 
   int got_initial_metadata;
   grpc_stream_op_buffer *recv_ops;
@@ -62,6 +64,7 @@ typedef struct channel_data {
   grpc_mdelem *scheme;
   grpc_mdelem *content_type;
   grpc_mdelem *status;
+  grpc_mdelem *default_authority;
   /** complete user agent mdelem */
   grpc_mdelem *user_agent;
 } channel_data;
@@ -100,6 +103,7 @@ static void hc_on_recv(void *user_data, int success) {
 
 static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) {
   grpc_call_element *elem = user_data;
+  call_data *calld = elem->call_data;
   channel_data *channeld = elem->channel_data;
   /* eat the things we'd like to set ourselves */
   if (md->key == channeld->method->key) return NULL;
@@ -107,6 +111,10 @@ static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) {
   if (md->key == channeld->te_trailers->key) return NULL;
   if (md->key == channeld->content_type->key) return NULL;
   if (md->key == channeld->user_agent->key) return NULL;
+  if (channeld->default_authority &&
+      channeld->default_authority->key == md->key) {
+    calld->sent_authority = 1;
+  }
   return md;
 }
 
@@ -130,6 +138,11 @@ static void hc_mutate_op(grpc_call_element *elem,
                                    GRPC_MDELEM_REF(channeld->method));
       grpc_metadata_batch_add_head(&op->data.metadata, &calld->scheme,
                                    GRPC_MDELEM_REF(channeld->scheme));
+      if (channeld->default_authority && !calld->sent_authority) {
+        grpc_metadata_batch_add_head(
+            &op->data.metadata, &calld->authority,
+            GRPC_MDELEM_REF(channeld->default_authority));
+      }
       grpc_metadata_batch_add_tail(&op->data.metadata, &calld->te_trailers,
                                    GRPC_MDELEM_REF(channeld->te_trailers));
       grpc_metadata_batch_add_tail(&op->data.metadata, &calld->content_type,
@@ -162,6 +175,7 @@ static void init_call_elem(grpc_call_element *elem,
   call_data *calld = elem->call_data;
   calld->sent_initial_metadata = 0;
   calld->got_initial_metadata = 0;
+  calld->sent_authority = 0;
   calld->on_done_recv = NULL;
   grpc_iomgr_closure_init(&calld->hc_on_recv, hc_on_recv, elem);
   if (initial_op) hc_mutate_op(elem, initial_op);
@@ -241,8 +255,10 @@ static grpc_mdstr *user_agent_from_args(grpc_mdctx *mdctx,
 
 /* Constructor for channel_data */
 static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
-                              const grpc_channel_args *args, grpc_mdctx *mdctx,
-                              int is_first, int is_last) {
+                              const grpc_channel_args *channel_args,
+                              grpc_mdctx *mdctx, int is_first, int is_last) {
+  size_t i;
+
   /* grab pointers to our data from the channel element */
   channel_data *channeld = elem->channel_data;
 
@@ -251,17 +267,32 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
      path */
   GPR_ASSERT(!is_last);
 
+  channeld->default_authority = NULL;
+  if (channel_args) {
+    for (i = 0; i < channel_args->num_args; i++) {
+      if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) {
+        if (channel_args->args[i].type != GRPC_ARG_STRING) {
+          gpr_log(GPR_ERROR, "%s: must be an string",
+                  GRPC_ARG_DEFAULT_AUTHORITY);
+        } else {
+          channeld->default_authority = grpc_mdelem_from_strings(
+              mdctx, ":authority", channel_args->args[i].value.string);
+        }
+      }
+    }
+  }
+
   /* initialize members */
   channeld->te_trailers = grpc_mdelem_from_strings(mdctx, "te", "trailers");
   channeld->method = grpc_mdelem_from_strings(mdctx, ":method", "POST");
-  channeld->scheme =
-      grpc_mdelem_from_strings(mdctx, ":scheme", scheme_from_args(args));
+  channeld->scheme = grpc_mdelem_from_strings(mdctx, ":scheme",
+                                              scheme_from_args(channel_args));
   channeld->content_type =
       grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc");
   channeld->status = grpc_mdelem_from_strings(mdctx, ":status", "200");
   channeld->user_agent = grpc_mdelem_from_metadata_strings(
       mdctx, grpc_mdstr_from_string(mdctx, "user-agent", 0),
-      user_agent_from_args(mdctx, args));
+      user_agent_from_args(mdctx, channel_args));
 }
 
 /* Destructor for channel data */
@@ -275,6 +306,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
   GRPC_MDELEM_UNREF(channeld->content_type);
   GRPC_MDELEM_UNREF(channeld->status);
   GRPC_MDELEM_UNREF(channeld->user_agent);
+  if (channeld->default_authority) {
+    GRPC_MDELEM_UNREF(channeld->default_authority);
+  }
 }
 
 const grpc_channel_filter grpc_http_client_filter = {
diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c
index 9d89eb9bf2..0955ae319a 100644
--- a/src/core/channel/http_server_filter.c
+++ b/src/core/channel/http_server_filter.c
@@ -44,6 +44,7 @@ typedef struct call_data {
   gpr_uint8 sent_status;
   gpr_uint8 seen_scheme;
   gpr_uint8 seen_te_trailers;
+  gpr_uint8 seen_authority;
   grpc_linked_mdelem status;
 
   grpc_stream_op_buffer *recv_ops;
@@ -125,6 +126,9 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
     }
     calld->seen_path = 1;
     return md;
+  } else if (md->key == channeld->authority_key) {
+    calld->seen_authority = 1;
+    return md;
   } else if (md->key == channeld->host_key) {
     /* translate host to :authority since :authority may be
        omitted */
@@ -132,6 +136,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
         channeld->mdctx, GRPC_MDSTR_REF(channeld->authority_key),
         GRPC_MDSTR_REF(md->value));
     GRPC_MDELEM_UNREF(md);
+    calld->seen_authority = 1;
     return authority;
   } else {
     return md;
@@ -154,12 +159,15 @@ static void hs_on_recv(void *user_data, int success) {
          (:method, :scheme, content-type, with :path and :authority covered
          at the channel level right now) */
       if (calld->seen_post && calld->seen_scheme && calld->seen_te_trailers &&
-          calld->seen_path) {
+          calld->seen_path && calld->seen_authority) {
         /* do nothing */
       } else {
         if (!calld->seen_path) {
           gpr_log(GPR_ERROR, "Missing :path header");
         }
+        if (!calld->seen_authority) {
+          gpr_log(GPR_ERROR, "Missing :authority header");
+        }
         if (!calld->seen_post) {
           gpr_log(GPR_ERROR, "Missing :method header");
         }
diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c
index ac401bc4d3..827b1a2be5 100644
--- a/src/core/client_config/resolvers/dns_resolver.c
+++ b/src/core/client_config/resolvers/dns_resolver.c
@@ -36,9 +36,11 @@
 #include <string.h>
 
 #include <grpc/support/alloc.h>
+#include <grpc/support/host_port.h>
 #include <grpc/support/string_util.h>
 
 #include "src/core/client_config/lb_policies/pick_first.h"
+#include "src/core/client_config/subchannel_factory_decorators/add_channel_arg.h"
 #include "src/core/iomgr/resolve_address.h"
 #include "src/core/support/string.h"
 
@@ -201,6 +203,9 @@ static grpc_resolver *dns_create(
     grpc_subchannel_factory *subchannel_factory) {
   dns_resolver *r;
   const char *path = uri->path;
+  grpc_arg default_host_arg;
+  char *host;
+  char *port;
 
   if (0 != strcmp(uri->authority, "")) {
     gpr_log(GPR_ERROR, "authority based uri's not supported");
@@ -209,6 +214,16 @@ static grpc_resolver *dns_create(
 
   if (path[0] == '/') ++path;
 
+  gpr_split_host_port(path, &host, &port);
+
+  default_host_arg.type = GRPC_ARG_STRING;
+  default_host_arg.key = GRPC_ARG_DEFAULT_AUTHORITY;
+  default_host_arg.value.string = host;
+  subchannel_factory = grpc_subchannel_factory_add_channel_arg(subchannel_factory, &default_host_arg);
+
+  gpr_free(host);
+  gpr_free(port);
+
   r = gpr_malloc(sizeof(dns_resolver));
   memset(r, 0, sizeof(*r));
   gpr_ref_init(&r->refs, 1);
@@ -218,7 +233,6 @@ static grpc_resolver *dns_create(
   r->default_port = gpr_strdup(default_port);
   r->subchannel_factory = subchannel_factory;
   r->lb_policy_factory = lb_policy_factory;
-  grpc_subchannel_factory_ref(subchannel_factory);
   return &r->base;
 }
 
diff --git a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c
new file mode 100644
index 0000000000..7dc6d99ebe
--- /dev/null
+++ b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.c
@@ -0,0 +1,43 @@
+/*
+ *
+ * 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 "src/core/client_config/subchannel_factory_decorators/add_channel_arg.h"
+#include "src/core/client_config/subchannel_factory_decorators/merge_channel_args.h"
+
+grpc_subchannel_factory *grpc_subchannel_factory_add_channel_arg(
+		grpc_subchannel_factory *input, const grpc_arg *arg) {
+	grpc_channel_args args;
+	args.num_args = 1;
+	args.args = (grpc_arg *)arg;
+	return grpc_subchannel_factory_merge_channel_args(input, &args);
+}
diff --git a/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h
new file mode 100644
index 0000000000..1937623374
--- /dev/null
+++ b/src/core/client_config/subchannel_factory_decorators/add_channel_arg.h
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H
+#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H
+
+#include "src/core/client_config/subchannel_factory.h"
+
+/** Takes a subchannel factory, returns a new one that mutates incoming
+    channel_args by adding a new argument; ownership of input, arg is retained
+    by the caller. */
+grpc_subchannel_factory *grpc_subchannel_factory_add_channel_arg(
+		grpc_subchannel_factory *input, const grpc_arg *arg);
+
+#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_ADD_CHANNEL_ARG_H */
diff --git a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c
new file mode 100644
index 0000000000..7e028857ac
--- /dev/null
+++ b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.c
@@ -0,0 +1,84 @@
+/*
+ *
+ * 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 "src/core/client_config/subchannel_factory_decorators/merge_channel_args.h"
+#include <grpc/support/alloc.h>
+#include "src/core/channel/channel_args.h"
+
+typedef struct {
+  grpc_subchannel_factory base;
+  gpr_refcount refs;
+  grpc_subchannel_factory *wrapped;
+  grpc_channel_args *merge_args;
+} merge_args_factory;
+
+static void merge_args_factory_ref(grpc_subchannel_factory *scf) {
+  merge_args_factory *f = (merge_args_factory *)scf;
+  gpr_ref(&f->refs);
+}
+
+static void merge_args_factory_unref(grpc_subchannel_factory *scf) {
+  merge_args_factory *f = (merge_args_factory *)scf;
+  if (gpr_unref(&f->refs)) {
+  	grpc_subchannel_factory_unref(f->wrapped);
+    grpc_channel_args_destroy(f->merge_args);
+    gpr_free(f);
+  }
+}
+
+static grpc_subchannel *merge_args_factory_create_subchannel(
+    grpc_subchannel_factory *scf, grpc_subchannel_args *args) {
+  merge_args_factory *f = (merge_args_factory *)scf;
+  grpc_channel_args *final_args =
+      grpc_channel_args_merge(args->args, f->merge_args);
+  grpc_subchannel *s;
+  args->args = final_args;
+  s = grpc_subchannel_factory_create_subchannel(f->wrapped, args);
+  grpc_channel_args_destroy(final_args);
+  return s;
+}
+
+static const grpc_subchannel_factory_vtable merge_args_factory_vtable = {
+    merge_args_factory_ref, merge_args_factory_unref,
+    merge_args_factory_create_subchannel};
+
+grpc_subchannel_factory *grpc_subchannel_factory_merge_channel_args(
+		grpc_subchannel_factory *input, const grpc_channel_args *args) {
+  merge_args_factory *f = gpr_malloc(sizeof(*f));
+  f->base.vtable = &merge_args_factory_vtable;
+  gpr_ref_init(&f->refs, 1);
+  grpc_subchannel_factory_ref(input);
+  f->wrapped = input;
+  f->merge_args = grpc_channel_args_copy(args);
+  return &f->base;
+}
diff --git a/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h
new file mode 100644
index 0000000000..73a03b752f
--- /dev/null
+++ b/src/core/client_config/subchannel_factory_decorators/merge_channel_args.h
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H
+#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H
+
+#include "src/core/client_config/subchannel_factory.h"
+
+/** Takes a subchannel factory, returns a new one that mutates incoming
+    channel_args by adding a new argument; ownership of input, args is retained
+    by the caller. */
+grpc_subchannel_factory *grpc_subchannel_factory_merge_channel_args(
+		grpc_subchannel_factory *input, const grpc_channel_args *args);
+
+#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_DECORATORS_MERGE_CHANNEL_ARGS_H */
diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index 583d350128..81f673f856 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -149,14 +149,17 @@ static grpc_call *grpc_channel_create_call_internal(
     grpc_channel *channel, grpc_completion_queue *cq, grpc_mdelem *path_mdelem,
     grpc_mdelem *authority_mdelem, gpr_timespec deadline) {
   grpc_mdelem *send_metadata[2];
+  int num_metadata = 0;
 
   GPR_ASSERT(channel->is_client);
 
-  send_metadata[0] = path_mdelem;
-  send_metadata[1] = authority_mdelem;
+  send_metadata[num_metadata++] = path_mdelem;
+  if (authority_mdelem != NULL) {
+    send_metadata[num_metadata++] = authority_mdelem;
+  }
 
   return grpc_call_create(channel, cq, NULL, send_metadata,
-                          GPR_ARRAY_SIZE(send_metadata), deadline);
+                          num_metadata, deadline);
 }
 
 grpc_call *grpc_channel_create_call(grpc_channel *channel,
@@ -168,9 +171,10 @@ grpc_call *grpc_channel_create_call(grpc_channel *channel,
       grpc_mdelem_from_metadata_strings(
           channel->metadata_context, GRPC_MDSTR_REF(channel->path_string),
           grpc_mdstr_from_string(channel->metadata_context, method, 0)),
+      host ?
       grpc_mdelem_from_metadata_strings(
           channel->metadata_context, GRPC_MDSTR_REF(channel->authority_string),
-          grpc_mdstr_from_string(channel->metadata_context, host, 0)),
+          grpc_mdstr_from_string(channel->metadata_context, host, 0)) : NULL,
       deadline);
 }
 
@@ -180,9 +184,9 @@ void *grpc_channel_register_call(grpc_channel *channel, const char *method,
   rc->path = grpc_mdelem_from_metadata_strings(
       channel->metadata_context, GRPC_MDSTR_REF(channel->path_string),
       grpc_mdstr_from_string(channel->metadata_context, method, 0));
-  rc->authority = grpc_mdelem_from_metadata_strings(
+  rc->authority = host ? grpc_mdelem_from_metadata_strings(
       channel->metadata_context, GRPC_MDSTR_REF(channel->authority_string),
-      grpc_mdstr_from_string(channel->metadata_context, host, 0));
+      grpc_mdstr_from_string(channel->metadata_context, host, 0)) : NULL;
   gpr_mu_lock(&channel->registered_call_mu);
   rc->next = channel->registered_calls;
   channel->registered_calls = rc;
@@ -196,7 +200,7 @@ grpc_call *grpc_channel_create_registered_call(
   registered_call *rc = registered_call_handle;
   return grpc_channel_create_call_internal(
       channel, completion_queue, GRPC_MDELEM_REF(rc->path),
-      GRPC_MDELEM_REF(rc->authority), deadline);
+      rc->authority ? GRPC_MDELEM_REF(rc->authority) : NULL, deadline);
 }
 
 #ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 732813fed0..9d6f78db55 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -45,8 +45,6 @@
 #include <grpc/support/atm.h>
 #include <grpc/support/log.h>
 
-#define MAX_PLUCKERS 4
-
 typedef struct {
   grpc_pollset_worker *worker;
   void *tag;
@@ -68,7 +66,7 @@ struct grpc_completion_queue {
   int shutdown_called;
   int is_server_cq;
   int num_pluckers;
-  plucker pluckers[MAX_PLUCKERS];
+  plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
 };
 
 grpc_completion_queue *grpc_completion_queue_create(void) {
@@ -203,12 +201,15 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
   return ret;
 }
 
-static void add_plucker(grpc_completion_queue *cc, void *tag,
-                        grpc_pollset_worker *worker) {
-  GPR_ASSERT(cc->num_pluckers != MAX_PLUCKERS);
+static int add_plucker(grpc_completion_queue *cc, void *tag,
+                       grpc_pollset_worker *worker) {
+  if (cc->num_pluckers == GRPC_MAX_COMPLETION_QUEUE_PLUCKERS) {
+    return 0;
+  }
   cc->pluckers[cc->num_pluckers].tag = tag;
   cc->pluckers[cc->num_pluckers].worker = worker;
   cc->num_pluckers++;
+  return 1;
 }
 
 static void del_plucker(grpc_completion_queue *cc, void *tag,
@@ -261,7 +262,16 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
       ret.type = GRPC_QUEUE_SHUTDOWN;
       break;
     }
-    add_plucker(cc, tag, &worker);
+    if (!add_plucker(cc, tag, &worker)) {
+      gpr_log(GPR_DEBUG, 
+              "Too many outstanding grpc_completion_queue_pluck calls: maximum is %d".
+              GRPC_MAX_COMPLETION_QUEUE_PLUCKERS);
+      gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+      memset(&ret, 0, sizeof(ret));
+      /* TODO(ctiller): should we use a different result here */
+      ret.type = GRPC_QUEUE_TIMEOUT;
+      break;
+    }
     if (!grpc_pollset_work(&cc->pollset, &worker, deadline)) {
       del_plucker(cc, tag, &worker);
       gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 35b3867620..63c1685a89 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -575,8 +575,10 @@ static void server_on_recv(void *ptr, int success) {
           gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) {
         calld->deadline = op->data.metadata.deadline;
       }
-      calld->got_initial_metadata = 1;
-      start_new_rpc(elem);
+      if (calld->host && calld->path) {
+        calld->got_initial_metadata = 1;
+        start_new_rpc(elem);
+      }
       break;
     }
   }
diff --git a/src/python/grpcio/.gitignore b/src/python/grpcio/.gitignore
index d89f3db999..efbe1737ba 100644
--- a/src/python/grpcio/.gitignore
+++ b/src/python/grpcio/.gitignore
@@ -1,5 +1,5 @@
 MANIFEST
-grpcio.egg-info/
+*.egg-info/
 build/
 dist/
 *.egg
diff --git a/src/python/grpcio/MANIFEST.in b/src/python/grpcio/MANIFEST.in
index 498b55f20a..9583dc7768 100644
--- a/src/python/grpcio/MANIFEST.in
+++ b/src/python/grpcio/MANIFEST.in
@@ -1,2 +1,3 @@
 graft grpc
 include commands.py
+include requirements.txt
diff --git a/src/python/requirements.txt b/src/python/grpcio/requirements.txt
index 43395df03b..43395df03b 100644
--- a/src/python/requirements.txt
+++ b/src/python/grpcio/requirements.txt
diff --git a/src/python/grpcio/setup.py b/src/python/grpcio/setup.py
index 1333ae0086..e408f2ace9 100644
--- a/src/python/grpcio/setup.py
+++ b/src/python/grpcio/setup.py
@@ -89,7 +89,7 @@ _PACKAGE_DIRECTORIES = {
 _INSTALL_REQUIRES = (
     'enum34==1.0.4',
     'futures==2.2.0',
-    'protobuf==3.0.0a3'
+    'protobuf==3.0.0a3',
 )
 
 _SETUP_REQUIRES = (
@@ -97,7 +97,7 @@ _SETUP_REQUIRES = (
 ) + _INSTALL_REQUIRES
 
 _COMMAND_CLASS = {
-    'doc': commands.SphinxDocumentation
+    'doc': commands.SphinxDocumentation,
 }
 
 setuptools.setup(
diff --git a/src/python/grpcio_test/.gitignore b/src/python/grpcio_test/.gitignore
new file mode 100644
index 0000000000..218e3a15ab
--- /dev/null
+++ b/src/python/grpcio_test/.gitignore
@@ -0,0 +1,10 @@
+MANIFEST
+*.egg-info/
+build/
+dist/
+*.egg
+*.egg/
+*.eggs/
+.coverage
+.coverage.*
+nosetests.xml
diff --git a/src/python/grpcio_test/MANIFEST.in b/src/python/grpcio_test/MANIFEST.in
new file mode 100644
index 0000000000..c9327307dc
--- /dev/null
+++ b/src/python/grpcio_test/MANIFEST.in
@@ -0,0 +1,4 @@
+graft grpc_interop
+graft grpc_test
+include commands.py
+include requirements.txt
diff --git a/src/python/grpcio_test/commands.py b/src/python/grpcio_test/commands.py
new file mode 100644
index 0000000000..c796d94c76
--- /dev/null
+++ b/src/python/grpcio_test/commands.py
@@ -0,0 +1,57 @@
+# 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.
+
+"""Provides distutils command classes for the GRPC Python test setup process."""
+
+import os
+import os.path
+import sys
+
+import setuptools
+
+
+class RunTests(setuptools.Command):
+  """Command to run all tests via py.test."""
+
+  description = ''
+  user_options = [('pytest-args=', 'a', 'arguments to pass to py.test')]
+
+  def initialize_options(self):
+    self.pytest_args = []
+
+  def finalize_options(self):
+    pass
+
+  def run(self):
+    # We import here to ensure that setup.py has had a chance to install the
+    # relevant package eggs first.
+    import pytest
+    result = pytest.main(self.pytest_args)
+    if result != 0:
+      raise SystemExit(result)
diff --git a/src/python/grpcio_test/requirements.txt b/src/python/grpcio_test/requirements.txt
new file mode 100644
index 0000000000..856198def5
--- /dev/null
+++ b/src/python/grpcio_test/requirements.txt
@@ -0,0 +1,5 @@
+pytest>=2.6
+pytest-cov>=2.0
+pytest-xdist>=1.11
+oauth2client>=1.4.7
+grpcio>=0.10.0a0
diff --git a/src/python/grpcio_test/setup.cfg b/src/python/grpcio_test/setup.cfg
new file mode 100644
index 0000000000..b32d3f5972
--- /dev/null
+++ b/src/python/grpcio_test/setup.cfg
@@ -0,0 +1,3 @@
+[pytest]
+norecursedirs = _cython
+python_files = *_test.py
diff --git a/src/python/grpcio_test/setup.py b/src/python/grpcio_test/setup.py
index 3f57ef04e0..925c32720f 100644
--- a/src/python/grpcio_test/setup.py
+++ b/src/python/grpcio_test/setup.py
@@ -29,8 +29,17 @@
 
 """A setup module for the GRPC Python interop testing package."""
 
+import os
+import os.path
+
 import setuptools
 
+# Ensure we're in the proper directory whether or not we're being used by pip.
+os.chdir(os.path.dirname(os.path.abspath(__file__)))
+
+# Break import-style to ensure we can actually find our commands module.
+import commands
+
 _PACKAGES = setuptools.find_packages('.', exclude=['*._cython', '*._cython.*'])
 
 _PACKAGE_DIRECTORIES = {
@@ -43,13 +52,28 @@ _PACKAGE_DATA = {
         'credentials/server1.pem',]
 }
 
-_INSTALL_REQUIRES = ['oauth2client>=1.4.7', 'grpcio>=0.10.0a0']
+_SETUP_REQUIRES = (
+    'pytest>=2.6',
+    'pytest-cov>=2.0',
+    'pytest-xdist>=1.11',
+)
+
+_INSTALL_REQUIRES = (
+    'oauth2client>=1.4.7',
+    'grpcio>=0.10.0a0',
+)
+
+_COMMAND_CLASS = {
+    'test': commands.RunTests
+}
 
 setuptools.setup(
     name='grpcio_test',
-    version='0.0.1',
+    version='0.10.0a0',
     packages=_PACKAGES,
     package_dir=_PACKAGE_DIRECTORIES,
     package_data=_PACKAGE_DATA,
-    install_requires=_INSTALL_REQUIRES
+    install_requires=_INSTALL_REQUIRES + _SETUP_REQUIRES,
+    setup_requires=_SETUP_REQUIRES,
+    cmdclass=_COMMAND_CLASS
 )