Merge pull request #71 from duhenglucky/schema_with_builder

[DISCUSS]feat(schema) add schema support for OMS

Author: duhenglucky

openmessaging-api-samples/src/main/java/io/openmessaging/samples/MessageSample.java

@@ -0,0 +1,32 @@
+/*
+ * 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;
+public class MessageSample {
+    private String name;
+
+    public MessageSample(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+}

openmessaging-api-samples/src/main/java/io/openmessaging/samples/consumer/GenericPullConsumerApp.java

@@ -0,0 +1,69 @@
+/*
+ * 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.consumer;
+
+import io.openmessaging.api.GenericMessage;
+import io.openmessaging.api.MessagingAccessPoint;
+import io.openmessaging.api.OMS;
+import io.openmessaging.api.OMSBuiltinKeys;
+import io.openmessaging.api.PullConsumer;
+import io.openmessaging.api.TopicPartition;
+import io.openmessaging.samples.MessageSample;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+public class GenericPullConsumerApp {
+    public static void main(String[] args) {
+        //Load and start the vendor implementation from a specific OMS driver URL.
+        final MessagingAccessPoint messagingAccessPoint =
+            OMS.builder()
+                .endpoint("http://mq-instance-xxx-1234567890-test:8080")
+                .region("Shanghai")
+                .driver("rocketmq")
+                .build();
+
+        Properties properties = new Properties();
+        properties.setProperty(OMSBuiltinKeys.DESERIALIZER, "io.openmessaging.openmeta.impl.Deserializer");
+        properties.setProperty(OMSBuiltinKeys.OPEN_META_URL, "http://localhost:1234");
+
+        //Start a PullConsumer to receive messages from the specific queue.
+        final PullConsumer consumer = messagingAccessPoint.createPullConsumer(properties);
+
+        //Register a shutdown hook to close the opened endpoints.
+        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+            @Override
+            public void run() {
+                consumer.shutdown();
+            }
+        }));
+
+        Set<TopicPartition> topicPartitions = consumer.topicPartitions("NS://TOPIC");
+        consumer.assign(topicPartitions);
+        consumer.start();
+
+        List<GenericMessage<MessageSample>> messages = consumer.pollGenericMessages(1000);
+        for (GenericMessage<MessageSample> message : messages) {
+            System.out.println("Received message: " + message.getValue());
+        }
+        //Acknowledge the consumed message
+        consumer.commitSync();
+        consumer.shutdown();
+
+    }
+}

openmessaging-api-samples/src/main/java/io/openmessaging/samples/consumer/GenericPushConsumerApp.java

@@ -0,0 +1,70 @@
+/*
+ * 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.consumer;
+
+import io.openmessaging.api.Action;
+import io.openmessaging.api.Consumer;
+import io.openmessaging.api.GenericMessage;
+import io.openmessaging.api.GenericMessageListener;
+import io.openmessaging.api.MessageConsumeContext;
+import io.openmessaging.api.MessagingAccessPoint;
+import io.openmessaging.api.OMS;
+import io.openmessaging.api.OMSBuiltinKeys;
+import io.openmessaging.samples.MessageSample;
+import java.util.Properties;
+
+public class GenericPushConsumerApp {
+    public static void main(String[] args) {
+        //Load and start the vendor implementation from a specific OMS driver URL.
+        final MessagingAccessPoint messagingAccessPoint =
+            OMS.builder()
+                .region("Shanghai")
+                .endpoint("127.0.0.1:9876")
+                .driver("rocketmq")
+                .withCredentials(new Properties())
+                .build();
+
+        Properties properties = new Properties();
+        properties.setProperty(OMSBuiltinKeys.DESERIALIZER, "io.openmessaging.openmeta.impl.Deserializer");
+        properties.setProperty(OMSBuiltinKeys.OPEN_META_URL, "http://localhost:1234");
+
+        final Consumer consumer = messagingAccessPoint.createConsumer(properties);
+        consumer.start();
+
+        //Register a shutdown hook to close the opened endpoints.
+        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+            @Override
+            public void run() {
+                consumer.shutdown();
+            }
+        }));
+
+        //Consume messages from a simple queue.
+        String topic = "NS://HELLO_TOPIC";
+
+        consumer.subscribe(topic, "*", new GenericMessageListener<MessageSample>() {
+            @Override
+            public Action consume(GenericMessage<MessageSample> message, MessageConsumeContext context) {
+                MessageSample messageSample = message.getValue();
+                System.out.println("Received message: " + messageSample);
+                return Action.CommitMessage;
+            }
+        });
+
+        consumer.shutdown();
+    }
+}

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

@@ -0,0 +1,93 @@
+/*
+ * 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 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;
+
+public class MessageBuilderImpl<T> implements MessageBuilder<T> {
+
+    private T body;
+
+    private String keys;
+
+    private String tags;
+
+    private String shardingKey;
+
+    private String topic;
+
+    private final Map<String, String> userProperties = new HashMap<>();
+
+    private final Properties properties;
+
+    private Serializer<T> serializer;
+
+    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();
+    }
+
+    @Override public MessageBuilder withTopic(String topic) {
+        this.topic = topic;
+        return this;
+    }
+
+    @Override public MessageBuilder withValue(T t) {
+        this.body = t;
+        return this;
+    }
+
+    @Override public MessageBuilder withKey(String keys) {
+        this.keys = keys;
+        return this;
+    }
+
+    @Override public MessageBuilder withTags(String tags) {
+        this.tags = tags;
+        return this;
+    }
+
+    @Override public MessageBuilder withProperty(String key, String value) {
+        this.properties.put(key, value);
+        return this;
+    }
+
+    @Override public MessageBuilder withShardingKey(String shardingKey) {
+        this.shardingKey = shardingKey;
+        return this;
+    }
+
+    @Override public String getTopic() {
+        return this.topic;
+    }
+
+    @Override public Message build() {
+        Message message = new Message();
+        message.setKey(this.keys);
+        message.setBody(this.serializer.serialize(this.topic, body));
+        return message;
+    }
+}

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

@@ -20,13 +20,16 @@
 import io.openmessaging.api.Message;
 import io.openmessaging.api.MessagingAccessPoint;
 import io.openmessaging.api.OMS;
+import io.openmessaging.api.OMSBuiltinKeys;
 import io.openmessaging.api.OnExceptionContext;
 import io.openmessaging.api.Producer;
 import io.openmessaging.api.SendCallback;
 import io.openmessaging.api.SendResult;
+import io.openmessaging.samples.MessageSample;
 import java.util.Properties;
 
 public class ProducerApp {
+
     public static void main(String[] args) {
         final MessagingAccessPoint messagingAccessPoint =
             OMS.builder()
@@ -36,7 +39,11 @@ public static void main(String[] args) {
                 .withCredentials(new Properties())
                 .build();
 
-        final Producer producer = messagingAccessPoint.createProducer(new Properties());
+        Properties properties = new Properties();
+        properties.setProperty(OMSBuiltinKeys.SERIALIZER, "io.openmessaging.openmeta.impl.Serializer");
+        properties.setProperty(OMSBuiltinKeys.OPEN_META_URL, "http://localhost:1234");
+
+        final Producer producer = messagingAccessPoint.createProducer(properties);
         producer.start();
 
         //Register a shutdown hook to close the opened endpoints.
@@ -47,9 +54,13 @@ public void run() {
             }
         }));
 
-        Message message = new Message("NS://Topic", "TagA", "Hello MQ".getBytes());
+        MessageSample messageSample = new MessageSample("Bob");
+
+        Message message = producer.messageBuilder().withTopic("NS://topicA")
+            .withValue(messageSample).withKey("messageKey").withTags("TagA").build();
 
         SendResult sendResult = producer.send(message);
+
         System.out.println("SendResult: " + sendResult);
 
         //Sends a message to the specified destination async.
@@ -61,7 +72,7 @@ public void onSuccess(SendResult sendResult) {
 
             @Override
             public void onException(OnExceptionContext context) {
-
+                context.getException().printStackTrace();
             }
         });
 

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

@@ -16,6 +16,6 @@
  */
 package io.openmessaging.api;
 
