st: byte-buffer SerDe seam + zero-copy BytesView

Per PR review on the Schema encode/decode signatures.

SerDe seam (Schema<T> + JSON/Avro/protobuf factories):
- encode writes into a caller-provided, reusable std::vector<std::byte>& (no
  per-message allocation) and returns Expected<void>;
- decode takes std::span<const std::byte> and returns Expected<T>, so malformed
  input / an unset schema are error values rather than throws;
- Bytes is now std::vector<std::byte>.

Client-facing API unchanged: Message<T>::value() still returns T (decode
failures are handled inside the SDK), Producer::send(const T&) and the examples
are as before; a rare encode error is stashed in the builder and surfaces from
send()/sendAsync().

Zero-copy bytes: new BytesView = std::span<const std::byte>. Schema<BytesView>
is the zero-copy counterpart of Schema<Bytes> -- Producer<BytesView> publishes
the caller's bytes without copying (the caller keeps them valid until the send
completes) and Message<BytesView>::value() returns a view into the message
buffer. OutgoingMessage carries an optional non-owning view.

Verified with clang + gcc:13 (-Wextra -Werror), clang-format-11, and a runtime
check that decode returns a view at the same address.

Signed-off-by: Matteo Merli <mmerli@apache.org>

Author: merlimat

include/pulsar/st/AvroSchema.h

@@ -21,9 +21,12 @@
 #include <pulsar/st/Schema.h>
 
 #include <cstddef>
+#include <exception>
 #include <rfl.hpp>
 #include <rfl/avro.hpp>
+#include <span>
 #include <string>
+#include <vector>
 
 // avroSchema<T>() is the Avro counterpart of jsonSchema<T>(): reflect-cpp derives
 // the SerDe and the Avro schema from T's fields — no per-type serializer. The
