Merge branch 'tor-github/pr/1040'

5 files changed, 80 insertions, 61 deletions

diff --git a/src/feature/control/btrack.c b/src/feature/control/btrack.c
index d3d12cb2b7..3ce97dc855 100644
--- a/src/feature/control/btrack.c
+++ b/src/feature/control/btrack.c
@@ -24,6 +24,7 @@
 #include "feature/control/btrack_circuit.h"
 #include "feature/control/btrack_orconn.h"
 #include "feature/control/btrack_sys.h"
+#include "lib/pubsub/pubsub.h"
 #include "lib/subsys/subsys.h"
 
 static int
@@ -31,8 +32,6 @@ btrack_init(void)
 {
   if (btrack_orconn_init())
     return -1;
-  if (btrack_circ_init())
-    return -1;
 
   return 0;
 }
@@ -44,10 +43,22 @@ btrack_fini(void)
   btrack_circ_fini();
 }
 
+static int
+btrack_add_pubsub(pubsub_connector_t *connector)
+{
+  if (btrack_orconn_add_pubsub(connector))
+    return -1;
+  if (btrack_circ_add_pubsub(connector))
+    return -1;
+
+  return 0;
+}
+
 const subsys_fns_t sys_btrack = {
   .name = "btrack",
   .supported = true,
   .level = -30,
   .initialize = btrack_init,
   .shutdown = btrack_fini,
+  .add_pubsub = btrack_add_pubsub,
 };
diff --git a/src/feature/control/btrack_circuit.c b/src/feature/control/btrack_circuit.c
index dcee9e460e..2980c77ddc 100644
--- a/src/feature/control/btrack_circuit.c
+++ b/src/feature/control/btrack_circuit.c
@@ -109,51 +109,53 @@ btc_update_evtype(const ocirc_cevent_msg_t *msg, btc_best_t *best,
   return false;
 }
 
+DECLARE_SUBSCRIBE(ocirc_state, btc_state_rcvr);
+DECLARE_SUBSCRIBE(ocirc_cevent, btc_cevent_rcvr);
+DECLARE_SUBSCRIBE(ocirc_chan, btc_chan_rcvr);
+
 static void
-btc_state_rcvr(const ocirc_state_msg_t *msg)
+btc_state_rcvr(const msg_t *msg, const ocirc_state_msg_t *arg)
 {
+  (void)msg;
   log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" state=%d onehop=%d",
-            msg->gid, msg->state, msg->onehop);
+            arg->gid, arg->state, arg->onehop);
 
-  btc_update_state(msg, &best_any_state, "ANY");
-  if (msg->onehop)
+  btc_update_state(arg, &best_any_state, "ANY");
+  if (arg->onehop)
     return;
-  btc_update_state(msg, &best_ap_state, "AP");
+  btc_update_state(arg, &best_ap_state, "AP");
 }
 
 static void
-btc_cevent_rcvr(const ocirc_cevent_msg_t *msg)
+btc_cevent_rcvr(const msg_t *msg, const ocirc_cevent_msg_t *arg)
 {
+  (void)msg;
   log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" evtype=%d reason=%d onehop=%d",
-            msg->gid, msg->evtype, msg->reason, msg->onehop);
+            arg->gid, arg->evtype, arg->reason, arg->onehop);
 
-  btc_update_evtype(msg, &best_any_evtype, "ANY");
-  if (msg->onehop)
+  btc_update_evtype(arg, &best_any_evtype, "ANY");
+  if (arg->onehop)
     return;
-  btc_update_evtype(msg, &best_ap_evtype, "AP");
+  btc_update_evtype(arg, &best_ap_evtype, "AP");
 }
 
 static void