-public class ConsumeContext {
+public class ConsumeContext implements MessageConsumeContext{
 
 }

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

@@ -48,6 +48,30 @@ public interface Consumer extends Admin {
      */
     void subscribe(final String topic, final MessageSelector selector, final MessageListener listener);
 
+    /**
+     * Subscribe message in order.
+     *
+     * @param topic message topic.
+     * @param subExpression Subscribe to the filter expression string, which the broker filters based on this
+     * expression. <br> eg: "tag1 || tag2 || tag3"<br>, if subExpression is equal to null or *, it means subscribe all
+     * messages.
+     * @param listener The message callback listener, the consumer receives the message and then passes it to the
+     * message callback listener for consumption.
+     */
+    void subscribe(final String topic, final String subExpression, final GenericMessageListener listener);
+
+    /**
+     * Subscribe to messages, which can be filtered using SQL expressions.
+     *
+     * @param topic
+     * @param selector Subscribe to the message selector (can be empty, indicating no filtering), the ONS server filters
+     * according to the expression in this selector. Currently supports two expression syntax: {@link
+     * ExpressionType#TAG}, {@link ExpressionType#SQL92} Among them, the effect of TAG filtering is consistent with the
+     * above interface.
+     * @param listener Message callback listener
+     */
+    void subscribe(final String topic, final MessageSelector selector, final GenericMessageListener listener);
+
     /**
      * Unsubscribe message
      *

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

@@ -0,0 +1,32 @@
+/*
+ * 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.api;
+
+/**
+ * An abstract class used for get the object instead of bytes.
+ *
+ * @param <T> Object deserialized from the message body.
+ * @version OMS 2.0.0
+ * @since OMS 2.0.0
+ */
+public abstract class GenericMessage<T> extends Message {
+
+    /**
+     * @return Object deserialized from the message body.
+     */
+   public abstract T getValue();
+}