Feat/add serialization option

openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/DefaultSerializer.java

@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+package io.openmessaging.samples.producer;
+
+import java.io.IOException;
+
+import io.openmessaging.api.serialization.Serializer;
+
+
+public class DefaultSerializer<T> implements Serializer<T> {
+
+    @Override
+    public byte[] serialize(String topic, T t) {
+        return new byte[0];
+    }
+
+    @Override
+    public void close() throws IOException {
+        // no ops
+    }
+}

openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/GenericTransactionProducerApp.java

@@ -83,7 +83,7 @@ public void run() {
         MessageSample messageSample = new MessageSample("Bob");
 
         Message genericMessage = producer.messageBuilder().withTopic("NS://topicA")
-                .withValue(messageSample).withKey("messageKey").withTags("TagA").build();
+                .withValue(messageSample).withKey("messageKey").withTags("TagA").withSerializationType("json").build();
 
         SendResult sendResult = producer.send(genericMessage, new GenericLocalTransactionExecuter<MessageSample>() {
             @Override

openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/JsonSerializer.java

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+package io.openmessaging.samples.producer;
+
+import java.io.IOException;
+
+import io.openmessaging.api.serialization.Serializer;
+
+public class JsonSerializer<T> implements Serializer<T> {
+
+    @Override
+    public byte[] serialize(String topic, T body) {
+        return new byte[0];
+    }
+
+    @Override
+    public void close() throws IOException {
+        // no ops
+    }
+}

openmessaging-api-samples/src/main/java/io/openmessaging/samples/producer/MessageBuilderImpl.java

@@ -18,9 +18,7 @@
 
 import io.openmessaging.api.Message;
 import io.openmessaging.api.MessageBuilder;
-import io.openmessaging.api.OMSBuiltinKeys;
 import io.openmessaging.api.serialization.Serializer;
-import java.lang.reflect.Constructor;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -45,9 +43,7 @@ public class MessageBuilderImpl<T> implements MessageBuilder<T> {
 
     public MessageBuilderImpl(Properties properties) throws Exception {
         this.properties = properties;
-        Class<?> clazz = Class.forName(properties.getProperty(OMSBuiltinKeys.ENDPOINT));
-        Constructor<T> constructor = (Constructor<T>) clazz.getDeclaredConstructor();
-        this.serializer = (Serializer<T>) constructor.newInstance();
+        this.serializer = new DefaultSerializer<>();
     }
 
     @Override public MessageBuilder withTopic(String topic) {
@@ -80,6 +76,13 @@ public MessageBuilderImpl(Properties properties) throws Exception {
         return this;
     }
 
+    @Override public MessageBuilder<T> withSerializationType(String serializationType) {
+        if (serializationType.equalsIgnoreCase("json")) {
+            this.serializer = new JsonSerializer<T>();
+        }
+        return this;
+    }
+
     @Override public String getTopic() {
         return this.topic;
     }

openmessaging-api/src/main/java/io/openmessaging/api/MessageBuilder.java

@@ -73,6 +73,14 @@ public interface MessageBuilder<T> {
      */
     MessageBuilder<T> withValue(T t);
 
+    /**
+     * Used for serializing message
+     *
+     * @param serializationType serialization type
+     * @return {@link MessageBuilder}
+     */
+    MessageBuilder<T> withSerializationType(String serializationType);
+
     /**
      * Get the topic which this {@code MessageBuilder} belongs to.
      *