Add new ALM transport

.cirrus.yml

@@ -108,7 +108,7 @@ freebsd_task:
 memcheck_task:
   container:
     # Just uses a recent/common distro to run memory error/leak checks.
-    dockerfile: ci/ubuntu-18.04/Dockerfile
+    dockerfile: ci/debian-9/Dockerfile
     cpu: 8
     # AddressSanitizer uses more memory than normal config.
     memory: 8GB

CMakeLists.txt

@@ -303,17 +303,19 @@ set(BROKER_SRC
   src/detail/prefix_matcher.cc
   src/detail/sqlite_backend.cc
   src/detail/store_actor.cc
+  src/domain_options.cc
   src/endpoint.cc
   src/endpoint_info.cc
+  src/entity_id.cc
   src/error.cc
   src/filter_type.cc
+  src/gateway.cc
   src/internal_command.cc
   src/mailbox.cc
   src/network_info.cc
   src/peer_status.cc
   src/port.cc
   src/publisher.cc
-  src/publisher_id.cc
   src/status.cc
   src/status_subscriber.cc
   src/store.cc
@@ -360,8 +362,9 @@ macro(add_tool name)
 endmacro()
 
 if (NOT BROKER_DISABLE_TOOLS)
-  add_tool(broker-pipe)
+  add_tool(broker-gateway)
   add_tool(broker-node)
+  add_tool(broker-pipe)
 endif ()
 
 # -- Bindings -----------------------------------------------------------------

bindings/python/_broker.cpp

@@ -51,6 +51,18 @@ PYBIND11_MAKE_OPAQUE(broker::set)
 PYBIND11_MAKE_OPAQUE(broker::table)
 PYBIND11_MAKE_OPAQUE(broker::vector)
 
+namespace {
+
+broker::endpoint_id node_from_str(const std::string& node_str) {
+  caf::node_id node;
+  if (auto err = caf::parse(node_str, node))
+    throw std::invalid_argument(
+      "endpoint::await_peer called with invalid endpoint ID");
+  return node;
+};
+
+} // namespace
+
 PYBIND11_MODULE(_broker, m) {
   m.doc() = "Broker python bindings";
   py::module mb = m.def_submodule("zeek", "Zeek-specific bindings");
@@ -124,7 +136,8 @@ PYBIND11_MODULE(_broker, m) {
     .def("drop_all_on_destruction", &broker::publisher::drop_all_on_destruction)
     .def("publish", (void (broker::publisher::*)(broker::data d)) &broker::publisher::publish)
     .def("publish_batch",
-       [](broker::publisher& p, std::vector<broker::data> xs) { p.publish(xs); });
+       [](broker::publisher& p, std::vector<broker::data> xs) { p.publish(xs); })
+    .def("reset", &broker::publisher::reset);
 
   using subscriber_base = broker::subscriber_base<broker::subscriber::value_type>;
   using topic_data_pair = std::pair<broker::topic, broker::data>;
@@ -190,7 +203,8 @@ PYBIND11_MODULE(_broker, m) {
 
   py::class_<broker::subscriber, subscriber_base>(m, "Subscriber")
     .def("add_topic", &broker::subscriber::add_topic)
-    .def("remove_topic", &broker::subscriber::remove_topic);
+    .def("remove_topic", &broker::subscriber::remove_topic)
+    .def("reset", &broker::subscriber::reset);
 
   py::bind_vector<std::vector<broker::status_subscriber::value_type>>(m, "VectorStatusSubscriberValueType");
 
@@ -222,7 +236,8 @@ PYBIND11_MODULE(_broker, m) {
          [](broker::status_subscriber& ep) -> std::vector<broker::status_subscriber::value_type> {
 	   return ep.poll(); })
     .def("available", &broker::status_subscriber::available)
-    .def("fd", &broker::status_subscriber::fd);
+    .def("fd", &broker::status_subscriber::fd)
+    .def("reset", &broker::status_subscriber::reset);
 
   py::class_<broker::status_subscriber::value_type>(status_subscriber, "ValueType")
     .def("is_error",
@@ -239,8 +254,7 @@ PYBIND11_MODULE(_broker, m) {
   py::class_<broker::broker_options>(m, "BrokerOptions")
     .def(py::init<>())
     .def_readwrite("disable_ssl", &broker::broker_options::disable_ssl)
-    .def_readwrite("ttl", &broker::broker_options::ttl)
-    .def_readwrite("forward", &broker::broker_options::forward)
+    .def_readwrite("disable_forwarding", &broker::broker_options::disable_forwarding)
     .def_readwrite("ignore_broker_conf", &broker::broker_options::ignore_broker_conf)
     .def_readwrite("use_real_time", &broker::broker_options::use_real_time);
 
@@ -324,6 +338,14 @@ PYBIND11_MODULE(_broker, m) {
          [](broker::endpoint& ep, const std::string& name) -> broker::expected<broker::store> {
 	        return ep.attach_clone(name);
 	    })
+    .def("await_peer",
+         [](broker::endpoint& ep, const std::string& node_str) {
+           return ep.await_peer(node_from_str(node_str));
+         })
+    .def("await_peer",
+         [](broker::endpoint& ep, const std::string& node_str, broker::timespan timeout) {
+           return ep.await_peer(node_from_str(node_str), timeout);
+         })
    ;
 }
 

bindings/python/broker/__init__.py

@@ -121,6 +121,13 @@ class Subscriber:
     def __init__(self, internal_subscriber):
         self._subscriber = internal_subscriber
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self._subscriber.reset()
+        self._subscriber = None
+
     def get(self, *args, **kwargs):
         msg = self._subscriber.get(*args, **kwargs)
 
@@ -161,6 +168,13 @@ class StatusSubscriber(_broker.Subscriber):
     def __init__(self, internal_subscriber):
         self._subscriber = internal_subscriber
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self._subscriber.reset()
+        self._subscriber = None
+
     def get(self, *args, **kwargs):
         x = self._subscriber.get(*args, **kwargs)
         return self._to_result(x)
@@ -198,6 +212,13 @@ class Publisher:
     def __init__(self, internal_publisher):
         self._publisher = internal_publisher
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self._publisher.reset()
+        self._publisher = None
+
     def demand(self):
         return self._publisher.demand()
 
@@ -226,12 +247,20 @@ def publish_batch(self, *batch):
 
 class Store:
     # This class does not derive from the internal class because we
-    # need to pass in existign instances. That means we need to
+    # need to pass in existing instances. That means we need to
     # wrap all methods, even those that just reuse the internal
     # implementation.
     def __init__(self, internal_store):
         self._store = internal_store
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self._store.reset()
+        self._parent = None
+        self._store = None
+
     def name(self):
         return self._store.name()
 
@@ -328,6 +357,16 @@ def pop(self, key, expiry=None):
     def _to_expiry(self, e):
         return (_broker.OptionalTimespan(_broker.Timespan(float(e))) if e is not None else _broker.OptionalTimespan())
 
+    def await_idle(self, timeout=None):
+        if timeout:
+            return self._store.await_idle(_broker.Timespan(float(timeout)))
+        else:
+            return self._store.await_idle()
+
+    # Points to the "owning" Endpoint to make sure Python cleans this object up
+    # before destroying the endpoint.
+    _parent = None
+
 class Endpoint(_broker.Endpoint):
     def make_subscriber(self, topics, qsize = 20):
         topics = _make_topics(topics)
@@ -360,13 +399,36 @@ def attach_master(self, name, type=None, opts={}):
         bopts = _broker.MapBackendOptions() # Generator expression doesn't work here.
         for (k, v) in opts.items():
             bopts[k] = Data.from_py(v)
-
         s = _broker.Endpoint.attach_master(self, name, type, bopts)
-        return Store(s.get()) if s.is_valid() else None
+        if not s.is_valid():
+            return None
+        result = Store(s.get())
+        # This extra reference establishes a parent-child relation between the
+        # two Python objects, making sure that the garbage collector destroys
+        # the store *before* cleaning up the endpoint
+        result._parent = self
+        return result
 
     def attach_clone(self, name):
         s = _broker.Endpoint.attach_clone(self, name)
-        return Store(s.get()) if s.is_valid() else None
+        if not s.is_valid():
+            return None
+        result = Store(s.get())
+        # Same as above: make sure Python cleans up the store first.
+        result._parent = self
+        return result
+
+    def await_peer(self, node, timeout=None):
+        if timeout:
+            return  _broker.Endpoint.await_peer(self, node, _broker.Timespan(float(timeout)))
+        else:
+            return  _broker.Endpoint.await_peer(self, node)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.shutdown()
 
 class Message:
     def to_broker(self):
@@ -505,56 +567,56 @@ def _try_bytes_decode(b):
 # class Store:
 #   def __init__(self, handle):
 #     self.store = handle
-# 
+#
 #   def name(self):
 #     return self.store.name()
-# 
+#
 # class Mailbox:
 #   def __init__(self, handle):
 #     self.mailbox = handle
-# 
+#
 #   def descriptor(self):
 #     return self.mailbox.descriptor()
-# 
+#
 #   def empty(self):
 #     return self.mailbox.empty()
-# 
+#
 #   def count(self, n = -1):
 #     return self.mailbox.count(n)
-# 
-# 
+#
+#
 # class Message:
 #   def __init__(self, handle):
 #     self.message = handle
-# 
+#
 #   def topic(self):
 #     return self.message.topic().string()
-# 
+#
 #   def data(self):
 #     return self.message.data() # TODO: unwrap properly
-# 
+#
 #   def __str__(self):
 #     return "%s -> %s" % (self.topic(), str(self.data()))
-# 
-# 
+#
+#
 # class BlockingEndpoint(Endpoint):
 #   def __init__(self, handle):
 #     super(BlockingEndpoint, self).__init__(handle)
-# 
+#
 #   def subscribe(self, topic):
 #     self.endpoint.subscribe(topic)
-# 
+#
 #   def unsubscribe(self, topic):
 #     self.endpoint.unsubscribe(topic)
-# 
+#
 #   def receive(self, x):
 #     if x == Status:
 #       return self.endpoint.receive()
 #     elif x == Message:
 #       return Message(self.endpoint.receive())
 #     else:
 #       raise BrokerError("invalid receive type")
-# 
+#
 #   #def receive(self):
 #   #  if fun1 is None:
 #   #    return Message(self.endpoint.receive())
@@ -565,34 +627,34 @@ def _try_bytes_decode(b):
 #   #      return self.endpoint.receive_msg(fun1)
 #   #    raise BrokerError("invalid receive callback arity; must be 1 or 2")
 #   #  return self.endpoint.receive_msg_or_status(fun1, fun2)
-# 
+#
 #   def mailbox(self):
 #     return Mailbox(self.endpoint.mailbox())
-# 
-# 
+#
+#
 # class NonblockingEndpoint(Endpoint):
 #   def __init__(self, handle):
 #     super(NonblockingEndpoint, self).__init__(handle)
-# 
+#
 #   def subscribe(self, topic, fun):
 #     self.endpoint.subscribe_msg(topic, fun)
-# 
+#
 #   def on_status(fun):
 #     self.endpoint.subscribe_status(fun)
-# 
+#
 #   def unsubscribe(self, topic):
 #     self.endpoint.unsubscribe(topic)
-# 
-# 
+#
+#
 # class Context:
 #   def __init__(self):
 #     self.context = _broker.Context()
-# 
+#
 #   def spawn(self, api):
 #     if api == Blocking:
 #       return BlockingEndpoint(self.context.spawn_blocking())
 #     elif api == Nonblocking:
 #       return NonblockingEndpoint(self.context.spawn_nonblocking())
 #     else:
 #       raise BrokerError("invalid API flag: " + str(api))
-# 
+#

bindings/python/store.cpp

@@ -48,11 +48,14 @@ void init_store(py::module& m) {
     .def("increment", &broker::store::increment)
     .def("decrement", &broker::store::decrement)
     .def("append", &broker::store::append)
-    .def("insert_into", (void (broker::store::*)(broker::data, broker::data, broker::optional<broker::timespan>) const) &broker::store::insert_into)
-    .def("insert_into", (void (broker::store::*)(broker::data, broker::data, broker::data, broker::optional<broker::timespan>) const) &broker::store::insert_into)
+    .def("insert_into", (void (broker::store::*)(broker::data, broker::data, broker::optional<broker::timespan>)) &broker::store::insert_into)
+    .def("insert_into", (void (broker::store::*)(broker::data, broker::data, broker::data, broker::optional<broker::timespan>)) &broker::store::insert_into)
     .def("remove_from", &broker::store::remove_from)
     .def("push", &broker::store::push)
-    .def("pop", &broker::store::pop);
+    .def("pop", &broker::store::pop)
+    .def("await_idle", [](broker::store& st) { return st.await_idle(); })
+    .def("await_idle", [](broker::store& st, broker::timespan timeout) { return st.await_idle(timeout); })
+    .def("reset", &broker::store::reset);
 
 // Don't need.
 //  py::class_<broker::store::response>(store, "Response")

doc/_images/gateway.drawio

@@ -0,0 +1 @@
+<mxfile modified="2020-07-12T13:25:38.874Z" host="Electron" agent="5.0 (Macintosh; Intel Mac OS X 10_15_5) AppleWebKit/537.36 (KHTML, like Gecko) draw.io/13.3.9 Chrome/83.0.4103.119 Electron/9.0.5 Safari/537.36" etag="LaupmYtKdVszVeVFEYXr" version="13.3.9" type="device"><diagram id="l8hXMBHkgcEJcSW0mbfh" name="Page-1">7Vpbc5s4FP41fqwHEGB4bBI7bbY7m2l6ye5LRwEZtMHII+TY7q+vZCRuAsdtcEw9fTI6knX5vqPvII5G4HKxuaZwGf9NQpSMLCPcjMDVyLJM03D5j7BspcUwQG6JKA6lrTTc4e9INZTWFQ5RVmvICEkYXtaNAUlTFLCaDVJK1vVmc5LUR13CCGmGuwAmuvUrDlmsFub6ZcU7hKNYDu1Zk7xiAVVjuZIshiFZV0xgOgKXlBCWPy02lygR6Clc8v/NOmqLiVGUskP+YN8sP4ZfP7ro8T68RgT67P6/N0B28wSTlVzxdMMQTWEiZ822CgreHUedFy74UpbCGCRkxXu/WMeYobslDIRxzT2B22K2SHjJ5I8RhSHms7wkCaG7vsBs5vsA8Lo5ThJlT0kqep/zYd7Jvz4hyjBn422Co5TbGBF9Q1lK0JyJP5CU3cl5Gqqce5Lp8rKOlFo17x1tKiaJHIdngRjd8iaq1pYsSj9WpK5LnwDKFlfcwfKkEUo/jIquS6r4g2TrZ5gDGnPv0xMzp3ihOQAd7L0CXY4xNLomGlt7WHqGFZgtc6mb4w0KG1RwikKIvHkg2GaUPKJKjRt46GHeD8Yc5BrGxRapgGy3YGwfC2JfRzTkWi6L0kUpWaWhAG3nehVQ0Qaze2kWz/+K57EjSimf3b1qJgrVuvCtiDLlANwyw2Liu57+R4xtpXPDFSPcRCiLSUT4Pv1AxH4oRr/aVIa/2qqCGLCs2pVUXSeLGVnRAO3BylSxFNIIsT0NLRnABZR7vYKiBDL8VA+bvXOsXgt6JNnskcZXIGQyLD7094ffXddAI3Z4J5Y10zo7iK0GxKZ/aozBkWSlNXL8xnFjcqBKKacdikzZZ7eHgDm0PeRokD67YczKhjGqG8YYHe1V6wUx+lDvV0eLoXi/24u6mV3qZhxL3V6BKtMbFlXe2QmVaw9NqPo5KNa0ayAKpYTneYVyBuX2at5n7Pan9vo2RI/3ijuU/WAfGgbsYe2H/s8jxhmRZQ6LLOfsxGtiDUy83LODuPkN5OQQ+wdAXNGgcvNPS2tDkloFokVIqtL1jAzlnKhUrNXG0mzmGcbLPnNYBx8fhvU5FhzyHvWHxI7AMxQS9Y9V15ChNdxqXHKtYXXo60CqjHp3qjZAImk80nO1CxyGSZea1j2kuNmwKzwiFsSjo6V27eZx0tCF02sRTutox0k9tfv+pMFJJe2vMOVdYSLYXKOMjfbn83vgxnEHlhK29K00HRw3CL4GN2Bw3BxwJYKsWIJTjqS63CVEJYRZXGiPkq0P8AEltyTDEtQHwhhZtOha/QZRIX9VCZV3ZBabSFxqG8N1BsbBKuP9IfotklKssT9zPIdLt06kVNxulTyGKrbceGlTxT7uu9CVsf7sgpt/bi4+LfCt8SX8y35jalz+1AG218NoI64fOfNlv/DdQf71lmA+xfK44Dtjz5l4rg0mruV6vt3YzfbYtC0TOK4FJqC4d6kGyJcn+yyp1YdpXuqZNBwkX7/W0c5HitX+utvYL3QblZEoL+70nnHdk1ctvrvkw1c/k3XmaXt0UXVKPjA727uPAmO/jzo9+WgzkjVF7Jd9lBfL+7h58/JaM5j+AA==</diagram></mxfile>