Rework orconn tracking to use pubsub

Part of ticket 29976.

5 files changed, 131 insertions, 47 deletions

diff --git a/src/test/test_btrack.c b/src/test/test_btrack.c
index 48486fb5a1..fef1da4842 100644
--- a/src/test/test_btrack.c
+++ b/src/test/test_btrack.c
@@ -4,6 +4,7 @@
 #include "core/or/or.h"
 
 #include "test/test.h"
+#include "test_helpers.h"
 #include "test/log_test_helpers.h"
 
 #define OCIRC_EVENT_PRIVATE
@@ -12,20 +13,37 @@
 #include "core/or/orconn_event.h"
 
 static void
+send_state(const orconn_state_msg_t *msg_in)
+{
+  orconn_state_msg_t *msg = tor_malloc(sizeof(*msg));
+
+  *msg = *msg_in;
+  orconn_state_publish(msg);
+}
+
+static void
+send_status(const orconn_status_msg_t *msg_in)
+{
+  orconn_status_msg_t *msg = tor_malloc(sizeof(*msg));
+
+  *msg = *msg_in;
+  orconn_status_publish(msg);
+}
+
+static void
 test_btrack_launch(void *arg)
 {
-  orconn_event_msg_t conn;
+  orconn_state_msg_t conn;
   ocirc_event_msg_t circ;
 
   (void)arg;
-  conn.type = ORCONN_MSGTYPE_STATE;
-  conn.u.state.gid = 1;
-  conn.u.state.chan = 1;
-  conn.u.state.proxy_type = PROXY_NONE;
-  conn.u.state.state = OR_CONN_STATE_CONNECTING;
+  conn.gid = 1;
+  conn.chan = 1;
+  conn.proxy_type = PROXY_NONE;
+  conn.state = OR_CONN_STATE_CONNECTING;
 
   setup_full_capture_of_logs(LOG_DEBUG);
-  orconn_event_publish(&conn);
+  send_state(&conn);
   expect_log_msg_containing("ORCONN gid=1 chan=1 proxy_type=0 state=1");
   expect_no_log_msg_containing("ORCONN BEST_");
   teardown_capture_of_logs();
@@ -40,11 +58,11 @@ test_btrack_launch(void *arg)
   expect_log_msg_containing("ORCONN BEST_ANY state -1->1 gid=1");
   teardown_capture_of_logs();
 
-  conn.u.state.gid = 2;
-  conn.u.state.chan = 2;
+  conn.gid = 2;
+  conn.chan = 2;
 
   setup_full_capture_of_logs(LOG_DEBUG);
-  orconn_event_publish(&conn);
+  send_state(&conn);
   expect_log_msg_containing("ORCONN gid=2 chan=2 proxy_type=0 state=1");
   expect_no_log_msg_containing("ORCONN BEST_");
   teardown_capture_of_logs();
@@ -65,27 +83,26 @@ test_btrack_launch(void *arg)
 static void
 test_btrack_delete(void *arg)
 {
-  orconn_event_msg_t conn;
+  orconn_state_msg_t state;
+  orconn_status_msg_t status;
 
   (void)arg;
-  conn.type = ORCONN_MSGTYPE_STATE;
-  conn.u.state.gid = 1;
-  conn.u.state.chan = 1;
-  conn.u.state.proxy_type = PROXY_NONE;
-  conn.u.state.state = OR_CONN_STATE_CONNECTING;
+  state.gid = 1;
+  state.chan = 1;
+  state.proxy_type = PROXY_NONE;
+  state.state = OR_CONN_STATE_CONNECTING;
 
   setup_full_capture_of_logs(LOG_DEBUG);
-  orconn_event_publish(&conn);
+  send_state(&state);
   expect_log_msg_containing("ORCONN gid=1 chan=1 proxy_type=0");
   teardown_capture_of_logs();
 
-  conn.type = ORCONN_MSGTYPE_STATUS;
-  conn.u.status.gid = 1;
-  conn.u.status.status = OR_CONN_EVENT_CLOSED;
-  conn.u.status.reason = 0;
+  status.gid = 1;
+  status.status = OR_CONN_EVENT_CLOSED;
+  status.reason = 0;
 
   setup_full_capture_of_logs(LOG_DEBUG);
-  orconn_event_publish(&conn);
+  send_status(&status);
   expect_log_msg_containing("ORCONN DELETE gid=1 status=3 reason=0");
   teardown_capture_of_logs();
 
@@ -94,7 +111,7 @@ test_btrack_delete(void *arg)
 }
 
 struct testcase_t btrack_tests[] = {
-  { "launch", test_btrack_launch, TT_FORK, 0, NULL },
-  { "delete", test_btrack_delete, TT_FORK, 0, NULL },
+  { "launch", test_btrack_launch, TT_FORK, &helper_pubsub_setup, NULL },
+  { "delete", test_btrack_delete, TT_FORK, &helper_pubsub_setup, NULL },
   END_OF_TESTCASES
 };
diff --git a/src/test/test_controller_events.c b/src/test/test_controller_events.c
index 910aacace3..14fe4fd661 100644
--- a/src/test/test_controller_events.c
+++ b/src/test/test_controller_events.c
@@ -7,6 +7,7 @@
 #define CONTROL_EVENTS_PRIVATE
 #define OCIRC_EVENT_PRIVATE
 #define ORCONN_EVENT_PRIVATE
+#include "app/main/subsysmgr.h"
 #include "core/or/or.h"
 #include "core/or/channel.h"
 #include "core/or/channeltls.h"
@@ -16,6 +17,7 @@
 #include "core/mainloop/connection.h"
 #include "feature/control/control_events.h"
 #include "test/test.h"
+#include "test/test_helpers.h"
 
 #include "core/or/or_circuit_st.h"
 #include "core/or/origin_circuit_st.h"
@@ -394,20 +396,22 @@ test_cntev_dirboot_defer_orconn(void *arg)
 }
 
 static void