-btc_event_rcvr(const ocirc_event_msg_t *msg)
+btc_chan_rcvr(const msg_t *msg, const ocirc_chan_msg_t *arg)
 {
-  switch (msg->type) {
-  case OCIRC_MSGTYPE_STATE:
-    return btc_state_rcvr(&msg->u.state);
-  case OCIRC_MSGTYPE_CHAN:
-    log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" chan=%"PRIu64" onehop=%d",
-              msg->u.chan.gid, msg->u.chan.chan, msg->u.chan.onehop);
-    break;
-  case OCIRC_MSGTYPE_CEVENT:
-    return btc_cevent_rcvr(&msg->u.cevent);
-  default:
-    break;
-  }
+  (void)msg;
+  log_debug(LD_BTRACK, "CIRC gid=%"PRIu32" chan=%"PRIu64" onehop=%d",
+            arg->gid, arg->chan, arg->onehop);
 }
 
 int
-btrack_circ_init(void)
+btrack_circ_add_pubsub(pubsub_connector_t *connector)
 {
-  ocirc_event_subscribe(btc_event_rcvr);
+  if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_chan))
+    return -1;
+  if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_cevent))
+    return -1;
+  if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_state))
+    return -1;
   return 0;
 }
 
diff --git a/src/feature/control/btrack_circuit.h b/src/feature/control/btrack_circuit.h
index 9e06fefb07..b326c22ccf 100644
--- a/src/feature/control/btrack_circuit.h
+++ b/src/feature/control/btrack_circuit.h
@@ -9,7 +9,10 @@
 #ifndef TOR_BTRACK_CIRCUIT_H
 #define TOR_BTRACK_CIRCUIT_H
 
+#include "lib/pubsub/pubsub.h"
+
 int btrack_circ_init(void);
 void btrack_circ_fini(void);
+int btrack_circ_add_pubsub(pubsub_connector_t *);
 
 #endif /* !defined(TOR_BTRACK_CIRCUIT_H) */
diff --git a/src/feature/control/btrack_orconn.c b/src/feature/control/btrack_orconn.c
index 93ebe8d9cc..922b542a0c 100644
--- a/src/feature/control/btrack_orconn.c
+++ b/src/feature/control/btrack_orconn.c
@@ -45,6 +45,11 @@
 #include "feature/control/btrack_orconn_cevent.h"
 #include "feature/control/btrack_orconn_maps.h"
 #include "lib/log/log.h"
+#include "lib/pubsub/pubsub.h"
+
+DECLARE_SUBSCRIBE(orconn_state, bto_state_rcvr);
+DECLARE_SUBSCRIBE(orconn_status, bto_status_rcvr);
+DECLARE_SUBSCRIBE(ocirc_chan, bto_chan_rcvr);
 
 /** Pair of a best ORCONN GID and with its state */
 typedef struct bto_best_t {
@@ -110,16 +115,17 @@ bto_reset_bests(void)
  * message comes from code in connection_or.c.
  **/
 static void
-bto_state_rcvr(const orconn_state_msg_t *msg)
+bto_state_rcvr(const msg_t *msg, const orconn_state_msg_t *arg)
 {
   bt_orconn_t *bto;
 
-  bto = bto_find_or_new(msg->gid, msg->chan);
+  (void)msg;
+  bto = bto_find_or_new(arg->gid, arg->chan);
   log_debug(LD_BTRACK, "ORCONN gid=%"PRIu64" chan=%"PRIu64
             " proxy_type=%d state=%d",
-            msg->gid, msg->chan, msg->proxy_type, msg->state);
-  bto->proxy_type = msg->proxy_type;
-  bto->state = msg->state;
+            arg->gid, arg->chan, arg->proxy_type, arg->state);
+  bto->proxy_type = arg->proxy_type;
+  bto->state = arg->state;
   if (bto->is_orig)
     bto_update_bests(bto);
 }
@@ -130,54 +136,38 @@ bto_state_rcvr(const orconn_state_msg_t *msg)
  * control.c.
  **/
 static void
