[feat] Support messages with generic types (#149)

### Motivation

Pulsar C++ client doesn't support schema yet. It only supports
configuring `SchemaInfo` when creating producers or consumers. The main
reason is that C++'s templates are processed at the time of compilation.
Templatizing `Producer`, `Consumer` or `Message` could expose all
internal code.

Currently, users might write the following code for serialization and
deserialization:

```c++
producer.send(MessageBuilder().setContent(encode(value)).build());
```

```c++
Message msg;
consumer.receive(msg);
auto value = decode(msg.getData(), msg.getSize());
```

However, the `encode` and `decode` functions are just possible
solutions from users, they might use some other interfaces like a class
with two virtual methods. There is no way to provide a common interface
for serialization and deserialization.

### Modifications

Add a `TypedMessageBuilder<T>` class template that accepts an encoder
function and a validation function. The validation function is used when
users want to simulate the Java client's `AUTO_PRODUCE` schema. Define a
full specialization for `std::string` template argument to avoid
encoding. Since it inherits the `MessageBuilder`, it's compatible with
the current code style:

```c++
// It should be noted you have to call `setValue` before methods in base class
auto msg = TypedMessageBuilder<T>(encoder).setValue(value).setPartitionKey(key).build();
```

Add a `TypedMessage<T>` class that only adds a decoder to the `Message`
instance and can be converted from the `Message` directly:

```c++
auto typedMsg = TypedMessage<T>(msg, decoder);
std::cout << typedMsg.getValue() << std::endl;  // decode the bytes
std::cout << typedMsg.getMessageId() << std::endl;  // call methods from Message
```

For convenience, the following APIs are added to `Consumer`:

```c++
template <typename T>
Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder);

template <typename T>
Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder);

template <typename T>
void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
                  typename TypedMessage<T>::Decoder decoder)
```

The `ConsumerConfiguration` can configure a listener that accepts a
`TypedMessage<T>` now:

```c++
template <typename T>
ConsumerConfiguration& setTypedMessageListener(
        std::function<void(Consumer&, const TypedMessage<T>&)> listener,
        typename TypedMessage<T>::Decoder decoder);
```

Since it calls the original listener and `Consumer` is only forward
declared in `ConsumerConfiguration.h`, the 1st argument is changed from
`Consumer` to `Consumer&`. It's an API change but it's backward
compatible because `std::function<void(Consumer, ...)>` can be cast to
`std::function<Consumer&, ...)>` implicitly.

Based on these API changes, we can write a separated header-only C++
library as the schema extension.

Author: BewareMyPower

include/pulsar/Consumer.h

@@ -21,6 +21,7 @@
 
 #include <pulsar/BrokerConsumerStats.h>
 #include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/TypedMessage.h>
 #include <pulsar/defines.h>
 
 #include <iostream>
@@ -91,6 +92,14 @@ class PULSAR_PUBLIC Consumer {
      */
     Result receive(Message& msg);
 
+    template <typename T>
+    Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder) {
+        Message rawMsg;
+        auto result = receive(rawMsg);
+        msg = TypedMessage<T>{rawMsg, decoder};
+        return result;
+    }
+
     /**
      *
      * @param msg a non-const reference where the received message will be copied
@@ -101,6 +110,14 @@ class PULSAR_PUBLIC Consumer {
      */
     Result receive(Message& msg, int timeoutMs);
 
+    template <typename T>
+    Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder) {
+        Message rawMsg;
+        auto result = receive(rawMsg, timeoutMs);
+        msg = TypedMessage<T>{rawMsg, decoder};
+        return result;
+    }
+
     /**
      * Receive a single message
      * <p>
@@ -114,6 +131,14 @@ class PULSAR_PUBLIC Consumer {
      */
     void receiveAsync(ReceiveCallback callback);
 
+    template <typename T>
+    void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
+                      typename TypedMessage<T>::Decoder decoder) {
+        receiveAsync([callback, decoder](Result result, const Message& msg) {
+            callback(result, TypedMessage<T>{msg, decoder});
+        });
+    }
+
     /**
      * Batch receiving messages.
      *

include/pulsar/ConsumerConfiguration.h

@@ -28,6 +28,7 @@
 #include <pulsar/Message.h>
 #include <pulsar/Result.h>
 #include <pulsar/Schema.h>
+#include <pulsar/TypedMessage.h>
 #include <pulsar/defines.h>
 
 #include <functional>
@@ -48,7 +49,7 @@ typedef std::function<void(Result, const Messages& msgs)> BatchReceiveCallback;
 typedef std::function<void(Result result, MessageId messageId)> GetLastMessageIdCallback;
 
 /// Callback definition for MessageListener
-typedef std::function<void(Consumer consumer, const Message& msg)> MessageListener;
+typedef std::function<void(Consumer& consumer, const Message& msg)> MessageListener;
 
 typedef std::shared_ptr<ConsumerEventListener> ConsumerEventListenerPtr;
 
@@ -126,6 +127,15 @@ class PULSAR_PUBLIC ConsumerConfiguration {
      */
     ConsumerConfiguration& setMessageListener(MessageListener messageListener);
 
