Merge branch 'issue/gh-318'

* issue/gh-318:
  Remove obsolete code
  CI: Use a python venv in test.sh to avoid pip errors on newer platforms
  CI: Pin Windows openssl to 3.1.1
  CI: Fix building of alpine image with regards to python packages
  Implement new variant type and builder API

CHANGES

@@ -1,3 +1,31 @@
+2.8.0-dev.41 | 2024-02-20 09:15:14 -0700
+
+  * Remove obsolete code (Dominik Charousset, Corelight)
+
+  * CI: Use a python venv in test.sh to avoid pip errors on newer platforms (Tim Wojtulewicz, Corelight)
+
+  * CI: Pin Windows openssl to 3.1.1 (Tim Wojtulewicz, Corelight)
+
+  * CI: Fix building of alpine image with regards to python packages (Tim Wojtulewicz, Corelight)
+
+  * Implement new variant type and builder API (Dominik Charousset, Corelight)
+
+    - The new `broker::variant` is similar to `broker::data` in so far that
+      it can represent similar types. However, once constructed, the new
+      `broker::variant` is a read-only type. Internally, the variant stores
+      the binary representation of the represented value. This makes
+      serializing into the binary format essentially a no-op. When
+      deserializing a value from the wire, we also keep the original byte
+      representation around. For the actual data structure on top of the
+      buffer, `broker::variant` uses a monotonic buffer resource in order to
+      efficiently allocate the data structure representing the value.
+    - For building a `broker::variant`, Broker now ships a new builder API.
+      A builder must be filled with the entire data set for a
+      `broker::variant` before constructing the final object.
+    - The new `broker::variant` also comes with a matching "envelope" type.
+      An envelope is basically the raw data plus meta data (such as topic).
+      The envelopes represent the old message types such as `data_message`.
+
 2.8.0-dev.35 | 2024-01-30 18:31:39 +0100
 
   * Implement JSON encoding for broker::data (Dominik Charousset, Corelight)

CMakeLists.txt

