Merge pull request #74 from dbl-x/feat_support_schema

support schema in consumer

Author: duhenglucky

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

@@ -57,6 +57,12 @@ public void run() {
         String topic = "NS://HELLO_TOPIC";
 
         consumer.subscribe(topic, "*", new GenericMessageListener<MessageSample>() {
+
+            @Override
+            public Class<MessageSample> payloadClass() {
+                return MessageSample.class;
+            }
+
             @Override
             public Action consume(GenericMessage<MessageSample> message, MessageConsumeContext context) {
                 MessageSample messageSample = message.getValue();

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

@@ -22,7 +22,7 @@
  * @version OMS 1.2.0
  * @since OMS 1.2.0
  */
-public interface Consumer extends Admin {
+public interface Consumer extends ConsumerBase, Admin {
 
     /**
      * Subscribe message in order.
@@ -39,7 +39,7 @@ public interface Consumer extends Admin {
     /**
      * Subscribe to messages, which can be filtered using SQL expressions.
      *
-     * @param topic
+     * @param topic message 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
@@ -58,24 +58,17 @@ public interface Consumer extends Admin {
      * @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);
+    <T> void subscribe(final String topic, final String subExpression, final GenericMessageListener<T> listener);
 
     /**
      * Subscribe to messages, which can be filtered using SQL expressions.
      *
-     * @param topic
+     * @param topic message 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
-     *
-     * @param topic
-     */
-    void unsubscribe(final String topic);
+    <T> void subscribe(final String topic, final MessageSelector selector, final GenericMessageListener<T> listener);
 }

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

@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/**
+ * {@code ConsumerBase} is the basic of other {@code Consumer} interfaces
+ * and is used to define some common methods.
+ *
+ * @version OMS 2.0.3
+ * @since OMS 2.0.3
+ */
+public interface ConsumerBase {
+    /**
+     * Unsubscribe message of the specified topic
+     *
+     * @param topic message topic
+     */
+    void unsubscribe(final String topic);
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/**
+ * Base listener for other generic message listener.
+ *
+ * @version OMS 2.0.0
+ * @since OMS 2.0.0
+ */
+public interface GenericListener<T> {
+
+    /**
+     * Returns the class of the object that the user wants to deserialize the message into.
+     *
+     * @return the class of message payload that user expects
+     */
+    Class<T> payloadClass();
+}

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

@@ -28,7 +28,7 @@
  * @version OMS 2.0.0
  * @since OMS 2.0.0
  */
-public interface GenericMessageListener<T> {
+public interface GenericMessageListener<T> extends GenericListener<T> {
 
     /**
      * Consumer message interface, implemented by the application, unstable situations such as network jitter may lead

openmessaging-api/src/main/java/io/openmessaging/api/batch/BatchConsumer.java

@@ -18,14 +18,15 @@
 package io.openmessaging.api.batch;
 
 import io.openmessaging.api.Admin;
+import io.openmessaging.api.ConsumerBase;
 
 /**
  * Batch message consumer, used to subscribe to messages in batch.
  *
  * @version OMS 1.2.0
  * @since OMS 1.2.0
  */
-public interface BatchConsumer extends Admin {
+public interface BatchConsumer extends ConsumerBase, Admin {
 
     /**
      * Subscribe message
@@ -47,12 +48,5 @@ public interface BatchConsumer extends Admin {
      * messages.
      * @param listener consume message callback listener.
      */
-    void subscribe(final String topic, final String subExpression, final GenericBatchMessageListener listener);
-
-    /**
-     * Unsubscribe topic
-     *
-     * @param topic
-     */
-    void unsubscribe(final String topic);
+    <T> void subscribe(final String topic, final String subExpression, final GenericBatchMessageListener<T> listener);
 }

openmessaging-api/src/main/java/io/openmessaging/api/batch/GenericBatchMessageListener.java

@@ -17,6 +17,7 @@
 package io.openmessaging.api.batch;
 
 import io.openmessaging.api.Action;
+import io.openmessaging.api.GenericListener;
 import io.openmessaging.api.GenericMessage;
 import io.openmessaging.api.MessageConsumeContext;
 import java.util.List;
@@ -27,7 +28,7 @@
  * @version OMS 2.0.0
  * @since OMS 2.0.0
  */
-public interface GenericBatchMessageListener<T> {
+public interface GenericBatchMessageListener<T> extends GenericListener<T> {
     /**
      * When message arrived, this method will be invoked by order.
      *

openmessaging-api/src/main/java/io/openmessaging/api/order/GenericMessageOrderListener.java

@@ -16,6 +16,7 @@
  */
 package io.openmessaging.api.order;
 
+import io.openmessaging.api.GenericListener;
 import io.openmessaging.api.GenericMessage;
 import io.openmessaging.api.MessageConsumeContext;
 
@@ -26,10 +27,10 @@
  * @version OMS 2.0.0
  * @since OMS 2.0.0
  */
-public interface GenericMessageOrderListener<T> {
+public interface GenericMessageOrderListener<T> extends GenericListener<T> {
 
     /**
-     * When message arrived, this method will be invoked by order.
+     * When message arrived, this method will be invoked in order.
      *
      * @param message received message
      * @param context

openmessaging-api/src/main/java/io/openmessaging/api/order/OrderConsumer.java

@@ -18,6 +18,7 @@
 package io.openmessaging.api.order;
 
 import io.openmessaging.api.Admin;
+import io.openmessaging.api.ConsumerBase;
 import io.openmessaging.api.ExpressionType;
 import io.openmessaging.api.MessageSelector;
 
@@ -27,7 +28,7 @@
  * @version OMS 1.2.0
  * @since OMS 1.2.0
  */
-public interface OrderConsumer extends Admin {
+public interface OrderConsumer extends ConsumerBase, Admin {
 
     /**
      * Subscribe message in order.
@@ -44,12 +45,36 @@ public interface OrderConsumer extends Admin {
     /**
      * Subscribe to messages, which can be filtered using SQL expressions.
      *
-     * @param topic
+     * @param topic message 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 MessageOrderListener 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.
+     */
+    <T> void subscribe(final String topic, final String subExpression, final GenericMessageOrderListener<T> listener);
+
+    /**
+     * Subscribe to messages, which can be filtered using SQL expressions.
+     *
+     * @param topic message 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
+     */
+    <T> void subscribe(final String topic, final MessageSelector selector, final GenericMessageOrderListener<T> listener);
 }

openmessaging-api/src/main/java/io/openmessaging/api/schema/GenericRecord.java

@@ -19,18 +19,39 @@
 import java.util.List;
 
 /**
- * Used for traverse the data in the message body When do not know the specific Schema.
+ * Used for traverse the data in the message body when do not know the specific Schema.
  *
  * @version OMS 2.0.0
  * @since OMS 2.0.0
  */
 public interface GenericRecord {
 
+    /**
+     * return schema id
+     *
+     * @return schema id
+     */
     String getSchemaId();
 
+    /**
+     * return the list of fields associated with the record
+     *
+     * @return the list of fields associated with the record
+     */
     List<Field> getFields();
 
+    /**
+     * retrieve the value of the provided field
+     *
+     * @param field the field to retrieve the value
+     * @return the value object
+     */
     Object getField(Field field);
 
-    Object getField(String field);
+    /**
+     * retrieve the value of the provided fieldName
+     * @param fieldName the field name to retrieve the value
+     * @return the value object
+     */
+    Object getField(String fieldName);
 }