-setup_orconn_state(orconn_event_msg_t *msg, uint64_t gid, uint64_t chan,
+setup_orconn_state(orconn_state_msg_t *msg, uint64_t gid, uint64_t chan,
                    int proxy_type)
 {
-  msg->type = ORCONN_MSGTYPE_STATE;
-  msg->u.state.gid = gid;
-  msg->u.state.chan = chan;
-  msg->u.state.proxy_type = proxy_type;
+  msg->gid = gid;
+  msg->chan = chan;
+  msg->proxy_type = proxy_type;
 }
 
 static void
-send_orconn_state(orconn_event_msg_t *msg, uint8_t state)
+send_orconn_state(const orconn_state_msg_t *msg_in, uint8_t state)
 {
-  msg->u.state.state = state;
-  orconn_event_publish(msg);
+  orconn_state_msg_t *msg = tor_malloc(sizeof(*msg));
+
+  *msg = *msg_in;
+  msg->state = state;
+  orconn_state_publish(msg);
 }
 
 static void
@@ -425,7 +429,7 @@ send_ocirc_chan(uint32_t gid, uint64_t chan, bool onehop)
 static void
 test_cntev_orconn_state(void *arg)
 {
-  orconn_event_msg_t conn;
+  orconn_state_msg_t conn;
 
   (void)arg;
   MOCK(queue_control_event_string, mock_queue_control_event_string);
@@ -442,8 +446,8 @@ test_cntev_orconn_state(void *arg)
   send_orconn_state(&conn, OR_CONN_STATE_OPEN);
   assert_bootmsg("15 TAG=handshake_done");
 
-  conn.u.state.gid = 2;
-  conn.u.state.chan = 2;
+  conn.gid = 2;
+  conn.chan = 2;
   send_orconn_state(&conn, OR_CONN_STATE_CONNECTING);
   /* It doesn't know it's an origin circuit yet */
   assert_bootmsg("15 TAG=handshake_done");
@@ -464,7 +468,7 @@ test_cntev_orconn_state(void *arg)
 static void
 test_cntev_orconn_state_pt(void *arg)
 {
-  orconn_event_msg_t conn;
+  orconn_state_msg_t conn;
 
   (void)arg;
   MOCK(queue_control_event_string, mock_queue_control_event_string);
@@ -484,8 +488,8 @@ test_cntev_orconn_state_pt(void *arg)
   assert_bootmsg("15 TAG=handshake_done");
 
   send_ocirc_chan(2, 2, false);
-  conn.u.state.gid = 2;
-  conn.u.state.chan = 2;
+  conn.gid = 2;
+  conn.chan = 2;
   send_orconn_state(&conn, OR_CONN_STATE_CONNECTING);
   assert_bootmsg("76 TAG=ap_conn_pt");
   send_orconn_state(&conn, OR_CONN_STATE_PROXY_HANDSHAKING);
@@ -499,7 +503,7 @@ test_cntev_orconn_state_pt(void *arg)
 static void
 test_cntev_orconn_state_proxy(void *arg)
 {
-  orconn_event_msg_t conn;
+  orconn_state_msg_t conn;
 
   (void)arg;
   MOCK(queue_control_event_string, mock_queue_control_event_string);
@@ -519,8 +523,8 @@ test_cntev_orconn_state_proxy(void *arg)
   assert_bootmsg("15 TAG=handshake_done");
 
   send_ocirc_chan(2, 2, false);
-  conn.u.state.gid = 2;
-  conn.u.state.chan = 2;
+  conn.gid = 2;
+  conn.chan = 2;
   send_orconn_state(&conn, OR_CONN_STATE_CONNECTING);
   assert_bootmsg("78 TAG=ap_conn_proxy");
   send_orconn_state(&conn, OR_CONN_STATE_PROXY_HANDSHAKING);
@@ -534,15 +538,18 @@ test_cntev_orconn_state_proxy(void *arg)
 #define TEST(name, flags)                               \
   { #name, test_cntev_ ## name, flags, 0, NULL }
 
+#define T_PUBSUB(name, setup)                                           \
+  { #name, test_cntev_ ## name, TT_FORK, &helper_pubsub_setup, NULL }
+
 struct testcase_t controller_event_tests[] = {
   TEST(sum_up_cell_stats, TT_FORK),
   TEST(append_cell_stats, TT_FORK),
   TEST(format_cell_stats, TT_FORK),
   TEST(event_mask, TT_FORK),
-  TEST(dirboot_defer_desc, TT_FORK),
-  TEST(dirboot_defer_orconn, TT_FORK),
-  TEST(orconn_state, TT_FORK),
-  TEST(orconn_state_pt, TT_FORK),
-  TEST(orconn_state_proxy, TT_FORK),
+  T_PUBSUB(dirboot_defer_desc, TT_FORK),
+  T_PUBSUB(dirboot_defer_orconn, TT_FORK),
+  T_PUBSUB(orconn_state, TT_FORK),
+  T_PUBSUB(orconn_state_pt, TT_FORK),
+  T_PUBSUB(orconn_state_proxy, TT_FORK),
   END_OF_TESTCASES
 };
diff --git a/src/test/test_extorport.c b/src/test/test_extorport.c
index 38aca90266..cb53a4e662 100644
--- a/src/test/test_extorport.c
+++ b/src/test/test_extorport.c
@@ -587,6 +587,6 @@ struct testcase_t extorport_tests[] = {
   { "cookie_auth", test_ext_or_cookie_auth, TT_FORK, NULL, NULL },
   { "cookie_auth_testvec", test_ext_or_cookie_auth_testvec, TT_FORK,
     NULL, NULL },
-  { "handshake", test_ext_or_handshake, TT_FORK, NULL, NULL },
+  { "handshake", test_ext_or_handshake, TT_FORK, &helper_pubsub_setup, NULL },
   END_OF_TESTCASES
 };
diff --git a/src/test/test_helpers.c b/src/test/test_helpers.c
index 489c257761..e856dc6cca 100644
--- a/src/test/test_helpers.c
+++ b/src/test/test_helpers.c
@@ -17,12 +17,17 @@
 #include "lib/buf/buffers.h"
 #include "app/config/config.h"
 #include "app/config/confparse.h"
+#include "app/main/subsysmgr.h"
 #include "core/mainloop/connection.h"
 #include "lib/crypt_ops/crypto_rand.h"
 #include "core/mainloop/mainloop.h"
 #include "feature/nodelist/nodelist.h"
 #include "core/or/relay.h"
 #include "feature/nodelist/routerlist.h"
+#include "lib/dispatch/dispatch.h"
+#include "lib/dispatch/dispatch_naming.h"
+#include "lib/pubsub/pubsub_build.h"
+#include "lib/pubsub/pubsub_connect.h"
 #include "lib/encoding/confline.h"
 #include "lib/net/resolve.h"
 
@@ -303,3 +308,52 @@ helper_parse_options(const char *conf)
   }
   return opt;
 }
+
+/**
+ * Dispatch alertfn callback: flush all messages right now. Implements
+ * DELIV_IMMEDIATE.
+ **/
+static void
+alertfn_immediate(dispatch_t *d, channel_id_t chan, void *arg)
+{
+  (void) arg;
+  dispatch_flush(d, chan, INT_MAX);
+}
+
+/**
+ * Setup helper for tests that need pubsub active
+ *
+ * Does not hook up mainloop events.  Does set immediate delivery for
+ * all channels.
+ */
+void *
+helper_setup_pubsub(const struct testcase_t *testcase)
+{
+  dispatch_t *dispatcher = NULL;
+  pubsub_builder_t *builder = pubsub_builder_new();
+  channel_id_t chan = get_channel_id("orconn");
+
+  (void)testcase;
+  (void)subsystems_add_pubsub(builder);
+  dispatcher = pubsub_builder_finalize(builder, NULL);
+  tor_assert(dispatcher);
+  dispatch_set_alert_fn(dispatcher, chan, alertfn_immediate, NULL);
+  return dispatcher;
+}
+
+/**
+ * Cleanup helper for tests that need pubsub active
+ */
+int
+helper_cleanup_pubsub(const struct testcase_t *testcase, void *dispatcher_)
+{
+  dispatch_t *dispatcher = dispatcher_;
+
+  (void)testcase;
+  dispatch_free(dispatcher);
+  return 1;
+}
+
+const struct testcase_setup_t helper_pubsub_setup = {
+  helper_setup_pubsub, helper_cleanup_pubsub
+};
diff --git a/src/test/test_helpers.h b/src/test/test_helpers.h
index 9e376a563d..d82072bb34 100644
--- a/src/test/test_helpers.h
+++ b/src/test/test_helpers.h
@@ -7,6 +7,7 @@
 #define BUFFERS_PRIVATE
 
 #include "core/or/or.h"
+#include "tinytest.h"
 
 const char *get_yesterday_date_str(void);
 
@@ -31,5 +32,10 @@ or_options_t *helper_parse_options(const char *conf);
 
 extern const char TEST_DESCRIPTORS[];
 
+void *helper_setup_pubsub(const struct testcase_t *);
+int helper_cleanup_pubsub(const struct testcase_t *, void *);
+
+extern const struct testcase_setup_t helper_pubsub_setup;
+
 #endif /* !defined(TOR_TEST_HELPERS_H) */