And OMS exceptions for java

Author: duhengforever

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

@@ -3,10 +3,8 @@
 import io.openmessaging.MessagingAccessPoint;
 import io.openmessaging.OMS;
 import io.openmessaging.common.Result;
-import io.openmessaging.consumer.BindResult;
 import io.openmessaging.consumer.Consumer;
 import io.openmessaging.consumer.ReceiveResult;
-import io.openmessaging.manager.QueueConfig;
 import io.openmessaging.manager.ResourceManager;
 
 public class PullConsumerApp {
@@ -17,15 +15,7 @@ public static void main(String[] args) {
 
         //Fetch a ResourceManager to create Queue resource.
         ResourceManager resourceManager = messagingAccessPoint.resourceManager();
-        Result createQueueResult = resourceManager.createQueue("NS://HELLO_QUEUE", new QueueConfig() {
-            @Override public void setFifo(boolean isFifo) {
-
-            }
-
-            @Override public boolean isFifo() {
-                return false;
-            }
-        });
+        Result createQueueResult = resourceManager.createQueue("NS://HELLO_QUEUE");
         if (createQueueResult.isSuccess()) {
             //Start a PullConsumer to receive messages from the specific queue.
             final Consumer consumer = messagingAccessPoint.createConsumer();
@@ -39,7 +29,7 @@ public void run() {
                 }
             }));
 