+    template <typename T>
+    ConsumerConfiguration& setTypedMessageListener(
+        std::function<void(Consumer&, const TypedMessage<T>&)> listener,
+        typename TypedMessage<T>::Decoder decoder) {
+        return setMessageListener([listener, decoder](Consumer& consumer, const Message& msg) {
+            listener(consumer, TypedMessage<T>{msg, decoder});
+        });
+    }
+
     /**
      * @return the message listener
      */

include/pulsar/Message.h

@@ -183,7 +183,7 @@ class PULSAR_PUBLIC Message {
 
     bool operator==(const Message& msg) const;
 
-   private:
+   protected:
     typedef std::shared_ptr<MessageImpl> MessageImplPtr;
     MessageImplPtr impl_;
 

include/pulsar/MessageBuilder.h

@@ -162,8 +162,11 @@ class PULSAR_PUBLIC MessageBuilder {
      */
     MessageBuilder& create();
 
+   protected:
+    const char* data() const;
+    std::size_t size() const;
+
    private:
-    MessageBuilder(const MessageBuilder&);
     void checkMetadata();
     static std::shared_ptr<MessageImpl> createMessageImpl();
     Message::MessageImplPtr impl_;

include/pulsar/TypedMessage.h

@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <pulsar/Message.h>
+
+#include <functional>
+
+namespace pulsar {
+
+template <typename T>
+class TypedMessage : public Message {
+   public:
+    using Decoder = std::function<T(const char*, std::size_t)>;
+
+    TypedMessage() = default;
+
+    TypedMessage(
+        const Message& message, Decoder decoder = [](const char*, std::size_t) { return T{}; })
+        : Message(message), decoder_(decoder) {}
+
+    T getValue() const { return decoder_(static_cast<const char*>(getData()), getLength()); }
+
+    TypedMessage& setDecoder(Decoder decoder) {
+        decoder_ = decoder;
+        return *this;
+    }
+
+   private:
+    Decoder decoder_;
+};
+
+}  // namespace pulsar

include/pulsar/TypedMessageBuilder.h

@@ -0,0 +1,79 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#pragma once
+
+#include <pulsar/MessageBuilder.h>
+
+#include <functional>
+
+namespace pulsar {
+
+template <typename T>
+class TypedMessageBuilder : public MessageBuilder {
+   public:
+    using Encoder = std::function<std::string(const T&)>;
+    using Validator = std::function<void(const char* data, size_t)>;
+
+    TypedMessageBuilder(
+        Encoder encoder, Validator validator = [](const char*, std::size_t) {})
+        : encoder_(encoder), validator_(validator) {}
+
+    TypedMessageBuilder& setValue(const T& value) {
+        setContent(encoder_(value));
+        if (validator_) {
+            validator_(data(), size());
+        }
+        return *this;
+    }
+
+   private:
+    const Encoder encoder_;
+    const Validator validator_;
+};
+
+template <>
+class TypedMessageBuilder<std::string> : public MessageBuilder {
+   public:
+    // The validator should throw an exception to indicate the message is corrupted.
+    using Validator = std::function<void(const char* data, size_t)>;
+
+    TypedMessageBuilder(Validator validator = nullptr) : validator_(validator) {}
+
+    TypedMessageBuilder& setValue(const std::string& value) {
+        if (validator_) {
+            validator_(value.data(), value.size());
+        }
+        setContent(value);
+        return *this;
+    }
+
+    TypedMessageBuilder& setValue(std::string&& value) {
+        if (validator_) {
+            validator_(value.data(), value.size());
+        }
+        setContent(std::move(value));
+        return *this;
+    }
+
+   private:
+    Validator validator_;
+};
+using BytesMessageBuilder = TypedMessageBuilder<std::string>;
+
+}  // namespace pulsar

lib/ConsumerImpl.cc

@@ -773,7 +773,8 @@ void ConsumerImpl::internalListener() {
     try {
         consumerStatsBasePtr_->receivedMessage(msg, ResultOk);
         lastDequedMessageId_ = msg.getMessageId();
-        messageListener_(Consumer(get_shared_this_ptr()), msg);
+        Consumer consumer{get_shared_this_ptr()};
+        messageListener_(consumer, msg);
     } catch (const std::exception& e) {
         LOG_ERROR(getName() << "Exception thrown from listener" << e.what());
     }

lib/MessageBuilder.cc

@@ -16,6 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+#include <assert.h>
 #include <pulsar/MessageBuilder.h>
 
 #include <memory>
@@ -155,4 +156,15 @@ MessageBuilder& MessageBuilder::disableReplication(bool flag) {
     r.Swap(impl_->metadata.mutable_replicate_to());
     return *this;
 }
+
+const char* MessageBuilder::data() const {
+    assert(impl_->payload.data());
+    return impl_->payload.data();
+}
+
+size_t MessageBuilder::size() const {
+    assert(impl_->payload.data());
+    return impl_->payload.readableBytes();
+}
+
 }  // namespace pulsar

lib/MultiTopicsConsumerImpl.cc

@@ -539,7 +539,8 @@ void MultiTopicsConsumerImpl::internalListener(Consumer consumer) {
     Message m;
     incomingMessages_.pop(m);
     try {
-        messageListener_(Consumer(get_shared_this_ptr()), m);
+        Consumer self{get_shared_this_ptr()};
+        messageListener_(self, m);
         messageProcessed(m);
     } catch (const std::exception& e) {
         LOG_ERROR("Exception thrown from listener of Partitioned Consumer" << e.what());