@@ -292,9 +292,12 @@ set(BROKER_SRC
   src/address.cc
   src/alm/multipath.cc
   src/alm/routing_table.cc
+  src/builder.cc
+  src/command_envelope.cc
   src/configuration.cc
   src/convert.cc
   src/data.cc
+  src/data_envelope.cc
   src/detail/abstract_backend.cc
   src/detail/filesystem.cc
   src/detail/flare.cc
@@ -313,6 +316,7 @@ set(BROKER_SRC
   src/endpoint_id.cc
   src/endpoint_info.cc
   src/entity_id.cc
+  src/envelope.cc
   src/error.cc
   src/filter_type.cc
   src/format/bin.cc
@@ -322,6 +326,7 @@ set(BROKER_SRC
   src/internal/connector_adapter.cc
   src/internal/core_actor.cc
   src/internal/flare_actor.cc
+  src/internal/json.cc
   src/internal/json_client.cc
   src/internal/json_type_mapper.cc
   src/internal/master_actor.cc
@@ -340,11 +345,14 @@ set(BROKER_SRC
   src/internal/wire_format.cc
   src/internal_command.cc
   src/mailbox.cc
-  src/message.cc
   src/network_info.cc
+  src/p2p_message_type.cc
   src/peer_status.cc
+  src/ping_envelope.cc
+  src/pong_envelope.cc
   src/port.cc
   src/publisher.cc
+  src/routing_update_envelope.cc
   src/shutdown_options.cc
   src/status.cc
   src/status_subscriber.cc
@@ -360,6 +368,12 @@ set(BROKER_SRC
   src/telemetry/metric_registry_impl.cc
   src/time.cc
   src/topic.cc
+  src/variant.cc
+  src/variant_data.cc
+  src/variant_list.cc
+  src/variant_set.cc
+  src/variant_table.cc
+  src/variant_tag.cc
   src/version.cc
   src/worker.cc
   src/zeek.cc

VERSION

@@ -1 +1 @@
-2.8.0-dev.35
+2.8.0-dev.41

bindings/python/_broker.cpp

@@ -25,6 +25,7 @@
 #include "broker/data.hh"
 #include "broker/endpoint.hh"
 #include "broker/endpoint_info.hh"
+#include "broker/message.hh"
 #include "broker/network_info.hh"
 #include "broker/peer_flags.hh"
 #include "broker/peer_info.hh"
@@ -173,7 +174,7 @@ PYBIND11_MODULE(_broker, m) {
     .def("capacity", &broker::publisher::capacity)
     .def("fd", &broker::publisher::fd)
     .def("drop_all_on_destruction", &broker::publisher::drop_all_on_destruction)
-    .def("publish", (void(broker::publisher::*)(broker::data d))
+    .def("publish", (void(broker::publisher::*)(const broker::data&))
                       & broker::publisher::publish)
     .def("publish_batch", [](broker::publisher& p,
                              std::vector<broker::data> xs) { p.publish(xs); })
@@ -196,18 +197,18 @@ PYBIND11_MODULE(_broker, m) {
     .def("get",
          [](broker::subscriber& ep) -> topic_data_pair {
            auto res = ep.get();
-           return std::make_pair(broker::get_topic(res), broker::get_data(res));
+           return std::make_pair(broker::topic{broker::get_topic(res)},
+                                 broker::get_data(res).to_data());
          })
 
     .def("get",
          [](broker::subscriber& ep,
             double secs) -> std::optional<topic_data_pair> {
-           auto res = ep.get(broker::to_duration(secs));
            std::optional<topic_data_pair> rval;
-           if (res) {
-             auto p = std::make_pair(broker::get_topic(*res),
-                                     broker::get_data(*res));
-             rval = std::optional<topic_data_pair>(std::move(p));
+           if (auto res = ep.get(broker::to_duration(secs))) {
+             rval.emplace();
+             rval->first = broker::get_topic(*res);
+             rval->second = broker::get_data(*res).to_data();
            }
            return rval;
          })
@@ -219,8 +220,8 @@ PYBIND11_MODULE(_broker, m) {
            std::vector<topic_data_pair> rval;
            rval.reserve(res.size());
            for (auto& e : res)
-             rval.emplace_back(
-               std::make_pair(broker::get_topic(e), broker::get_data(e)));
+             rval.emplace_back(broker::topic{broker::get_topic(e)},
+                               broker::get_data(e).to_data());
            return rval;
          })
 
@@ -231,8 +232,8 @@ PYBIND11_MODULE(_broker, m) {
            std::vector<topic_data_pair> rval;
            rval.reserve(res.size());
            for (auto& e : res)
-             rval.emplace_back(
-               std::make_pair(broker::get_topic(e), broker::get_data(e)));
+             rval.emplace_back(broker::topic{broker::get_topic(e)},
+                               broker::get_data(e).to_data());
            return rval;
          })
 
@@ -242,8 +243,8 @@ PYBIND11_MODULE(_broker, m) {
            std::vector<topic_data_pair> rval;
            rval.reserve(res.size());
            for (auto& e : res)
-             rval.emplace_back(
-               std::make_pair(broker::get_topic(e), broker::get_data(e)));
+             rval.emplace_back(broker::topic{broker::get_topic(e)},
+                               broker::get_data(e).to_data());
            return rval;
          })
     .def("available", &broker::subscriber::available)
@@ -392,18 +393,15 @@ PYBIND11_MODULE(_broker, m) {
     .def("peers", &broker::endpoint::peers)
     .def("peer_subscriptions", &broker::endpoint::peer_subscriptions)
     .def("forward", &broker::endpoint::forward)
-    .def("publish", (void(broker::endpoint::*)(broker::topic t, broker::data d))
+    .def("publish", (void(broker::endpoint::*)(broker::topic, broker::data))
                       & broker::endpoint::publish)
-    .def("publish", (void(broker::endpoint::*)(const broker::endpoint_info& dst,
-                                               broker::topic t, broker::data d))
+    .def("publish", (void(broker::endpoint::*)(const broker::endpoint_info&,
+                                               broker::topic, broker::data))
                       & broker::endpoint::publish)
     .def("publish_batch",
          [](broker::endpoint& ep, std::vector<topic_data_pair> batch) {
-           std::vector<broker::data_message> xs;
-           xs.reserve(batch.size());
-           for (auto& m : batch)
-             xs.emplace_back(std::move(m.first), std::move(m.second));
-           ep.publish(std::move(xs));
+           for (auto& item : batch)
+             ep.publish(std::move(item.first), item.second);
          })
     .def("make_publisher", &broker::endpoint::make_publisher)
     .def("make_subscriber", &broker::endpoint::make_subscriber,

bindings/python/zeek.cpp

@@ -21,22 +21,22 @@ using namespace pybind11::literals;
 
 void init_zeek(py::module& m) {
   py::class_<broker::zeek::Message>(m, "Message")
-    .def("as_data",
-         static_cast<const broker::data& (broker::zeek::Message::*) () const>(
-           &broker::zeek::Message::as_data));
-
+    .def("as_data", [](const broker::zeek::Message& msg) {
+      return msg.as_data().to_data();
+    });
   py::class_<broker::zeek::Event, broker::zeek::Message>(m, "Event")
-    .def(py::init(
-      [](broker::data data) { return broker::zeek::Event(std::move(data)); }))
-    .def(py::init([](std::string name, broker::data args,
-                     std::optional<broker::data> metadata) {
-           if (metadata)
-             return broker::zeek::Event(
-               std::move(name), std::move(broker::get<broker::vector>(args)),
-               std::move(broker::get<broker::vector>(*metadata)));
-           else
-             return broker::zeek::Event(
-               std::move(name), std::move(broker::get<broker::vector>(args)));
+    .def(py::init([](const broker::data& content) {
+      auto topic_str = broker::topic::reserved;
+      auto msg = broker::make_data_message(broker::topic{topic_str}, content);
+      return broker::zeek::Event(msg->value());
+    }))
+    .def(py::init([](const std::string& name, const broker::data& args,
+                     const std::optional<broker::data>& metadata) {
+           const auto& args_vec = broker::get<broker::vector>(args);
+           if (!metadata)
+             return broker::zeek::Event(name, args_vec);
+           return broker::zeek::Event(name, args_vec,
+                                      broker::get<broker::vector>(*metadata));
          }),
          py::arg("name"), py::arg("args"), py::arg("metadata") = py::none())
     .def("valid",
@@ -47,15 +47,15 @@ void init_zeek(py::module& m) {
            return ev.valid();
          })
     .def("name",
-         [](const broker::zeek::Event& ev) -> const std::string& {
+         [](const broker::zeek::Event& ev) -> std::string {
            auto t = broker::zeek::Message::type(ev.as_data());
            if (t != broker::zeek::Message::Type::Event) {
              throw std::invalid_argument("invalid Event data/type");
            }
            if (!ev.valid()) {
              throw std::invalid_argument("invalid Event data");
            }
-           return ev.name();
+           return std::string{ev.name()};
          })
     .def("metadata",
          [](const broker::zeek::Event& ev) -> std::optional<broker::vector> {
@@ -66,20 +66,25 @@ void init_zeek(py::module& m) {
            if (!ev.valid()) {
              throw std::invalid_argument("invalid Event data");
            }
-
-           if (const auto vec = ev.metadata().get_vector())
-             return *vec;
+           if (auto meta = ev.metadata().raw(); !meta.empty()) {
+             broker::vector result;
+             convert(meta, result);
+             return result;
+           }
 
            return std::nullopt;
          })
-    .def("args", [](const broker::zeek::Event& ev) -> const broker::vector& {
+    .def("args", [](const broker::zeek::Event& ev) -> broker::vector {
       auto t = broker::zeek::Message::type(ev.as_data());
       if (t != broker::zeek::Message::Type::Event) {
         throw std::invalid_argument("invalid Event data/type");
       }
       if (!ev.valid()) {
         throw std::invalid_argument("invalid Event data");
       }
-      return ev.args();
+      auto args = ev.args();
+      broker::vector result;
+      convert(args, result);
+      return result;
     });
 }

ci/alpine/Dockerfile

@@ -16,7 +16,8 @@ RUN apk add --no-cache \
   make \
   openssl-dev \
   py3-pip \
+  py3-websockets \
   python3 \
   python3-dev
 
-RUN pip3 install websockets junit2html
+RUN pip3 install --break-system-packages junit2html

ci/debian-11/Dockerfile

@@ -17,5 +17,6 @@ RUN apt-get update && apt-get -y install \
     python3 \
     python3-dev \
     python3-pip \
+    python3-venv \
   && apt autoclean \
   && rm -rf /var/lib/apt/lists/*

ci/debian-12/Dockerfile

@@ -19,6 +19,7 @@ RUN apt-get update && apt-get -y install \
     python3 \
     python3-dev \
     python3-pip \
+    python3-venv \
   && apt autoclean \
   && rm -rf /var/lib/apt/lists/*
 

ci/test.sh

@@ -37,7 +37,9 @@ fi
 if command -v pip3 >/dev/null 2>&1 ; then
     BinDir="$(python3 -m site --user-base)/bin"
     export PATH="$PATH:$BinDir"
-    pip3 install --user btest websockets
+    python3 -m venv test-env
+    source test-env/bin/activate
+    pip3 install btest websockets
     cd $BaseDir/tests/btest
     btest || result=1
     [[ -d .tmp ]] && tar -czf tmp.tar.gz .tmp

ci/windows/Dockerfile

@@ -33,8 +33,8 @@ RUN choco install -y --no-progress visualstudio2022buildtools
 RUN choco install -y --no-progress visualstudio2022-workload-vctools
 
 # Install some dependencies that aren't included in the MSVC tools
-RUN choco install -y --no-progress openssl
-RUN choco install -y --no-progress --version=3.10.0 python
+RUN choco install -y --no-progress openssl --version=3.1.1
+RUN choco install -y --no-progress python --version=3.10.0
 RUN choco install -y --no-progress msysgit
 
 # Add git to the global PATH