@@ -43,9 +46,19 @@ namespace detail {
 template <typename T>
 struct AvroSerDe {
     SchemaInfo info() const { return SchemaInfo(SchemaType::AVRO, "AVRO", rfl::avro::to_schema<T>()); }
-    std::string encode(const T& value) const { return rfl::avro::write(value); }
-    T decode(const char* data, std::size_t size) const {
-        return rfl::avro::read<T>(std::string(data, size)).value();
+    Expected<void> encode(const T& value, std::vector<std::byte>& out) const {
+        const std::string s = rfl::avro::write(value);
+        const auto* p = reinterpret_cast<const std::byte*>(s.data());
+        out.assign(p, p + s.size());
+        return {};
+    }
+    Expected<T> decode(std::span<const std::byte> data) const {
+        try {
+            return rfl::avro::read<T>(std::string(reinterpret_cast<const char*>(data.data()), data.size()))
+                .value();
+        } catch (const std::exception& e) {
+            return unexpected(pulsar::ResultInvalidMessage, e.what());
+        }
     }
 };
 }  // namespace detail
@@ -63,9 +76,8 @@ struct AvroSerDe {
  *
  * @tparam T the struct type to serialize as Avro; its fields must be reflectable
  *         by reflect-cpp.
- * @return a `Schema<T>` whose `encode`/`decode` use Avro.
- * @throws std::runtime_error (from reflect-cpp) at decode time if the input bytes
- *         are not a valid Avro encoding for `T`.
+ * @return a `Schema<T>` whose `encode`/`decode` use Avro. `decode` reports input
+ *         that is not a valid Avro encoding for `T` as an `Error` rather than throwing.
  */
 template <typename T>
 Schema<T> avroSchema() {

include/pulsar/st/JsonSchema.h

@@ -21,9 +21,12 @@
 #include <pulsar/st/Schema.h>
 
 #include <cstddef>
+#include <exception>
 #include <rfl.hpp>
 #include <rfl/json.hpp>
+#include <span>
 #include <string>
+#include <vector>
 
 // jsonSchema<T>() derives BOTH the JSON SerDe and the declared schema from T's
 // fields via reflect-cpp (https://github.com/getml/reflect-cpp) — no per-type
@@ -40,9 +43,19 @@ namespace detail {
 template <typename T>
 struct JsonSerDe {
     SchemaInfo info() const { return SchemaInfo(SchemaType::JSON, "JSON", rfl::json::to_schema<T>()); }
-    std::string encode(const T& value) const { return rfl::json::write(value); }
-    T decode(const char* data, std::size_t size) const {
-        return rfl::json::read<T>(std::string(data, size)).value();
+    Expected<void> encode(const T& value, std::vector<std::byte>& out) const {
+        const std::string s = rfl::json::write(value);
+        const auto* p = reinterpret_cast<const std::byte*>(s.data());
+        out.assign(p, p + s.size());
+        return {};
+    }
+    Expected<T> decode(std::span<const std::byte> data) const {
+        try {
+            return rfl::json::read<T>(std::string(reinterpret_cast<const char*>(data.data()), data.size()))
+                .value();
+        } catch (const std::exception& e) {
+            return unexpected(pulsar::ResultInvalidMessage, e.what());
+        }
     }
 };
 }  // namespace detail
@@ -63,9 +76,8 @@ struct JsonSerDe {
  *
  * @tparam T the struct type to serialize as JSON; its fields must be reflectable
  *         by reflect-cpp.
- * @return a `Schema<T>` whose `encode`/`decode` use JSON.
- * @throws std::runtime_error (from reflect-cpp) at decode time if the input bytes
- *         are not valid JSON for `T`.
+ * @return a `Schema<T>` whose `encode`/`decode` use JSON. `decode` reports input
+ *         that is not valid JSON for `T` as an `Error` rather than throwing.
  */
 template <typename T>
 Schema<T> jsonSchema() {

include/pulsar/st/Message.h

@@ -27,6 +27,7 @@
 #include <cstddef>
 #include <cstdint>
 #include <optional>
+#include <span>
 #include <string>
 #include <vector>
 
@@ -61,12 +62,17 @@ class Message {
     /**
      * Decode the payload through `Schema<T>` and return the typed value.
      *
-     * Decoding happens on every call (the result is not cached). May throw if the
-     * payload bytes are malformed for the schema.
+     * Decoding happens on every call (the result is not cached). The SDK handles a
+     * payload that cannot be decoded internally — such a message is not delivered to
+     * the application — so this does not surface decode failures to the caller. The
+     * raw bytes remain available via `data()` / `size()`.
      *
      * @return the decoded value of type `T`.
      */
-    T value() const { return schema_.decode(core_.data(), core_.size()); }
+    T value() const {
+        const auto* bytes = reinterpret_cast<const std::byte*>(core_.data());
+        return schema_.decode(std::span<const std::byte>(bytes, core_.size())).value();
+    }
 
     /**
      * Pointer to the raw, undecoded payload bytes.

include/pulsar/st/Producer.h

@@ -30,10 +30,13 @@
 #include <pulsar/st/detail/ProducerCore.h>
 
 #include <chrono>
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <optional>
+#include <span>
 #include <string>
+#include <type_traits>
 #include <vector>
 
 namespace pulsar::st {
@@ -99,8 +102,14 @@ struct ProducerConfig {
  * its fluent setters and hands the result to the producer core for publishing.
  */
 struct OutgoingMessage {
-    /** Encoded message payload (the value serialized through `Schema<T>`). */
-    std::string payload;
+    /** Encoded message payload — the value serialized to bytes through `Schema<T>`.
+     *  Published unless `usesView` is set. */
+    std::vector<std::byte> payload;
+    /** Non-owning view of already-encoded bytes for zero-copy publishing
+     *  (`Schema<BytesView>`); the caller keeps them valid until the send completes. */
+    std::span<const std::byte> payloadView;
+    /** When true, publish `payloadView` directly without copying; otherwise `payload`. */
+    bool usesView = false;
     /** Whether a routing/ordering key is set. `false` (the default) means no key. */
     bool hasKey = false;
     /** Partition/ordering key; meaningful only when `hasKey` is true. */
@@ -145,11 +154,22 @@ class MessageBuilder {
      * Set the message value, encoding it to bytes through this producer's
      * `Schema<T>`.
      *
+     * For a zero-copy `Schema<BytesView>` producer the bytes are not copied — the
+     * view is published directly, so the caller must keep them valid until the send
+     * completes. A rare encoding failure (e.g. an unset schema) is not reported here,
+     * so the fluent chain stays unbroken; it surfaces from the terminal `send()` /
+     * `sendAsync()` instead.
+     *
      * @param v the typed value to publish.
      * @return `*this`, for chaining.
      */
     MessageBuilder& value(const T& v) {
-        message_.payload = schema_.encode(v);
+        if constexpr (std::is_same_v<T, BytesView>) {
+            message_.payloadView = v;
+            message_.usesView = true;
+        } else {
+            if (auto r = schema_.encode(v, message_.payload); !r) encodeError_ = r.error();
+        }
         return *this;
     }
     /**
@@ -246,7 +266,14 @@ class MessageBuilder {
      * @return a `Future<MessageId>` that completes with the assigned id on success
      *         or the failure. The future may be ignored for fire-and-forget sends.
      */
-    Future<MessageId> sendAsync() { return core_.sendAsync(std::move(message_)); }
+    Future<MessageId> sendAsync() {
+        if (encodeError_) {
+            detail::Promise<MessageId> promise;
+            promise.setError(*encodeError_);
+            return promise.getFuture();
+        }
+        return core_.sendAsync(std::move(message_));
+    }
 
    private:
     friend class Producer<T>;
@@ -260,6 +287,7 @@ class MessageBuilder {
     detail::ProducerCore core_;
     Schema<T> schema_;
     OutgoingMessage message_;
+    std::optional<Error> encodeError_;
 };
 
 /**

include/pulsar/st/ProtobufNativeSchema.h

@@ -24,8 +24,10 @@
 #include <pulsar/st/Schema.h>
 
 #include <cstddef>
+#include <span>
 #include <string>
 #include <type_traits>
+#include <vector>
 
 namespace pulsar::st {
 
@@ -38,11 +40,16 @@ struct ProtobufNativeSerDe {
     static_assert(std::is_base_of_v<google::protobuf::Message, T>,
                   "protobufNativeSchema<T> requires T to be a generated protobuf Message");
     SchemaInfo info() const { return pulsar::createProtobufNativeSchema(T::descriptor()); }
-    std::string encode(const T& value) const { return value.SerializeAsString(); }
-    T decode(const char* data, std::size_t size) const {
+    Expected<void> encode(const T& value, std::vector<std::byte>& out) const {
+        out.resize(value.ByteSizeLong());
+        if (!value.SerializeToArray(out.data(), static_cast<int>(out.size())))
+            return unexpected(pulsar::ResultInvalidMessage, "failed to serialize protobuf message");
+        return {};
+    }
+    Expected<T> decode(std::span<const std::byte> data) const {
         T message;
-        message.ParseFromArray(data, static_cast<int>(size));
-        return message;
+        if (message.ParseFromArray(data.data(), static_cast<int>(data.size()))) return message;
+        return unexpected(pulsar::ResultInvalidMessage, "failed to parse protobuf message");
     }
 };
 }  // namespace detail