Merge pull request #61 from duhenglucky/PubSubModel

feat(OMS) polish MessagingAccessPoint builder

Author: duhenglucky

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

@@ -30,7 +30,12 @@ public class PullConsumerApp {
     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");
+            OMS.builder()
+                .endpoint("http://mq-instance-xxx-1234567890-test:8080")
+                .region("Shenzhen")
+                .driver("rocketmq")
+                .build();
+
         Properties properties = new Properties();
         //Start a PullConsumer to receive messages from the specific queue.
         final PullConsumer consumer = messagingAccessPoint.createPullConsumer(properties);

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

@@ -30,7 +30,12 @@ public class PushConsumerApp {
     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:9876");
+            OMS.builder()
+                .region("Shenzhen")
+                .endpoint("127.0.0.1:9876")
+                .driver("rocketmq")
+                .withCredentials(new Properties())
+                .build();
 
         Properties properties = new Properties();
         final Consumer consumer = messagingAccessPoint.createConsumer(properties);

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

@@ -29,7 +29,12 @@
 public class ProducerApp {
     public static void main(String[] args) {
         final MessagingAccessPoint messagingAccessPoint =
-            OMS.getMessagingAccessPoint("oms:rocketmq://alice@rocketmq.apache.org");
+            OMS.builder()
+                .region("shanghai,shenzhen")
+                .endpoint("127.0.0.1:9876")
+                .driver("rocketmq")
+                .withCredentials(new Properties())
+                .build();
 
         final Producer producer = messagingAccessPoint.createProducer(new Properties());
         producer.start();

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

@@ -30,7 +30,12 @@
 public class TransactionProducerApp {
     public static void main(String[] args) {
         final MessagingAccessPoint messagingAccessPoint =
-            OMS.getMessagingAccessPoint("oms:rocketmq://alice@rocketmq.apache.org/us-east");
+            OMS.builder()
+                .region("Shenzhen")
+                .endpoint("127.0.0.1:9876")
+                .driver("rocketmq")
+                .withCredentials(new Properties())
+                .build();
 
         final TransactionProducer producer = messagingAccessPoint.createTransactionProducer(new Properties(), new LocalTransactionChecker() {
             @Override

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

@@ -54,11 +54,10 @@ public interface MessagingAccessPoint {
      * <p>
      * There are some standard attributes defined by OMS for {@code MessagingAccessPoint}:
      * <ul>
-     * <li> {@link OMSBuiltinKeys#ACCESS_POINTS}, the specified access points.
+     * <li> {@link OMSBuiltinKeys#ENDPOINT}, the specified access points.
      * <li> {@link OMSBuiltinKeys#DRIVER_IMPL}, the fully qualified class name of the specified MessagingAccessPoint's
      * implementation, the default value is {@literal io.openmessaging.<driver_type>.MessagingAccessPointImpl}.
      * <li> {@link OMSBuiltinKeys#REGION}, the region the resources reside in.
-     * <li> {@link OMSBuiltinKeys#ACCOUNT_ID}, the ID of the specific account system that owns the resource.
      * </ul>
      *
      * @return the attributes

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

@@ -17,57 +17,133 @@
 
 package io.openmessaging.api;
 
-import io.openmessaging.api.exception.OMSRuntimeException;
 import io.openmessaging.api.internal.MessagingAccessPointAdapter;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 
 /**
+ * <p>
  * The oms class provides some static methods to create a {@code MessagingAccessPoint} from the specified OMS driver url
  * and some useful util methods.
- * <p>
- * The complete OMS driver URL syntax is:
- * <p>
- * {@literal oms:<driver_type>://[account_id@]host1[:port1][,host2[:port2],...[,hostN[:portN]]]/<region>}
- * <p>
- * The first part of the URL specifies which OMS implementation is to be used, rocketmq is an optional driver type.
+ * </p>
+ *
  * <p>
  * The brackets indicate that the extra access points are optional, and a correct OMS driver url needs at least one
  * access point, which consists of hostname and port, like localhost:8081.
+ * </p>
  *
- * @version  OMS 1.1.0
- * @since  OMS 1.1.0
+ * @version OMS 1.2.0
+ * @since OMS 1.1.0
  */
 public final class OMS {
+
+    private final Properties properties = new Properties();
+
+    public static OMS builder() {
+        return new OMS();
+    }
+
     /**
-     * Returns a {@code MessagingAccessPoint} instance from the specified OMS driver url.
+     * Set the endpoint provided by messaging vendor.
      *
-     * @param url the specified OMS driver url
-     * @return a {@code MessagingAccessPoint} instance
-     * @throws OMSRuntimeException if the factory fails to create a {@code MessagingAccessPoint} due to some driver url
-     * some syntax error or internal error.
+     * @param endpoint
+     * @return
      */
-    public static MessagingAccessPoint getMessagingAccessPoint(String url) {
-        return getMessagingAccessPoint(url, new Properties());
+    public OMS endpoint(String endpoint) {
+        this.properties.put(OMSBuiltinKeys.ENDPOINT, endpoint);
+        return this;
     }
 
     /**
-     * Returns a {@code MessagingAccessPoint} instance from the specified OMS driver url with some preset attributes,
-     * which will be passed to MessagingAccessPoint's implementation class as a unique constructor parameter.
+     * Set the region provided by messaging vendor.
      *
-     * There are some standard attributes defined by OMS for this method, the same as {@link
-     * MessagingAccessPoint#attributes()} ()}
+     * @param region
+     * @return
+     */
+    public OMS region(String region) {
+        this.properties.put(OMSBuiltinKeys.REGION, region);
+        return this;
+    }
+
+    /**
+     * <p>
+     * Set the the driver type of the specified MessagingAccessPoint's * implementation, the default value is {@literal
+     * io.openmessaging.<driver_type>.MessagingAccessPointImpl}.
+     * </p>
+     *
+     * <p>
+     * But if the {@link OMS#driverImpl(String)} attribute was set, this attribute will be ignored.
+     * </p>
      *
-     * @param url the specified OMS driver url
-     * @return a {@code MessagingAccessPoint} instance
-     * @throws OMSRuntimeException if the factory fails to create a {@code MessagingAccessPoint} due to some driver url
-     * some syntax error or internal error.
+     * @param driver
+     * @return
      */
-    public static MessagingAccessPoint getMessagingAccessPoint(String url, Properties attributes) {
-        return MessagingAccessPointAdapter.getMessagingAccessPoint(url, attributes);
+    public OMS driver(String driver) {
+        this.properties.put(OMSBuiltinKeys.DRIVER, driver);
+        return this;
     }
 
+    /**
+     * <p>
+     * Set the the fully qualified class name of the specified MessagingAccessPoint's * implementation, the default
+     * value is {@literal io.openmessaging.<driver_type>.MessagingAccessPointImpl}.
+     * </p>
+     *
+     * <p>
+     * If this attribute was set, {@link OMS#driver(String)} will be ignored.
+     * </p>
+     *
+     * @param driverImpl
+     * @return
+     */
+    public OMS driverImpl(String driverImpl) {
+        this.properties.put(OMSBuiltinKeys.DRIVER_IMPL, driverImpl);
+        return this;
+    }
+
+    /**
+     * <p>
+     * Set credentials used by the client.
+     * </p>
+     *
+     * @param credentials provided by vendors.
+     * @return
+     */
+    public OMS withCredentials(Properties credentials) {
+        if (credentials.getProperty(OMSBuiltinKeys.ACCESS_KEY) != null) {
+            this.properties.put(OMSBuiltinKeys.ACCESS_KEY, credentials.getProperty(OMSBuiltinKeys.ACCESS_KEY));
+        }
+        if (credentials.getProperty(OMSBuiltinKeys.SECRET_KEY) != null) {
+            this.properties.put(OMSBuiltinKeys.SECRET_KEY, credentials.getProperty(OMSBuiltinKeys.SECRET_KEY));
+        }
+        if (credentials.getProperty(OMSBuiltinKeys.SECURITY_TOKEN) != null) {
+            this.properties.put(OMSBuiltinKeys.SECURITY_TOKEN, credentials.getProperty(OMSBuiltinKeys.SECURITY_TOKEN));
+        }
+        return this;
+    }
+
+    public MessagingAccessPoint build() {
+        return MessagingAccessPointAdapter.getMessagingAccessPoint(this.properties);
+    }
+
+    /**
+     * Set extra custom configs.
+     *
+     * @param config extra configs
+     * @return
+     */
+    public MessagingAccessPoint build(Properties config) {
+        Set<Map.Entry<Object, Object>> entrySet = config.entrySet();
+        for (Map.Entry<Object, Object> entry : entrySet) {
+            if (!this.properties.containsKey(entry.getKey())) {
+                this.properties.put(entry.getKey(), entry.getValue());
+            }
+        }
+        return this.build(properties);
+    }
 
     /**
      * The version format is X.Y.Z (Major.Minor.Patch), a pre-release version may be denoted by appending a hyphen and a

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

@@ -27,34 +27,37 @@ public interface OMSBuiltinKeys {
     /**
      * The {@code DRIVER_IMPL} key represents the vendor implementation entry of {@link MessagingAccessPoint}.
      */
-    String DRIVER_IMPL = "DRIVER_IMPL";
-
-    /**
-     * The {@code ACCESS_POINTS} key shows the specified access points in OMS driver schema.
-     *
-     * @see <a href="https://github.com/openmessaging/specification/blob/master/oms_access_point_schema.md">Access Point
-     * Schema</a>
-     */
-    String ACCESS_POINTS = "ACCESS_POINTS";
+    String DRIVER_IMPL = "driverImpl";
 
     /**
      * The {@code ACCESS_KEY} key shows the specified access key in OMS driver schema.
      */
-    String ACCESS_KEY = "AccessKey";
+    String ACCESS_KEY = "accessKey";
 
     /**
      * The {@code SECRET_KEY} key shows the specified secret key in OMS attribute.
      */
-    String SECRET_KEY = "SecretKey";
+    String SECRET_KEY = "secretKey";
 
     /**
      * The {@code SECURITY_TOKEN} key shows the specified security token in OMS attribute.
      */
-    String SECURITY_TOKEN = "SecurityToken";
+    String SECURITY_TOKEN = "securityToken";
 
     /**
      * The {@code REGION} key shows the specified region in OMS driver schema.
      */
-    String REGION = "REGION";
+    String REGION = "region";
+
+    /**
+     * The {@code ENDPOINT} key shows the specified host in OMS attribute.
+     */
+    String ENDPOINT = "endpoint";
+
+    /**
+     * The {@code DRIVER} key represents the vendor type of {@link MessagingAccessPoint}, but if {@code DRIVER_IMPL} is
+     * not empty, this {@code DRIVER} value will be ignored.
+     */
+    String DRIVER = "driver";
 
 }

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

@@ -17,7 +17,7 @@
 package io.openmessaging.api;
 
 /**
- * Used for describe a topic and
+ * Used for describe a topic
  * @version OMS 1.2.0
  * @since OMS 1.2.0
  */
@@ -80,4 +80,11 @@ public boolean equals(Object obj) {
         }
         return true;
     }
+
+    @Override public String toString() {
+        return "TopicPartition{" +
+            "topic='" + topic + '\'' +
+            ", partition='" + partition + '\'' +
+            '}';
+    }
 }