- * This method resumes the {@code Consumer} instance after it was suspended. The instance will not receive new - * messages between the suspend and resume calls. - * - * @throws OMSRuntimeException if the instance has not been suspended. - * @see Consumer#suspend() - */ - void resume(); - - /** - * Suspends the {@code Consumer} in push model for later resumption. - *
- * This method suspends the consumer until it is resumed. The consumer will not receive new messages between the - * suspend and resume calls. - *
- * This method behaves exactly as if it simply performs the call {@code suspend(0)}. - * - * @throws OMSRuntimeException if the instance is not currently running. - * @see Consumer#resume() - */ - void suspend(); - - /** - * Suspends the {@code Consumer} in push model for later resumption. - *
- * This method suspends the consumer until it is resumed or a specified amount of time has elapsed. The consumer - * will not receive new messages during the suspended state. - *
- * This method is similar to the {@link #suspend()} method, but it allows finer control over the amount of time to - * suspend, and the consumer will be suspended until it is resumed if the timeout is zero. - * - * @param timeout the maximum time to suspend in milliseconds. - * @throws OMSRuntimeException if the instance is not currently running. - */ - void suspend(long timeout); - - /** - * This method is used to find out whether the {@code Consumer} in push model is suspended. - * - * @return true if this {@code Consumer} is suspended, false otherwise. - */ - boolean isSuspended(); - - /** - * Bind the {@code Consumer} to a specified queue in pull model, user can use {@link Consumer#receive(long)} to get - * message from bind queue. - *
- * {@link MessageListener#onReceived(Message, MessageListener.Context)} will be called when new delivered message is - * coming. - * - * @param queueName a specified queue. - * @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); - - /** - * Bind the {@code Consumer} to a collection of queue in pull model, user can use {@link Consumer#receive(long)} to get - * messages from a collection of queue. - *
- * {@link MessageListener#onReceived(Message, MessageListener.Context)} will be called when new delivered message is
- * coming.
- *
- * @param queueNames a collection of queues.
- * @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(List
- * {@link MessageListener#onReceived(Message, MessageListener.Context)} will be called when new delivered message is
- * coming.
- *
- * @param queueName a specified queue.
- * @param listener a specified listener to receive new message.
- * @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, MessageListener listener);
-
- /**
- * Bind the {@code Consumer} to a collection of queue, with a {@code MessageListener}.
- *
- * {@link MessageListener#onReceived(Message, MessageListener.Context)} will be called when new delivered message is
- * coming.
- *
- * @param queueNames a collection of queues.
- * @param listener a specified listener to receive new message.
- * @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 bindQueues(List
- * {@link BatchMessageListener#onReceived(List, 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);
-
- /**
- * Bind the {@code Consumer} to a collection of queue, with a {@code BatchMessageListener}.
- *
- * {@link BatchMessageListener#onReceived(List, BatchMessageListener.Context)} will be called when new delivered
- * messages is coming.
- *
- * @param queueNames a collection of queues.
- * @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 bindQueues(List
- * After the success call, this consumer won't receive new message from the specified queue any more.
- *
- * @param queueName a specified queue.
- */
- void unbindQueue(String queueName);
-
- /**
- * Unbind the {@code Consumer} from a collection of queues.
- *
- * After the success call, this consumer won't receive new message from the specified queue any more.
- *
- * @param queueNames a collection of queues.
- */
- void unbindQueues(List
- * This call blocks indefinitely until a message is arrives, the timeout expires, or until this {@code PullConsumer}
- * is shut down.
- *
- * @param timeout receive message will blocked at most
- * This call blocks indefinitely until a message is arrives, the timeout expires, or until this {@code PullConsumer}
- * is shut down.
- *
- * @param queueName receive message from which queueName in Message Queue.
- * @param partitionId receive message from which partition in Message Queue.
- * @param receiptId receive message from which receipt position in Message Queue.
- * @param timeout receive message will blocked at most
- *
- * @param timeout receive messages will blocked at most
- *
- * @param queueName receive message from which queueName in Message Queue.
- * @param partitionId receive message from which partition in Message Queue.
- * @param receiptId receive message from which receipt position in Message Queue.
- * @param timeout receive messages will blocked at most
+ *
+ * @param queueNames a collection of queues.
+ * @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(Collection
+ * After the success call, this consumer won't receive new message from the specified queue any more.
+ *
+ * @param queueNames a collection of queues.
+ */
+ void unbindQueue(Collection
+ * This call blocks indefinitely until a message is arrives, the timeout expires, or until this {@code PullConsumer}
+ * is shut down.
+ *
+ * @return the next message received from the attached queues, or null if the consumer is concurrently shut down or
+ * the timeout expires
+ * @throws OMSRuntimeException if the consumer fails to pull the next message due to some internal error.
+ */
+ Message receive();
+
+ /**
+ * Receives the next message from the bind queues of this consumer in pull model.
+ *
+ * This call blocks indefinitely until a message is arrives, the timeout expires, or until this {@code PullConsumer}
+ * is shut down.
+ *
+ * @param timeout receive message will blocked at most
+ * This call blocks indefinitely until a message is arrives, the timeout expires, or until this {@code PullConsumer}
+ * is shut down.
+ *
+ * @param queueName receive message from which queueName in Message Queue.
+ * @param queueMetaData receive message from which partition in Message Queue.
+ * @param messageReceipt receive message from which receipt position in Message Queue.
+ * @param timeout receive message will blocked at most
+ *
+ * @param timeout receive messages will blocked at most
+ *
+ * @param queueName receive message from which queueName in Message Queue.
+ * @param queueMetaData receive message from which partition in Message Queue.
+ * @param messageReceipt receive message from which receipt position in Message Queue.
+ * @param timeout receive messages will blocked at most
+ * Messages that have been received but not acknowledged may be redelivered.
+ *
+ * @param receiptHandle the receipt handle associated with the consumed message
+ * @throws OMSRuntimeException if the consumer fails to acknowledge the messages due to some internal error.
+ */
+ void ack(MessageReceipt receiptHandle);
+
+}
\ No newline at end of file
diff --git a/openmessaging-api/src/main/java/io/openmessaging/consumer/PushConsumer.java b/openmessaging-api/src/main/java/io/openmessaging/consumer/PushConsumer.java
new file mode 100644
index 00000000..3eaf5ae6
--- /dev/null
+++ b/openmessaging-api/src/main/java/io/openmessaging/consumer/PushConsumer.java
@@ -0,0 +1,120 @@
+/*
+ * 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.KeyValue;
+import io.openmessaging.MessagingAccessPoint;
+import io.openmessaging.exception.OMSDestinationException;
+import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.exception.OMSSecurityException;
+import io.openmessaging.message.Message;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * A {@code PushConsumer} receives messages from multiple queues, these messages are pushed from
+ * MOM server to {@code PushConsumer} client.
+ *
+ * @version OMS 1.0.0
+ * @see MessagingAccessPoint#createPushConsumer()
+ * @since OMS 1.0.0
+ */
+public interface PushConsumer extends Consumer{
+ /**
+ * Resumes the {@code Consumer} in push model after a suspend.
+ *
+ * This method resumes the {@code Consumer} instance after it was suspended. The instance will not receive new
+ * messages between the suspend and resume calls.
+ *
+ * @throws OMSRuntimeException if the instance has not been suspended.
+ * @see PushConsumer#suspend()
+ */
+ void resume();
+
+ /**
+ * Suspends the {@code Consumer} in push model for later resumption.
+ *
+ * This method suspends the consumer until it is resumed. The consumer will not receive new messages between the
+ * suspend and resume calls.
+ *
+ * This method behaves exactly as if it simply performs the call {@code suspend(0)}.
+ *
+ * @throws OMSRuntimeException if the instance is not currently running.
+ * @see PushConsumer#resume()
+ */
+ void suspend();
+
+ /**
+ * Suspends the {@code Consumer} in push model for later resumption.
+ *
+ * This method suspends the consumer until it is resumed or a specified amount of time has elapsed. The consumer
+ * will not receive new messages during the suspended state.
+ *
+ * This method is similar to the {@link #suspend()} method, but it allows finer control over the amount of time to
+ * suspend, and the consumer will be suspended until it is resumed if the timeout is zero.
+ *
+ * @param timeout the maximum time to suspend in milliseconds.
+ * @throws OMSRuntimeException if the instance is not currently running.
+ */
+ void suspend(long timeout);
+
+ /**
+ * This method is used to find out whether the {@code Consumer} in push model is suspended.
+ *
+ * @return true if this {@code Consumer} is suspended, false otherwise.
+ */
+ boolean isSuspended();
+
+ /**
+ * Bind the {@code Consumer} to a collection of queue, with a {@code MessageListener}.
+ *
+ * {@link MessageListener#onReceived(Message, MessageListener.Context)} will be called when new delivered message is
+ * coming.
+ *
+ * @param queueNames a collection of queues.
+ * @param listener a specified listener to receive new message.
+ * @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(Collection
+ * {@link BatchMessageListener#onReceived(List, BatchMessageListener.Context)} will be called when new delivered
+ * messages is coming.
+ *
+ * @param queueNames a collection of queues.
+ * @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(Collection
+ * After the success call, this consumer won't receive new message from the specified queue any more.
+ *
+ * @param queueNames a collection of queues.
+ */
+ void unbindQueue(Collection
@@ -46,5 +47,5 @@ public interface Extension {
* @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.
*/
- QueueMetaData getQueueMetaData(String queueName);
+ Set
*
+ * In order to improve performance, in some scenarios where message persistence is required, some message middleware
+ * will store messages on multiple partitions in multi servers.
+ *
+ *
+ * In some scenarios, it is very useful to get the relevant partitions meta data for a queue.
+ *
* @version OMS 1.0.0
* @since OMS 1.0.0
*/
@@ -31,40 +38,29 @@
public interface QueueMetaData {
/**
- * In order to improve performance, in some scenarios where message persistence is required, some message middleware
- * will store messages on multiple partitions in multi servers.
- *
- * In some scenarios, it is very useful to get the relevant partitions meta data for a queue.
+ * Set queueName to this Message Queue.
+ * @param queueName
*/
- interface Partition {
- /**
- * Partition identifier
- *
- * @return Partition identifier
- */
- int partitionId();
+ void setQueueName(String queueName);
- /**
- * The host of the server where the partition is located
- *
- *
- * @return The host of the server where the partition is located
- */
- String partitonHost();
- }
+ /**
+ * Set the specified partition.
+ * @param partitionId
+ */
+ void setPartitionId(int partitionId);
/**
- * Queue name
- *
+ * Get partition identifier of target queue.
*
- * @return Queue name.
+ * @return Partition identifier
*/
- String queueName();
+ int partitionId();
/**
- * Get partition list belongs to the {@code queueName}
+ * Queue name
+ *
*
- * @return List of {@link Partition} belongs to the specified queue.
+ * @return Queue name.
*/
- Listtimeout milliseconds.
- * @return the next message 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.
- */
- Message receive(long timeout);
-
- /**
- * Receives the next message from the which bind queue,partition and receiptId of this consumer in pull model.
- * timeout milliseconds.
- * @return the next message 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.
- */
- @Optional
- Message receive(String queueName, int partitionId, long receiptId, long timeout);
-
-
- /**
- * Receive message in asynchronous way. This call doesn't block user's thread, and user's message resolve logic
- * should implement in the {@link MessageListener}.
- * timeout 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.
- */
- Listtimeout 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.
- */
- @Optional
- Listtimeout milliseconds.
+ * @return the next message 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.
+ */
+ Message receive(long timeout);
+
+ /**
+ * Receives the next message from the which bind queue,partition and receiptId of this consumer in pull model.
+ * timeout milliseconds.
+ * @return the next message 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.
+ */
+ @Optional
+ Message receive(String queueName, QueueMetaData queueMetaData, MessageReceipt messageReceipt, long timeout);
+
+ /**
+ * Receive message in asynchronous way. This call doesn't block user's thread, and user's message resolve logic
+ * should implement in the {@link MessageListener}.
+ * timeout 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.
+ */
+ Listtimeout 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.
+ */
+ @Optional
+ List