Merge pull request grpc#3586 from dgquintas/conn_state_watcher_plucking

Added subchannel ability to unsubscribe to connectivity state changes.
mcanthony · Oct 2, 2015 · 8a50de7 · 8a50de7
2 parents 7cc2777 + 3b14781
commit 8a50de7
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 4 deletions.
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
@@ -413,6 +413,17 @@ void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
   }
 }
 
+int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
+                                             grpc_subchannel *c,
+                                             grpc_closure *subscribed_notify) {
+  int success;
+  gpr_mu_lock(&c->mu);
+  success = grpc_connectivity_state_change_unsubscribe(
+      exec_ctx, &c->state_tracker, subscribed_notify);
+  gpr_mu_unlock(&c->mu);
+  return success;
+}
+
 void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
                                           grpc_subchannel *c,
                                           grpc_transport_op *op) {

diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h
@@ -64,13 +64,13 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
 #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS
 #endif
 
-void grpc_subchannel_ref(grpc_subchannel *channel
-                             GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_ref(
+    grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
                            grpc_subchannel *channel
                                GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
-void grpc_subchannel_call_ref(grpc_subchannel_call *call
-                                  GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_call_ref(
+    grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
 void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
                                 grpc_subchannel_call *call
                                     GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
@@ -98,6 +98,13 @@ void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
                                             grpc_connectivity_state *state,
                                             grpc_closure *notify);
 
+/** Remove \a subscribed_notify from the list of closures to be called on a
+ * state change if present, returning 1. Otherwise, nothing is done and return
+ * 0. */
+int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
+                                             grpc_subchannel *channel,
+                                             grpc_closure *subscribed_notify);
+
 /** express interest in \a channel's activities through \a pollset. */
 void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
                                           grpc_subchannel *channel,

diff --git a/src/core/transport/connectivity_state.c b/src/core/transport/connectivity_state.c
@@ -115,6 +115,27 @@ int grpc_connectivity_state_notify_on_state_change(
   return tracker->current_state == GRPC_CHANNEL_IDLE;
 }
 
+int grpc_connectivity_state_change_unsubscribe(
+    grpc_exec_ctx *exec_ctx, grpc_connectivity_state_tracker *tracker,
+    grpc_closure *subscribed_notify) {
+  grpc_connectivity_state_watcher *w = tracker->watchers;
+  if (w != NULL && w->notify == subscribed_notify) {
+    tracker->watchers = w->next;
+    gpr_free(w);
+    return 1;
+  }
+  while (w != NULL) {
+    grpc_connectivity_state_watcher *rm_candidate = w->next;
+    if (rm_candidate != NULL && rm_candidate->notify == subscribed_notify) {
+      w->next = w->next->next;
+      gpr_free(rm_candidate);
+      return 1;
+    }
+    w = w->next;
+  }
+  return 0;
+}
+
 void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
                                  grpc_connectivity_state_tracker *tracker,
                                  grpc_connectivity_state state,

diff --git a/src/core/transport/connectivity_state.h b/src/core/transport/connectivity_state.h
@@ -78,4 +78,11 @@ int grpc_connectivity_state_notify_on_state_change(
     grpc_exec_ctx *exec_ctx, grpc_connectivity_state_tracker *tracker,
     grpc_connectivity_state *current, grpc_closure *notify);
 
+/** Remove \a subscribed_notify from the list of closures to be called on a
+ * state change if present, returning 1. Otherwise, nothing is done and return
+ * 0. */
+int grpc_connectivity_state_change_unsubscribe(
+    grpc_exec_ctx *exec_ctx, grpc_connectivity_state_tracker *tracker,
+    grpc_closure *subscribed_notify);
+
 #endif /* GRPC_INTERNAL_CORE_TRANSPORT_CONNECTIVITY_STATE_H */