Allow topic extraction via json::decode()

libbroker/broker/format/json.cc

@@ -50,12 +50,20 @@ size_t encode_to_buf(timestamp value, std::array<char, 32>& buf) {
   return pos;
 }
 
-error decode(std::string_view str, variant& result) {
+error decode(std::string_view str, variant& result, std::string* topic) {
   // Parse the JSON text into a JSON object.
   auto val = caf::json_value::parse_shallow(str);
   if (!val)
     return error{ec::invalid_json};
   auto obj = val->to_object();
+
+  if (topic) {
+    auto maybe_topic = obj.value("topic");
+    if (maybe_topic.is_string())
+      *topic = std::string{maybe_topic.to_string().data(),
+                           maybe_topic.to_string().size()};
+  }
+
   // Try to convert the JSON structure into our binary serialization format.
   std::vector<std::byte> buf;
   buf.reserve(512); // Allocate some memory to avoid small allocations.

libbroker/broker/format/json.hh

@@ -2,6 +2,7 @@
 
 #include "broker/config.hh"
 #include "broker/data.hh"
+#include "broker/error.hh"
 #include "broker/fwd.hh"
 #include "broker/message.hh"
 
@@ -389,7 +390,9 @@ OutIter encode(const data_message& msg, OutIter out) {
 /// Tries to decode a JSON object from `str`. On success, the result is stored
 /// in `result` and the functions a default-constructed `error`. Otherwise, the
 /// function returns a non-empty error and leaves `result` in an unspecified
-/// state.
-error decode(std::string_view str, variant& result);
+/// state. If `topic` is not NULL and the top-level object contains a topic
+/// key, it is stored into the string instance pointed to.
+error decode(std::string_view str, variant& result,
+             std::string* topic = nullptr);
 
 } // namespace broker::format::json::v1