aboutsummaryrefslogtreecommitdiff
path: root/src/lib/pubsub
diff options
context:
space:
mode:
authorNick Mathewson <nickm@torproject.org>2019-01-15 09:01:20 -0500
committerNick Mathewson <nickm@torproject.org>2019-03-25 16:35:33 -0400
commit9e60482b8073f2d43187c36c9159fd4367d7140a (patch)
treee73931e80e7f67442f2f865b630d98a338b1b3cb /src/lib/pubsub
parent24b945f713a713bba0ec4f0d8297b49cbc45c5a1 (diff)
downloadtor-9e60482b8073f2d43187c36c9159fd4367d7140a.tar.gz
tor-9e60482b8073f2d43187c36c9159fd4367d7140a.zip
Pubsub: an OO layer on top of lib/dispatch
This "publish/subscribe" layer sits on top of lib/dispatch, and tries to provide more type-safety and cross-checking for the lower-level layer. Even with this commit, we're still not done: more checking will come in the next commit, and a set of usability/typesafety macros will come after.
Diffstat (limited to 'src/lib/pubsub')
-rw-r--r--src/lib/pubsub/.may_include10
-rw-r--r--src/lib/pubsub/include.am24
-rw-r--r--src/lib/pubsub/pub_binding_st.h36
-rw-r--r--src/lib/pubsub/pubsub.h85
-rw-r--r--src/lib/pubsub/pubsub_build.c286
-rw-r--r--src/lib/pubsub/pubsub_build.h87
-rw-r--r--src/lib/pubsub/pubsub_builder_st.h161
-rw-r--r--src/lib/pubsub/pubsub_connect.h47
-rw-r--r--src/lib/pubsub/pubsub_flags.h32
-rw-r--r--src/lib/pubsub/pubsub_publish.c70
-rw-r--r--src/lib/pubsub/pubsub_publish.h15
11 files changed, 853 insertions, 0 deletions
diff --git a/src/lib/pubsub/.may_include b/src/lib/pubsub/.may_include
new file mode 100644
index 0000000000..5623492f00
--- /dev/null
+++ b/src/lib/pubsub/.may_include
@@ -0,0 +1,10 @@
+orconfig.h
+
+lib/cc/*.h
+lib/container/*.h
+lib/dispatch/*.h
+lib/intmath/*.h
+lib/log/*.h
+lib/malloc/*.h
+lib/pubsub/*.h
+lib/string/*.h
diff --git a/src/lib/pubsub/include.am b/src/lib/pubsub/include.am
new file mode 100644
index 0000000000..9856c94a5d
--- /dev/null
+++ b/src/lib/pubsub/include.am
@@ -0,0 +1,24 @@
+
+noinst_LIBRARIES += src/lib/libtor-pubsub.a
+
+if UNITTESTS_ENABLED
+noinst_LIBRARIES += src/lib/libtor-pubsub-testing.a
+endif
+
+src_lib_libtor_pubsub_a_SOURCES = \
+ src/lib/pubsub/pubsub_build.c \
+ src/lib/pubsub/pubsub_publish.c
+
+src_lib_libtor_pubsub_testing_a_SOURCES = \
+ $(src_lib_libtor_pubsub_a_SOURCES)
+src_lib_libtor_pubsub_testing_a_CPPFLAGS = $(AM_CPPFLAGS) $(TEST_CPPFLAGS)
+src_lib_libtor_pubsub_testing_a_CFLAGS = $(AM_CFLAGS) $(TEST_CFLAGS)
+
+noinst_HEADERS += \
+ src/lib/pubsub/pub_binding_st.h \
+ src/lib/pubsub/pubsub.h \
+ src/lib/pubsub/pubsub_build.h \
+ src/lib/pubsub/pubsub_builder_st.h \
+ src/lib/pubsub/pubsub_connect.h \
+ src/lib/pubsub/pubsub_flags.h \
+ src/lib/pubsub/pubsub_publish.h
diff --git a/src/lib/pubsub/pub_binding_st.h b/src/lib/pubsub/pub_binding_st.h
new file mode 100644
index 0000000000..e8c0d47efd
--- /dev/null
+++ b/src/lib/pubsub/pub_binding_st.h
@@ -0,0 +1,36 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_build.h
+ * @brief Declaration of pub_binding_t.
+ */
+
+#ifndef TOR_PUB_BINDING_ST_H
+#define TOR_PUB_BINDING_ST_H
+
+#include "lib/dispatch/msgtypes.h"
+struct dispatch_t;
+
+/**
+ * A pub_binding_t is an opaque object that subsystems use to publish
+ * messages. The DISPATCH_ADD_PUB*() macros set it up.
+ **/
+typedef struct pub_binding_t {
+ /**
+ * A pointer to a configured dispatch_t object. This is filled in
+ * when the dispatch_t is finally constructed.
+ **/
+ struct dispatch_t *dispatch_ptr;
+ /**
+ * A template for the msg_t fields that are filled in for this message.
+ * This is copied into outgoing messages, ensuring that their fields are set
+ * corretly.
+ **/
+ msg_t msg_template;
+} pub_binding_t;
+
+#endif
diff --git a/src/lib/pubsub/pubsub.h b/src/lib/pubsub/pubsub.h
new file mode 100644
index 0000000000..303b36ad5b
--- /dev/null
+++ b/src/lib/pubsub/pubsub.h
@@ -0,0 +1,85 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub.h
+ * @brief Header for OO publish-subscribe functionality.
+ *
+ * This module provides a wrapper around the "dispatch" module,
+ * ensuring type-safety and allowing us to do static analysis on
+ * publication and subscriptions.
+ *
+ * With this module, we enforce:
+ * <ul>
+ * <li>that every message has (potential) publishers and subscribers;
+ * <li>that every message is published and subscribed from the correct
+ * channels, with the correct type ID, every time it is published.
+ * <li>that type IDs correspond to a single C type, and that the C types are
+ * used correctly.
+ * <li>that when a message is published or subscribed, it is done with
+ * a correct subsystem identifier
+ * </ul>
+ *
+ * We do this by making "publication requests" and "subscription requests"
+ * into objects, and doing some computation on them before we create
+ * a dispatch_t with them.
+ *
+ * Rather than using the dispatch module directly, a publishing module
+ * receives a "binding" object that it uses to send messages with the right
+ * settings.
+ */
+
+/*
+ *
+ * Overview: Messages are sent over channels. Before sending a message on a
+ * channel, or receiving a message on a channel, a subsystem needs to register
+ * that it publishes, or subscribes, to that message, on that channel.
+ *
+ * Messages, channels, and subsystems are represented internally as short
+ * integers, though they are associated with human-readable strings for
+ * initialization and debugging.
+ *
+ * When registering for a message, a subsystem must say whether it is an
+ * exclusive publisher/subscriber to that message type, or whether other
+ * subsystems may also publish/subscribe to it.
+ *
+ * All messages and their publishers/subscribers must be registered early in
+ * the initialization process.
+ *
+ * By default, it is an error for a message type to have publishers and no
+ * subscribers on a channel, or subscribers and no publishers on a channel.
+ *
+ * A subsystem may register for a message with a note that delivery or
+ * production is disabled -- for example, because the subsystem is
+ * disabled at compile-time. It is not an error for a message type to
+ * have all of its publishers or subscribers disabled.
+ *
+ * After a message is sent, it is delivered to every recipient. This
+ * delivery happens from the top level of the event loop; it may be
+ * interleaved with network events, timers, etc.
+ *
+ * Messages may have associated data. This data is typed, and is owned
+ * by the message. Strings, byte-arrays, and integers have built-in
+ * support. Other types may be added. If objects are to be sent,
+ * they should be identified by handle. If an object requires cleanup,
+ * it should be declared with an associated free function.
+ *
+ * Semantically, if two subsystems communicate only by this kind of
+ * message passing, neither is considered to depend on the other, though
+ * both are considered to have a dependency on the message and on any
+ * types it contains.
+ *
+ * (Or generational index?)
+ */
+#ifndef TOR_PUBSUB_PUBSUB_H
+#define TOR_PUBSUB_PUBSUB_H
+
+#include "lib/pubsub/pub_binding_st.h"
+#include "lib/pubsub/pubsub_connect.h"
+#include "lib/pubsub/pubsub_flags.h"
+#include "lib/pubsub/pubsub_publish.h"
+
+#endif
diff --git a/src/lib/pubsub/pubsub_build.c b/src/lib/pubsub/pubsub_build.c
new file mode 100644
index 0000000000..72f2eacea8
--- /dev/null
+++ b/src/lib/pubsub/pubsub_build.c
@@ -0,0 +1,286 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_build.c
+ * @brief Construct a dispatch_t in safer, more OO way.
+ **/
+
+#define PUBSUB_PRIVATE
+
+#include "lib/dispatch/dispatch.h"
+#include "lib/dispatch/dispatch_cfg.h"
+#include "lib/dispatch/dispatch_naming.h"
+#include "lib/dispatch/msgtypes.h"
+#include "lib/pubsub/pubsub_flags.h"
+#include "lib/pubsub/pub_binding_st.h"
+#include "lib/pubsub/pubsub_build.h"
+#include "lib/pubsub/pubsub_builder_st.h"
+#include "lib/pubsub/pubsub_connect.h"
+
+#include "lib/container/smartlist.h"
+#include "lib/log/util_bug.h"
+#include "lib/malloc/malloc.h"
+
+ #include <string.h>
+
+/** Construct and return a new empty pubsub_items_t. */
+static pubsub_items_t *
+pubsub_items_new(void)
+{
+ pubsub_items_t *cfg = tor_malloc_zero(sizeof(*cfg));
+ cfg->items = smartlist_new();
+ cfg->type_items = smartlist_new();
+ return cfg;
+}
+
+/** Release all storage held in a pubsub_items_t. */
+void
+pubsub_items_free_(pubsub_items_t *cfg)
+{
+ if (! cfg)
+ return;
+ SMARTLIST_FOREACH(cfg->items, pubsub_cfg_t *, item, tor_free(item));
+ SMARTLIST_FOREACH(cfg->type_items,
+ pubsub_type_cfg_t *, item, tor_free(item));
+ smartlist_free(cfg->items);
+ smartlist_free(cfg->type_items);
+ tor_free(cfg);
+}
+
+/** Construct and return a new pubsub_builder_t. */
+pubsub_builder_t *
+pubsub_builder_new(void)
+{
+ dispatch_naming_init();
+
+ pubsub_builder_t *pb = tor_malloc_zero(sizeof(*pb));
+ pb->cfg = dcfg_new();
+ pb->items = pubsub_items_new();
+ return pb;
+}
+
+/**
+ * Release all storage held by a pubsub_builder_t.
+ *
+ * You'll (mostly) only want to call this function on an error case: if you're
+ * constructing a dispatch_t instead, you should call
+ * pubsub_builder_finalize() to consume the pubsub_builder_t.
+ */
+void
+pubsub_builder_free_(pubsub_builder_t *pb)
+{
+ if (pb == NULL)
+ return;
+ pubsub_items_free(pb->items);
+ dcfg_free(pb->cfg);
+ tor_free(pb);
+}
+
+/**
+ * Create and return a pubsub_connector_t for the subsystem with ID
+ * <b>subsys</b> to use in adding publications, subscriptions, and types to
+ * <b>builder</b>.
+ **/
+pubsub_connector_t *
+pubsub_connector_for_subsystem(pubsub_builder_t *builder,
+ subsys_id_t subsys)
+{
+ tor_assert(builder);
+ ++builder->n_connectors;
+
+ pubsub_connector_t *con = tor_malloc_zero(sizeof(*con));
+
+ con->builder = builder;
+ con->subsys_id = subsys;
+
+ return con;
+}
+
+/**
+ * Release all storage held by a pubsub_connector_t.
+ **/
+void
+pubsub_connector_free_(pubsub_connector_t *con)
+{
+ if (!con)
+ return;
+
+ if (con->builder) {
+ --con->builder->n_connectors;
+ tor_assert(con->builder->n_connectors >= 0);
+ }
+ tor_free(con);
+}
+
+/**
+ * Use <b>con</b> to add a request for being able to publish messages of type
+ * <b>msg</b> with auxiliary data of <b>type</b> on <b>channel</b>.
+ **/
+int
+pubsub_add_pub_(pubsub_connector_t *con,
+ pub_binding_t *out,
+ channel_id_t channel,
+ message_id_t msg,
+ msg_type_id_t type,
+ unsigned flags,
+ const char *file,
+ unsigned line)
+{
+ pubsub_cfg_t *cfg = tor_malloc_zero(sizeof(*cfg));
+
+ memset(out, 0, sizeof(*out));
+ cfg->is_publish = true;
+
+ out->msg_template.sender = cfg->subsys = con->subsys_id;
+ out->msg_template.channel = cfg->channel = channel;
+ out->msg_template.msg = cfg->msg = msg;
+ out->msg_template.type = cfg->type = type;
+
+ cfg->flags = flags;
+ cfg->added_by_file = file;
+ cfg->added_by_line = line;
+
+ /* We're grabbing a pointer to the pub_binding_t so we can tell it about
+ * the dispatcher later on.
+ */
+ cfg->pub_binding = out;
+
+ smartlist_add(con->builder->items->items, cfg);
+
+ if (dcfg_msg_set_type(con->builder->cfg, msg, type) < 0)
+ goto err;
+ if (dcfg_msg_set_chan(con->builder->cfg, msg, channel) < 0)
+ goto err;
+
+ return 0;
+ err:
+ ++con->builder->n_errors;
+ return -1;
+}
+
+/**
+ * Use <b>con</b> to add a request for being able to publish messages of type
+ * <b>msg</b> with auxiliary data of <b>type</b> on <b>channel</b>,
+ * passing them to the callback in <b>recv_fn</b>.
+ **/
+int
+pubsub_add_sub_(pubsub_connector_t *con,
+ recv_fn_t recv_fn,
+ channel_id_t channel,
+ message_id_t msg,
+ msg_type_id_t type,
+ unsigned flags,
+ const char *file,
+ unsigned line)
+{
+ pubsub_cfg_t *cfg = tor_malloc_zero(sizeof(*cfg));
+
+ cfg->is_publish = false;
+ cfg->subsys = con->subsys_id;
+ cfg->channel = channel;
+ cfg->msg = msg;
+ cfg->type = type;
+ cfg->flags = flags;
+ cfg->added_by_file = file;
+ cfg->added_by_line = line;
+
+ cfg->recv_fn = recv_fn;
+
+ smartlist_add(con->builder->items->items, cfg);
+
+ if (dcfg_msg_set_type(con->builder->cfg, msg, type) < 0)
+ goto err;
+ if (dcfg_msg_set_chan(con->builder->cfg, msg, channel) < 0)
+ goto err;
+ if (! (flags & DISP_FLAG_STUB)) {
+ if (dcfg_add_recv(con->builder->cfg, msg, cfg->subsys, recv_fn) < 0)
+ goto err;
+ }
+
+ return 0;
+ err:
+ ++con->builder->n_errors;
+ return -1;
+}
+
+/**
+ * Use <b>con</b> to define a the functions to use for manipulating the type
+ * <b>type</b>. Any function pointers left as NULL will be implemented as
+ * no-ops.
+ **/
+int
+pubsub_connector_define_type_(pubsub_connector_t *con,
+ msg_type_id_t type,
+ dispatch_typefns_t *fns,
+ const char *file,
+ unsigned line)
+{
+ pubsub_type_cfg_t *cfg = tor_malloc_zero(sizeof(*cfg));
+ cfg->type = type;
+ memcpy(&cfg->fns, fns, sizeof(*fns));
+ cfg->subsys = con->subsys_id;
+ cfg->added_by_file = file;
+ cfg->added_by_line = line;
+
+ smartlist_add(con->builder->items->type_items, cfg);
+
+ if (dcfg_type_set_fns(con->builder->cfg, type, fns) < 0)
+ goto err;
+
+ return 0;
+ err:
+ ++con->builder->n_errors;
+ return -1;
+}
+
+/**
+ * Initialize the dispatch_ptr field in every relevant publish binding
+ * for <b>d</b>.
+ */
+static void
+dispatch_fill_pub_binding_backptrs(pubsub_builder_t *builder,
+ dispatch_t *d)
+{
+ SMARTLIST_FOREACH_BEGIN(builder->items->items, pubsub_cfg_t *, cfg) {
+ if (cfg->pub_binding) {
+ // XXXX we could skip this for STUB publishers, and for any publishers
+ // XXXX where all subscribers are STUB.
+ cfg->pub_binding->dispatch_ptr = d;
+ }
+ } SMARTLIST_FOREACH_END(cfg);
+}
+
+/**
+ * Create a new dispatcher as configured in a pubsub_builder_t.
+ *
+ * Consumes and frees its input.
+ **/
+dispatch_t *
+pubsub_builder_finalize(pubsub_builder_t *builder)
+{
+ dispatch_t *dispatcher = NULL;
+ tor_assert_nonfatal(builder->n_connectors == 0);
+
+ if (builder->n_errors)
+ goto err;
+
+ /* Coming in the next commit.
+ if (pubsub_builder_check(builder) < 0)
+ goto err;
+ */
+
+ dispatcher = dispatch_new(builder->cfg);
+
+ if (!dispatcher)
+ goto err;
+
+ dispatch_fill_pub_binding_backptrs(builder, dispatcher);
+
+ err:
+ pubsub_builder_free(builder);
+ return dispatcher;
+}
diff --git a/src/lib/pubsub/pubsub_build.h b/src/lib/pubsub/pubsub_build.h
new file mode 100644
index 0000000000..199eab219a
--- /dev/null
+++ b/src/lib/pubsub/pubsub_build.h
@@ -0,0 +1,87 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_build.h
+ * @brief Header used for constructing the OO publish-subscribe facility.
+ *
+ * (See pubsub.h for more general information on this API.)
+ **/
+
+#ifndef TOR_PUBSUB_BUILD_H
+#define TOR_PUBSUB_BUILD_H
+
+struct dispatch_t;
+
+/**
+ * A "dispatch builder" is an incomplete dispatcher, used when
+ * registering messages. It does not have the same integrity guarantees
+ * as a dispatcher. It cannot actually handle messages itself: once all
+ * subsystems have registered, it is converted into a dispatch_t.
+ **/
+typedef struct pubsub_builder_t pubsub_builder_t;
+
+/**
+ * A "dispatch connector" is a view of the dispatcher that a subsystem
+ * uses while initializing itself. It is specific to the subsystem, and
+ * ensures that each subsystem doesn't need to identify itself
+ * repeatedly while registering its messages.
+ **/
+typedef struct pubsub_connector_t pubsub_connector_t;
+
+/**
+ * Create a new pubsub_builder. This should only happen in the
+ * main-init code.
+ */
+pubsub_builder_t *pubsub_builder_new(void);
+
+/** DOCDOC */
+int pubsub_builder_check(pubsub_builder_t *);
+
+/**
+ * Free a pubsub builder. This should only happen on error paths, where
+ * we have decided not to construct a dispatcher for some reason.
+ */
+#define pubsub_builder_free(db) \
+ FREE_AND_NULL(pubsub_builder_t, pubsub_builder_free_, (db))
+
+/** Internal implementation of pubsub_builder_free(). */
+void pubsub_builder_free_(pubsub_builder_t *);
+
+/**
+ * Create a pubsub connector that a single subsystem will use to
+ * register its messages. The main-init code does this during susbsystem
+ * initialization.
+ */
+pubsub_connector_t *pubsub_connector_for_subsystem(pubsub_builder_t *,
+ subsys_id_t);
+
+/**
+ * The main-init code does this after subsystem initialization.
+ */
+#define pubsub_connector_free(c) \
+ FREE_AND_NULL(pubsub_connector_t, pubsub_connector_free_, (c))
+
+void pubsub_connector_free_(pubsub_connector_t *);
+
+/**
+ * Constructs a dispatcher from a dispatch_builder, after checking that the
+ * invariances on the messages, channels, and connections have been
+ * respected.
+ *
+ * This should happen after every subsystem has initialized, and before
+ * entering the mainloop.
+ */
+struct dispatch_t *pubsub_builder_finalize(pubsub_builder_t *);
+
+#ifdef PUBSUB_PRIVATE
+struct pubsub_items_t;
+#define pubsub_items_free(cfg) \
+ FREE_AND_NULL(pubsub_items_t, pubsub_items_free_, (cfg))
+void pubsub_items_free_(struct pubsub_items_t *cfg);
+#endif
+
+#endif
diff --git a/src/lib/pubsub/pubsub_builder_st.h b/src/lib/pubsub/pubsub_builder_st.h
new file mode 100644
index 0000000000..a1cc6e718b
--- /dev/null
+++ b/src/lib/pubsub/pubsub_builder_st.h
@@ -0,0 +1,161 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_builder_st.h
+ *
+ * @brief private structures used for configuring dispatchers and messages.
+ */
+
+#ifndef TOR_PUBSUB_BUILDER_ST_H
+#define TOR_PUBSUB_BUILDER_ST_H
+
+#ifdef PUBSUB_PRIVATE
+
+#include <stdbool.h>
+#include <stddef.h>
+
+struct dispatch_cfg_t;
+struct smartlist_t;
+struct pub_binding_t;
+
+/**
+ * Configuration for a single publication or subscription request.
+ *
+ * These can be stored while the dispatcher is in use, but are only used for
+ * setup, teardown, and debugging.
+ *
+ * There are various fields in this request describing the message; all of
+ * them must match other descriptions of the message, or a bug has occurred.
+ **/
+typedef struct pubsub_cfg_t {
+ /** True if this is a publishing request; false for a subscribing request. */
+ bool is_publish;
+ /** The system making this request. */
+ subsys_id_t subsys;
+ /** The channel on which the message is to be sent. */
+ channel_id_t channel;
+ /** The message ID to be sent or received. */
+ message_id_t msg;
+ /** The C type associated with the message. */
+ msg_type_id_t type;
+ /** One or more DISP_FLAGS_* items, combined with bitwise OR. */
+ unsigned flags;
+
+ /**
+ * Publishing only: a pub_binding object that will receive the binding for
+ * this request. We will finish filling this in when the dispatcher is
+ * constructed, so that the subsystem can publish then and not before.
+ */
+ struct pub_binding_t *pub_binding;
+
+ /**
+ * Subscribing only: a function to receive message objects for this request.
+ */
+ recv_fn_t recv_fn;
+
+ /** The file from which this message was configured */
+ const char *added_by_file;
+ /** The line at which this message was configured */
+ unsigned added_by_line;
+} pubsub_cfg_t;
+
+/**
+ * Configuration request for a single C type.
+ *
+ * These are stored while the dispatcher is in use, but are only used for
+ * setup, teardown, and debugging.
+ **/
+typedef struct pubsub_type_cfg_t {
+ /**
+ * The identifier for this type.
+ */
+ msg_type_id_t type;
+ /**
+ * Functions to use when manipulating the type.
+ */
+ dispatch_typefns_t fns;
+
+ /** The subsystem that configured this type. */
+ subsys_id_t subsys;
+ /** The file from which this type was configured */
+ const char *added_by_file;
+ /** The line at which this type was configured */
+ unsigned added_by_line;
+} pubsub_type_cfg_t;
+
+/**
+ * The set of configuration requests for a dispatcher, as made by various
+ * subsystems.
+ **/
+typedef struct pubsub_items_t {
+ /** List of pubsub_cfg_t. */
+ struct smartlist_t *items;
+ /** List of pubsub_type_cfg_t. */
+ struct smartlist_t *type_items;
+} pubsub_items_t;
+
+/**
+ * Type used to construct a dispatcher. We use this type to build up the
+ * configuration for a dispatcher, and then pass ownership of that
+ * configuration to the newly constructed dispatcher.
+ **/
+struct pubsub_builder_t {
+ /** Number of outstanding pubsub_connector_t objects pointing to this
+ * pubsub_builder_t. */
+ int n_connectors;
+ /** Number of errors encountered while constructing this object so far. */
+ int n_errors;
+ /** In-progress configuration that we're constructing, as a list of the
+ * requests that have been made. */
+ pubsub_items_t *items;
+ /** In-progress configuration that we're constructing, in a form that can
+ * be converted to a dispatch_t. */
+ struct dispatch_cfg_t *cfg;
+};
+
+/**
+ * Type given to a subsystem when adding connections to a pubsub_builder_t.
+ * We use this type to force each subsystem to get blamed for the
+ * publications, subscriptions, and types that it adds.
+ **/
+struct pubsub_connector_t {
+ /** The pubsub_builder that this connector refers to. */
+ struct pubsub_builder_t *builder;
+ /** The subsystem that has been given this connector. */
+ subsys_id_t subsys_id;
+};
+
+/**
+ * Helper structure used when constructing a dispatcher that sorts the
+ * pubsub_cfg_t objects in various ways.
+ **/
+typedef struct pubsub_adjmap_t {
+ /* XXXX The next three fields are currently constructed but not yet
+ * XXXX used. I believe we'll want them in the future, though. -nickm
+ */
+ /** Number of subsystems; length of the *_by_subsys arrays. */
+ size_t n_subsystems;
+ /** Array of lists of publisher pubsub_cfg_t objects, indexed by
+ * subsystem. */
+ struct smartlist_t **pub_by_subsys;
+ /** Array of lists of subscriber pubsub_cfg_t objects, indexed by
+ * subsystem. */
+ struct smartlist_t **sub_by_subsys;
+
+ /** Number of message IDs; length of the *_by_msg arrays. */
+ size_t n_msgs;
+ /** Array of lists of publisher pubsub_cfg_t objects, indexed by
+ * message ID. */
+ struct smartlist_t **pub_by_msg;
+ /** Array of lists of subscriber pubsub_cfg_t objects, indexed by
+ * message ID. */
+ struct smartlist_t **sub_by_msg;
+} pubsub_adjmap_t;
+
+#endif
+
+#endif
diff --git a/src/lib/pubsub/pubsub_connect.h b/src/lib/pubsub/pubsub_connect.h
new file mode 100644
index 0000000000..b63f9dc438
--- /dev/null
+++ b/src/lib/pubsub/pubsub_connect.h
@@ -0,0 +1,47 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_connect.h
+ * @brief Header for functions that add relationships to a pubsub builder.
+ *
+ * These functions are used by modules that need to add publication and
+ * subscription requests.
+ **/
+
+#ifndef TOR_PUBSUB_CONNECT_H
+#define TOR_PUBSUB_CONNECT_H
+
+#include "lib/dispatch/msgtypes.h"
+
+struct pub_binding_t;
+struct pubsub_connector_t;
+
+int pubsub_add_pub_(struct pubsub_connector_t *con,
+ struct pub_binding_t *out,
+ channel_id_t channel,
+ message_id_t msg,
+ msg_type_id_t type,
+ unsigned flags,
+ const char *file,
+ unsigned line);
+
+int pubsub_add_sub_(struct pubsub_connector_t *con,
+ recv_fn_t recv_fn,
+ channel_id_t channel,
+ message_id_t msg,
+ msg_type_id_t type,
+ unsigned flags,
+ const char *file,
+ unsigned line);
+
+int pubsub_connector_define_type_(struct pubsub_connector_t *,
+ msg_type_id_t,
+ dispatch_typefns_t *,
+ const char *file,
+ unsigned line);
+
+#endif
diff --git a/src/lib/pubsub/pubsub_flags.h b/src/lib/pubsub/pubsub_flags.h
new file mode 100644
index 0000000000..d07a06be7b
--- /dev/null
+++ b/src/lib/pubsub/pubsub_flags.h
@@ -0,0 +1,32 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_flags.h
+ * @brief Flags that can be set on publish/subscribe messages.
+ **/
+
+#ifndef TOR_PUBSUB_FLAGS_H
+#define TOR_PUBSUB_FLAGS_H
+
+/**
+ * Flag for registering a message: declare that no other module is allowed to
+ * publish this message if we are publishing it, or subscribe to it if we are
+ * subscribing to it.
+ */
+#define DISP_FLAG_EXCL (1u<<0)
+
+/**
+ * Flag for registering a message: declare that this message is a stub, and we
+ * will not actually publish/subscribe it, but that the dispatcher should
+ * treat us as if we did when typechecking.
+ *
+ * We use this so that messages aren't treated as "dangling" if they are
+ * potentially used by some other build of Tor.
+ */
+#define DISP_FLAG_STUB (1u<<1)
+
+#endif
diff --git a/src/lib/pubsub/pubsub_publish.c b/src/lib/pubsub/pubsub_publish.c
new file mode 100644
index 0000000000..8c469e8add
--- /dev/null
+++ b/src/lib/pubsub/pubsub_publish.c
@@ -0,0 +1,70 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * @file pubsub_publish.h
+ * @brief Header for functions to publish using a pub_binding_t.
+ **/
+
+#define PUBSUB_PRIVATE
+#define DISPATCH_PRIVATE
+#include "orconfig.h"
+
+#include "lib/dispatch/dispatch.h"
+#include "lib/dispatch/dispatch_st.h"
+
+#include "lib/pubsub/pub_binding_st.h"
+#include "lib/pubsub/pubsub_publish.h"
+
+#include "lib/malloc/malloc.h"
+#include "lib/log/util_bug.h"
+
+#include <string.h>
+
+/**
+ * Publish a message from the publication binding <b>pub</b> using the
+ * auxiliary data <b>auxdata</b>.
+ *
+ * Return 0 on success, -1 on failure.
+ **/
+int
+pubsub_pub_(const pub_binding_t *pub, msg_aux_data_t auxdata)
+{
+ dispatch_t *d = pub->dispatch_ptr;
+ if (BUG(! d)) {
+ /* Tried to publish a message before the dispatcher was configured. */
+ /* (Without a dispatcher, we don't know how to free auxdata.) */
+ return -1;
+ }
+
+ if (BUG(pub->msg_template.type >= d->n_types)) {
+ /* The type associated with this message is not known to the dispatcher. */
+ /* (Without a correct type, we don't know how to free auxdata.) */
+ return -1;
+ }
+
+ if (BUG(pub->msg_template.msg >= d->n_msgs) ||
+ BUG(pub->msg_template.channel >= d->n_queues)) {
+ /* The message ID or channel ID was out of bounds. */
+ d->typefns[pub->msg_template.type].free_fn(auxdata);
+ return -1;
+ }
+
+ if (! d->table[pub->msg_template.msg]) {
+ /* Fast path: nobody wants this data. */
+
+ // XXXX Faster path: we could store this in the pub_binding_t.
+ d->typefns[pub->msg_template.type].free_fn(auxdata);
+ return 0;
+ }
+
+ /* Construct the message object */
+ msg_t *m = tor_malloc(sizeof(msg_t));
+ memcpy(m, &pub->msg_template, sizeof(msg_t));
+ m->aux_data__ = auxdata;
+
+ return dispatch_send_msg_unchecked(d, m);
+}
diff --git a/src/lib/pubsub/pubsub_publish.h b/src/lib/pubsub/pubsub_publish.h
new file mode 100644
index 0000000000..0250fd0760
--- /dev/null
+++ b/src/lib/pubsub/pubsub_publish.h
@@ -0,0 +1,15 @@
+/* Copyright (c) 2001, Matej Pfajfar.
+ * Copyright (c) 2001-2004, Roger Dingledine.
+ * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
+ * Copyright (c) 2007-2018, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#ifndef TOR_PUBSUB_PUBLISH_H
+#define TOR_PUBSUB_PUBLISH_H
+
+#include "lib/dispatch/msgtypes.h"
+struct pub_binding_t;
+
+int pubsub_pub_(const struct pub_binding_t *pub, msg_aux_data_t auxdata);
+
+#endif