transport: add support for setting custom payload

A custom payload can now be added to response_message.
If it is set, it will be sent to client and the custom_payload
flag will be set.

write_string_bytes_map method is added to response class
and a missing custom_payload flag is added to
cql_frame_flags.

transport/messages/result_message_base.hh

@@ -9,6 +9,7 @@
 
 #pragma once
 
+#include <map>
 #include <vector>
 #include <seastar/core/sstring.hh>
 
@@ -19,6 +20,7 @@ namespace messages {
 
 class result_message {
     std::vector<sstring> _warnings;
+    std::optional<std::unordered_map<sstring, bytes>> _custom_payload;
 public:
     class visitor;
     class visitor_base;
@@ -35,6 +37,17 @@ public:
         return _warnings;
     }
 
+    void add_custom_payload(sstring key, bytes value) {
+        if (!_custom_payload) {
+            _custom_payload = std::optional<std::unordered_map<sstring, bytes>>{std::unordered_map<sstring, bytes>()};
+        }
+        _custom_payload.value()[key] = value;
+    }
+
+    const std::optional<std::unordered_map<sstring, bytes>>& custom_payload() const {
+        return _custom_payload;
+    }
+
     virtual std::optional<unsigned> move_to_shard() const {
         return std::nullopt;
     }

transport/response.hh

@@ -73,6 +73,7 @@ public:
     void write_consistency(db::consistency_level c);
     void write_string_map(std::map<sstring, sstring> string_map);
     void write_string_multimap(std::multimap<sstring, sstring> string_map);
+    void write_string_bytes_map(const std::unordered_map<sstring, bytes>& map);
     void write_value(bytes_opt value);
     void write_value(std::optional<managed_bytes_view> value);
     void write(const cql3::metadata& m, bool skip = false);

transport/server.cc

@@ -1539,6 +1539,10 @@ make_result(int16_t stream, messages::result_message& msg, const tracing::trace_
         response->set_frame_flag(cql_frame_flags::warning);
         response->write_string_list(msg.warnings());
     }
+    if (msg.custom_payload()) {
+        response->set_frame_flag(cql_frame_flags::custom_payload);
+        response->write_string_bytes_map(msg.custom_payload().value());
+    }
     cql_server::fmt_visitor fmt{version, *response, skip_metadata};
     msg.accept(fmt);
     return response;
@@ -1788,6 +1792,15 @@ void cql_server::response::write_string_multimap(std::multimap<sstring, sstring>
     }
 }
 
+void cql_server::response::write_string_bytes_map(const std::unordered_map<sstring, bytes>& map)
+{
+    write_short(cast_if_fits<uint16_t>(map.size()));
+    for (auto&& s : map) {
+        write_string(s.first);
+        write_bytes(s.second);
+    }
+}
+
 void cql_server::response::write_value(bytes_opt value)
 {
     if (!value) {

transport/server.hh

@@ -75,6 +75,7 @@ enum class cql_compression {
 enum cql_frame_flags {
     compression = 0x01,
     tracing     = 0x02,
+    custom_payload = 0x04,
     warning     = 0x08,
 };