Merge branch 'grpclb_resolver_changes2' into service_config

1 files changed, 18 insertions, 8 deletions

diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index c50585d154..2cf5490122 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -184,6 +184,8 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
     grpc_lb_policy_args lb_policy_args;
     lb_policy_args.addresses =
         grpc_resolver_result_get_addresses(chand->incoming_resolver_result);
+    lb_policy_args.additional_args = grpc_resolver_result_get_lb_policy_args(
+        chand->incoming_resolver_result);
     lb_policy_args.client_channel_factory = chand->client_channel_factory;
     lb_policy = grpc_lb_policy_create(
         exec_ctx,
@@ -413,13 +415,15 @@ typedef struct client_channel_call_data {
   grpc_mdstr *path;
   grpc_method_config *method_config;
 
-  grpc_transport_stream_op *waiting_ops;
+  grpc_transport_stream_op **waiting_ops;
   size_t waiting_ops_count;
   size_t waiting_ops_capacity;
 
   grpc_closure next_step;
 
   grpc_call_stack *owning_call;
+
+  grpc_linked_mdelem lb_token_mdelem;
 } call_data;
 
 static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
@@ -430,7 +434,7 @@ static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
         gpr_realloc(calld->waiting_ops,
                     calld->waiting_ops_capacity * sizeof(*calld->waiting_ops));
   }
-  calld->waiting_ops[calld->waiting_ops_count++] = *op;
+  calld->waiting_ops[calld->waiting_ops_count++] = op;
   GPR_TIMER_END("add_waiting_locked", 0);
 }
 
@@ -439,14 +443,14 @@ static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld,
   size_t i;
   for (i = 0; i < calld->waiting_ops_count; i++) {
     grpc_transport_stream_op_finish_with_failure(
-        exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error));
+        exec_ctx, calld->waiting_ops[i], GRPC_ERROR_REF(error));
   }
   calld->waiting_ops_count = 0;
   GRPC_ERROR_UNREF(error);
 }
 
 typedef struct {
-  grpc_transport_stream_op *ops;
+  grpc_transport_stream_op **ops;
   size_t nops;
   grpc_subchannel_call *call;
 } retry_ops_args;
@@ -455,7 +459,7 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
   retry_ops_args *a = args;
   size_t i;
   for (i = 0; i < a->nops; i++) {
-    grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]);
+    grpc_subchannel_call_process_op(exec_ctx, a->call, a->ops[i]);
   }
   GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops");
   gpr_free(a->ops);
@@ -463,6 +467,10 @@ static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) {
 }
 
 static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
+  if (calld->waiting_ops_count == 0) {
+    return;
+  }
+
   retry_ops_args *a = gpr_malloc(sizeof(*a));
   a->ops = calld->waiting_ops;
   a->nops = calld->waiting_ops_count;
@@ -599,9 +607,11 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
     int r;
     GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
     gpr_mu_unlock(&chand->mu);
-    r = grpc_lb_policy_pick(exec_ctx, lb_policy, calld->pollent,
-                            initial_metadata, initial_metadata_flags,
-                            connected_subchannel, on_ready);
+    const grpc_lb_policy_pick_args inputs = {calld->pollent, initial_metadata,
+                                             initial_metadata_flags,
+                                             &calld->lb_token_mdelem};
+    r = grpc_lb_policy_pick(exec_ctx, lb_policy, &inputs, connected_subchannel,
+                            NULL, on_ready);
     GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel");
     GPR_TIMER_END("pick_subchannel", 0);
     return r;