-bto_status_rcvr(const orconn_status_msg_t *msg)
+bto_status_rcvr(const msg_t *msg, const orconn_status_msg_t *arg)
 {
-  switch (msg->status) {
+  (void)msg;
+  switch (arg->status) {
   case OR_CONN_EVENT_FAILED:
   case OR_CONN_EVENT_CLOSED:
     log_info(LD_BTRACK, "ORCONN DELETE gid=%"PRIu64" status=%d reason=%d",
-             msg->gid, msg->status, msg->reason);
-    return bto_delete(msg->gid);
+             arg->gid, arg->status, arg->reason);
+    return bto_delete(arg->gid);
   default:
     break;
   }
 }
 
-/** Dispatch to individual ORCONN message handlers */
-static void
-bto_event_rcvr(const orconn_event_msg_t *msg)
-{
-  switch (msg->type) {
-  case ORCONN_MSGTYPE_STATE:
-    return bto_state_rcvr(&msg->u.state);
-  case ORCONN_MSGTYPE_STATUS:
-    return bto_status_rcvr(&msg->u.status);
-  default:
-    tor_assert(false);
-  }
-}
-
 /**
  * Create or update a cached ORCONN state for a newly launched
  * connection, including whether it's launched by an origin circuit
  * and whether it's a one-hop circuit.
  **/
 static void
-bto_chan_rcvr(const ocirc_event_msg_t *msg)
+bto_chan_rcvr(const msg_t *msg, const ocirc_chan_msg_t *arg)
 {
   bt_orconn_t *bto;
 
-  /* Ignore other kinds of origin circuit events; we don't need them */
-  if (msg->type != OCIRC_MSGTYPE_CHAN)
-    return;
-
-  bto = bto_find_or_new(0, msg->u.chan.chan);
-  if (!bto->is_orig || (bto->is_onehop && !msg->u.chan.onehop)) {
+  (void)msg;
+  bto = bto_find_or_new(0, arg->chan);
+  if (!bto->is_orig || (bto->is_onehop && !arg->onehop)) {
     log_debug(LD_BTRACK, "ORCONN LAUNCH chan=%"PRIu64" onehop=%d",
-              msg->u.chan.chan, msg->u.chan.onehop);
+              arg->chan, arg->onehop);
   }
   bto->is_orig = true;
-  if (!msg->u.chan.onehop)
+  if (!arg->onehop)
     bto->is_onehop = false;
   bto_update_bests(bto);
 }
@@ -190,12 +180,22 @@ int
 btrack_orconn_init(void)
 {
   bto_init_maps();
-  orconn_event_subscribe(bto_event_rcvr);
-  ocirc_event_subscribe(bto_chan_rcvr);
 
   return 0;
 }
 
+int
+btrack_orconn_add_pubsub(pubsub_connector_t *connector)
+{
+  if (DISPATCH_ADD_SUB(connector, orconn, orconn_state))
+    return -1;
+  if (DISPATCH_ADD_SUB(connector, orconn, orconn_status))
+    return -1;
+  if (DISPATCH_ADD_SUB(connector, ocirc, ocirc_chan))
+    return -1;
+  return 0;
+}
+
 /** Clear the hash maps and reset the "best" states */
 void
 btrack_orconn_fini(void)
diff --git a/src/feature/control/btrack_orconn.h b/src/feature/control/btrack_orconn.h
index f8f5c1096c..07b1b755f3 100644
--- a/src/feature/control/btrack_orconn.h
+++ b/src/feature/control/btrack_orconn.h
@@ -9,6 +9,8 @@
 #ifndef TOR_BTRACK_ORCONN_H
 #define TOR_BTRACK_ORCONN_H
 
+#include "lib/pubsub/pubsub.h"
+
 #ifdef BTRACK_ORCONN_PRIVATE
 
 #include "ht.h"
@@ -33,6 +35,7 @@ typedef struct bt_orconn_t {
 #endif /* defined(BTRACK_ORCONN_PRIVATE) */
 
 int btrack_orconn_init(void);
+int btrack_orconn_add_pubsub(pubsub_connector_t *);
 void btrack_orconn_fini(void);
 
 #endif /* !defined(TOR_BTRACK_ORCONN_H) */