-            BindResult bindQueueResult = consumer.bindQueue("NS://HELLO_QUEUE");
+            Result bindQueueResult = consumer.bindQueue("NS://HELLO_QUEUE");
             if (bindQueueResult.isSuccess()) {
                 ReceiveResult receiveResult = consumer.receive(1000);
                 if (receiveResult.isSuccess()) {

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

@@ -23,11 +23,10 @@
 import io.openmessaging.common.Result;
 import io.openmessaging.consumer.Consumer;
 import io.openmessaging.consumer.MessageListener;
-import io.openmessaging.exception.OMSResourceNotExistException;
 import io.openmessaging.manager.ResourceManager;
 
 public class PushConsumerApp {
-    public static void main(String[] args) throws OMSResourceNotExistException {
+    public static void main(String[] args) {
         //Load and start the vendor implementation from a specific OMS driver URL.
         final MessagingAccessPoint messagingAccessPoint =
             OMS.getMessagingAccessPoint("oms:rocketmq://localhost:10911/us-east");

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

@@ -17,13 +17,17 @@
 
 package io.openmessaging.samples.producer;
 
+import io.openmessaging.Future;
 import io.openmessaging.Message;
 import io.openmessaging.MessagingAccessPoint;
 import io.openmessaging.OMS;
-import io.openmessaging.manager.ResourceManager;
+import io.openmessaging.interceptor.Context;
+import io.openmessaging.interceptor.ProducerInterceptor;
 import io.openmessaging.producer.Producer;
 import io.openmessaging.producer.SendResult;
 import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
 
 public class ProducerApp {
     public static void main(String[] args) {
@@ -32,7 +36,16 @@ public static void main(String[] args) {
 
         final Producer producer = messagingAccessPoint.createProducer();
         producer.startup();
-        final ResourceManager manager = messagingAccessPoint.resourceManager();
+        ProducerInterceptor interceptor = new ProducerInterceptor() {
+            @Override
+            public void preSend(Message message, Context attributes) {
+            }
+
+            @Override
+            public void postSend(Message message, Context attributes) {
+            }
+        };
+        producer.addInterceptor(interceptor);
 
         //Register a shutdown hook to close the opened endpoints.
         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@@ -43,18 +56,33 @@ public void run() {
         }));
 
         //Sends a message to the specified destination synchronously.
-        {
-            Message message = producer.createMessage(
-                "NS://HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8")));
-            SendResult sendResult = producer.send(message);
+        Message message = producer.createMessage(
+            "NS://HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8")));
+        SendResult sendResult = producer.send(message);
 
+        if (sendResult.isSuccess()) {
+            System.out.println("Send sync message OK, message id is: " + sendResult.messageId());
+        } else {
+            System.out.println("Error: " + sendResult.getError().getErrorCode() + " error message: " + sendResult.getError().getErrorMessage());
+        }
 
-            if (sendResult.isSuccess()) {
-                System.out.println("Send sync message OK, message id is: " + sendResult.messageId());
-            } else {
-                System.out.println("Error: " + sendResult.getError().getErrorCode() + " error message: " + sendResult.getError().getErrorMessage());
-            }
+        //Sends a message to the specified destination async.
+        Future<SendResult> sendResultFuture = producer.sendAsync(message);
+        sendResult = sendResultFuture.get(1000);
+        if (sendResult.isSuccess()) {
+            System.out.println("Send message async: " + sendResult);
         }
+        //Sends a message to the specified destination in one way mode.
+        producer.sendOneway(message);
 
+        //Sends messages to the specified destination in batch mode.
+        List<Message> messages = new ArrayList<Message>(10);
+        for (int i = 0; i < 10; i++) {
+            Message msg = producer.createMessage("NS://HELLO_QUEUE", ("Hello" + i).getBytes());
+            messages.add(msg);
+        }
+        producer.send(messages);
+        producer.removeInterceptor(interceptor);
+        producer.shutdown();
     }
 }

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

@@ -20,8 +20,8 @@
 import io.openmessaging.Message;
 import io.openmessaging.MessagingAccessPoint;
 import io.openmessaging.OMS;
-import io.openmessaging.producer.CheckListener;
 import io.openmessaging.producer.Producer;
+import io.openmessaging.producer.TransactionStateCheckListener;
 import io.openmessaging.producer.TransactionalResult;
 import java.nio.charset.Charset;
 
@@ -30,7 +30,11 @@ public static void main(String[] args) {
         final MessagingAccessPoint messagingAccessPoint =
             OMS.getMessagingAccessPoint("oms:rocketmq://alice@rocketmq.apache.org/us-east");
 
-        final Producer producer = messagingAccessPoint.createProducer();
+        final Producer producer = messagingAccessPoint.createProducer(new TransactionStateCheckListener() {
+            @Override public void check(Message message, TransactionalContext context) {
+
+            }
+        });
         producer.startup();
 
         //Register a shutdown hook to close the opened endpoints.
@@ -45,14 +49,10 @@ public void run() {
             "NS://HELLO_QUEUE", "HELLO_BODY".getBytes(Charset.forName("UTF-8")));
 
         //Sends a transaction message to the specified destination synchronously.
-        producer.register(new CheckListener() {
-            @Override public void check(Message message, TransactionalContext context) {
-                context.commit();
-            }
-        });
         TransactionalResult result = producer.prepare(message);
         executeLocalTransaction(result);
         result.commit();
+        producer.shutdown();
         System.out.println("Send transaction message OK, message id is: " + result.messageId());
     }
 

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

@@ -22,12 +22,11 @@
 import io.openmessaging.OMS;
 import io.openmessaging.consumer.Consumer;
 import io.openmessaging.consumer.MessageListener;
-import io.openmessaging.exception.OMSResourceNotExistException;
 import io.openmessaging.manager.ResourceManager;
 import io.openmessaging.producer.Producer;
 
 public class RoutingApp {
-    public static void main(String[] args) throws OMSResourceNotExistException {
+    public static void main(String[] args)  {
         //Load and start the vendor implementation from a specific OMS driver URL.
         final MessagingAccessPoint messagingAccessPoint =
             OMS.getMessagingAccessPoint("oms:rocketmq://alice@rocketmq.apache.org/us-east");

openmessaging-api-samples/src/main/java/io/openmessaging/samples/routing/RoutingApp.java

@@ -17,6 +17,8 @@
 
 package io.openmessaging;
 
+import io.openmessaging.exception.OMSMessageFormatException;
+
 /**
  * A factory interface for creating {@code Message} objects.
  *
@@ -31,6 +33,8 @@ public interface MessageFactory {
      * @param queueName the target queue to send
      * @param body the body data for a message
      * @return the created {@code Message} object
+     * @throws OMSMessageFormatException when body exceed the maximum length or others.
+     *
      */
     Message createMessage(String queueName, byte[] body);
 }

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

@@ -19,13 +19,12 @@
 
 import io.openmessaging.common.Result;
 import io.openmessaging.consumer.Consumer;
-import io.openmessaging.consumer.ConsumerConfig;
 import io.openmessaging.consumer.MessageListener;
-import io.openmessaging.consumer.StreamingConsumer;
 import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.exception.OMSSecurityException;
 import io.openmessaging.manager.ResourceManager;
 import io.openmessaging.producer.Producer;
-import io.openmessaging.producer.ProducerConfig;
+import io.openmessaging.producer.TransactionStateCheckListener;
 
 /**
  * An instance of {@code MessagingAccessPoint} may be obtained from {@link OMS}, which is capable of creating {@code
@@ -75,18 +74,21 @@ public interface MessagingAccessPoint extends Result {
      * @return the created {@code Producer}
      * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
      * error
+     * @throws OMSSecurityException if have no authority to create a producer.
      */
     Producer createProducer();
 
     /**
-     * Creates a new {@code Producer} for the specified {@code MessagingAccessPoint} with some preset attributes.
+     * Creates a new transactional {@code Producer} for the specified {@code MessagingAccessPoint}, the producer is able
+     * to respond to requests from the server to check the status of the transaction.
      *
-     * @param producerConfig the preset producer config
+     * @param transactionStateCheckListener transactional check listener {@link TransactionStateCheckListener}
      * @return the created {@code Producer}
      * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
      * error
+     * @throws OMSSecurityException if have no authority to create a producer.
      */
-    Producer createProducer(ProducerConfig producerConfig);
+    Producer createProducer(TransactionStateCheckListener transactionStateCheckListener);
 
     /**
      * Creates a new {@code PushConsumer} for the specified {@code MessagingAccessPoint}. The returned {@code Consumer}
@@ -95,46 +97,17 @@ public interface MessagingAccessPoint extends Result {
      * @return the created {@code PushConsumer}
      * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
      * error
+     * @throws OMSSecurityException if have no authority to create a consumer.
      */
     Consumer createConsumer();
 
-    /**
-     * Creates a new {@code PushConsumer} for the specified {@code MessagingAccessPoint}. The returned {@code Consumer}
-     * isn't bind to any queue, uses {@link Consumer#bindQueue(String, MessageListener)} to bind queues.
-     *
-     * @param consumerConfig the preset config
-     * @return the created {@code PushConsumer}
-     * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
-     * error
-     */
-    Consumer createConsumer(ConsumerConfig consumerConfig);
-
-    /**
-     * Creates a new {@code StreamingConsumer} for the specified {@code MessagingAccessPoint}.
-     *
-     * @return the created {@code Stream}
-     * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
-     * error
-     */
-    StreamingConsumer createStreamingConsumer();
-
-    /**
-     * Creates a new {@code StreamingConsumer} for the specified {@code MessagingAccessPoint} with some preset
-     * attributes.
-     *
-     * @param consumerConfig the preset client config
-     * @return the created consumer
-     * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
-     * error
-     */
-    StreamingConsumer createStreamingConsumer(ConsumerConfig consumerConfig);
-
     /**
      * Gets a lightweight {@code ResourceManager} instance from the specified {@code MessagingAccessPoint}.
      *
      * @return the resource manger
      * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request due to some internal
      * error
+     * @throws OMSSecurityException if have no authority to obtain a resource manager.
      */
     ResourceManager resourceManager();
 }

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

@@ -46,16 +46,6 @@ public interface OMSBuiltinKeys {
      */
     String REGION = "REGION";
 
-    /**
-     * The {@code PRODUCER_ID} key represents the the unique producer id of a producer instance.
-     */
-    String PRODUCER_ID = "PRODUCER_ID";
-
-    /**
-     * The {@code CONSUMER_ID} key represents the the unique consumer id of a consumer instance.
-     */
-    String CONSUMER_ID = "CONSUMER_ID";
-
     /**
      * The {@code OPERATION_TIMEOUT} key defines the timeout of almost all the method calls in OMS.
      */

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

@@ -49,6 +49,6 @@ public interface ServiceLifecycle {
      * Used for get service current state, for execution of some operations is dependent on the current service state.
      * @return This service current state {@link ServiceLifeState}
      */
-    ServiceLifeState getServiceLifeState();
+    ServiceLifeState currentState();
 
 }