Merge pull request #32 from nera0/master

Add BatchMessageListener interface and enrich the Producer and Consumer interface

Author: duhenglucky

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

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+import java.util.List;
+
+public interface BatchMessage {
+    /**
+     * @return all messages in this {@code BatchMessage}
+     */
+    List<Message> messages();
+}

openmessaging-api/src/main/java/io/openmessaging/consumer/BatchMessageListener.java

@@ -0,0 +1,59 @@
+/*
+ * 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.consumer;
+
+import io.openmessaging.BatchMessage;
+import io.openmessaging.Message;
+import io.openmessaging.exception.OMSRuntimeException;
+
+/**
+ * A message listener can implement this {@code BathMessageListener} interface and register itself to a consumer instance
+ * to asynchronously receive messages.
+ *
+ * @version OMS 1.0.0
+ * @since OMS 1.0.0
+ */
+public interface BatchMessageListener {
+    /**
+     * Callback method to receive incoming messages.
+     * <p>
+     * A message listener should handle different types of {@code BatchMessage}.
+     *
+     * @param batchMessage the received batchMessage.
+     */
+    void onReceived(BatchMessage batchMessage, Context context);
+
+
+    interface Context {
+        /**
+         * Acknowledges the specified and consumed message, which is related to this {@code MessageContext}.
+         * <p>
+         * Messages that have been received but not acknowledged may be redelivered.
+         *
+         * @throws OMSRuntimeException if the consumer fails to acknowledge the messages due to some internal error.
+         */
+        void success(Message... messages);
+        /**
+         * Acknowledges messages, which is related to this {@code MessageContext}.
+         * <p>
+         *
+         * @throws OMSRuntimeException if the consumer fails to acknowledge the messages due to some internal error.
+         */
+        void ack();
+    }
+}

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

@@ -26,6 +26,8 @@
 import io.openmessaging.exception.OMSTimeOutException;
 import io.openmessaging.interceptor.ConsumerInterceptor;
 
+import java.util.List;
+
 /**
  * A {@code PushConsumer} receives messages from multiple queues, these messages are pushed from MOM server to {@code
  * PushConsumer} client.
@@ -109,6 +111,20 @@ public interface Consumer extends ServiceLifecycle {
      */
     void bindQueue(String queueName, MessageListener listener);
 
+    /**
+     * Bind the {@code Consumer} to a specified queue, with a {@code BatchMessageListener}.
+     * <p>
+     * {@link BatchMessageListener#onReceived(List<Message>, BatchMessageListener.Context)} will be called when new delivered messages is
+     * coming.
+     *
+     * @param queueName a specified queue.
+     * @param listener a specified listener to receive new messages.
+     * @throws OMSSecurityException when have no authority to bind to this queue.
+     * @throws OMSDestinationException when have no given destination in the server.
+     * @throws OMSRuntimeException when the {@code Producer} fails to send the message due to some internal error.
+     */
+    void bindQueue(String queueName, BatchMessageListener listener);
+
     /**
      * Unbind the {@code Consumer} from a specified queue.
      * <p>
@@ -118,6 +134,20 @@ public interface Consumer extends ServiceLifecycle {
      */
     void unbindQueue(String queueName);
 
+    /**
+     * This method is used to find out whether the {@code Consumer} in bind queue.
+     *
+     * @return true if this {@code Consumer} is bind, false otherwise.
+     */
+    boolean isBindQueue();
+
+    /**
+     * This method is used to find out the queue bind to {@code Consumer}.
+     *
+     * @return the queue this consumer is bind, or null if the consumer is not bind queue.
+     */
+    String getBindQueue();
+
     /**
      * Adds a {@code ConsumerInterceptor} instance to this consumer.
      *
@@ -146,6 +176,20 @@ public interface Consumer extends ServiceLifecycle {
      */
     Message receive(long timeout);
 
+    /**
+     * Receives the next batch messages from the bind queues of this consumer in pull model.
+     * <p>
+     * This call blocks indefinitely until the messages is arrives, the timeout expires, or until this {@code PullConsumer}
+     * is shut down.
+     *
+     * @param timeout receive messages will blocked at most <code>timeout</code> milliseconds.
+     * @return the next batch messages received from the bind queues, or null if the consumer is concurrently shut down.
+     * @throws OMSSecurityException when have no authority to receive messages from this queue.
+     * @throws OMSTimeOutException when the given timeout elapses before the send operation completes.
+     * @throws OMSRuntimeException when the {@code Producer} fails to send the message due to some internal error.
+     */
+    List<Message> batchReceive(long timeout);
+
     /**
      * Acknowledges the specified and consumed message with the unique message receipt handle, in the scenario of using
      * manual commit.

openmessaging-api/src/main/java/io/openmessaging/producer/Producer.java

@@ -96,6 +96,29 @@ public interface Producer extends MessageFactory, ServiceLifecycle {
      */
     void send(List<Message> messages);
 
+    /**
+     * Send messages to the specified destination asynchronously, the destination should be preset to {@link
+     * Message#headers()}, other header fields as well.
+     * <p>
+     * The returned {@code Promise} will have the result once the operation completes, and the registered {@code
+     * FutureListener} will be notified, either because the operation was successful or because of an error.
+     *
+     * @param messages a batch messages will be sent.
+     * @return the {@code Promise} of an asynchronous messages send operation.
+     * @see Future
+     * @see FutureListener
+     */
+    Future<List<SendResult>> sendAsync(List<Message> messages);
+
+    /**
+     * <p>
+     * There is no {@code Promise} related or {@code RuntimeException} thrown. The calling thread doesn't care about the
+     * send result and also have no context to get the result.
+     *
+     * @param messages a  batch message will be sent.
+     */
+    void sendOneway(List<Message> messages);
+
     /**
      * Adds a {@code ProducerInterceptor} to intercept send operations of producer.
      *