Merge pull request #15334 from riyafa/jms

Add data binding support for Artemis Connector

examples/artemis-anycast-session-consumer/artemis_anycast_session_consumer.out

@@ -1,7 +1,7 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_multicast_consumer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_anycast_session_consumer.bal' file and execute the
 # 'ballerina run' command as follows.
-$ ballerina run artemis_multicast_consumer.bal
-# The ActiveMQ Artemis consumer runs as a Ballerina service and listens to the subscribed queue.
+$ ballerina run artemis_anycast_session_consumer.bal
+[ballerina/artemis] Client Consumer created for queue queue1
 

examples/artemis-anycast-session-producer/artemis_anycast_session_producer.out

@@ -1,5 +1,5 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_anycast_session_producer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_anycast_session_producer.bal' file and execute the
 # 'ballerina run' command as follows.
 $ ballerina run artemis_anycast_session_producer.bal

examples/artemis-data-binding/artemis_data_binding.bal

@@ -0,0 +1,37 @@
+import ballerina/artemis;
+import ballerina/http;
+import ballerina/log;
+
+type Order record {
+    string description?;
+    int id;
+    float cost;
+};
+
+// Consumer listens to the queue (i.e., "my_queue") with the address
+// (i.e., "my_address").
+@artemis:ServiceConfig {
+    queueConfig: {
+        queueName: "my_queue",
+        addressName: "my_address"
+    }
+}
+// Attaches the service to the listener.
+service artemisConsumer on new artemis:Listener(
+                            { host: "localhost", port: 61616 }) {
+
+    // This resource is triggered when a valid `Order` is received.
+    resource function onMessage(artemis:Message message, Order orderDetails)
+                         returns error? {
+
+        // Posts order details to the backend and awaits response.
+        http:Client clientEP = new("http://www.mocky.io");
+        var response = clientEP->post("/v2/5cde49ef3000005e004307f0",
+                             untaint check json.convert(orderDetails));
+        if (response is http:Response) {
+            log:printInfo(check response.getTextPayload());
+        } else {
+            log:printError("Invalid response ", err = response);
+        }
+    }
+}

examples/artemis-data-binding/artemis_data_binding.description

@@ -0,0 +1,4 @@
+// Artemis data binding helps access payload through the last resource 
+// parameter. `string`, `json`, `xml`, `byte[]`, `record`, `map<string>`, `map<int>`, 
+// `map<float>`, `map<byte>`, `map<byte[]>` and `map<boolean>` are supported as
+// parameter types.

examples/artemis-data-binding/artemis_data_binding.out

@@ -0,0 +1,6 @@
+# Make sure to have the ActiveMQ Artemis broker running.
+
+# Navigate to the directory that contains the 'artemis_data_binding.bal' file
+# and execute the 'ballerina run' command as follows
+$ ballerina run artemis_data_binding.bal 
+[ballerina/artemis] Client Consumer created for queue my_queue

examples/artemis-multicast-consumer/artemis_multicast_consumer.out

@@ -1,7 +1,7 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_multicast_consumer.bal' file and issue the
-# 'ballerina run' command.
-$ ballerina run artemis_multicast_consumer.bal as follows.
-# The ActiveMQ Artemis consumer runs as a Ballerina service and listens to the subscribed queue.
+# Navigate to the directory that contains the 'artemis_multicast_consumer.bal' file and execute the
+# 'ballerina run' command as follows.
+$ ballerina run artemis_multicast_consumer.bal
+[ballerina/artemis] Client Consumer created for queue my_queue
 

examples/artemis-multicast-producer/artemis_multicast_producer.out

@@ -1,5 +1,5 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_multicast_producer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_multicast_producer.bal' file and execute the
 # 'ballerina run' command as follows.
 $ ballerina run artemis_multicast_producer.bal

examples/artemis-simple-transaction-consumer/artemis_simple_transaction_consumer.out

@@ -1,9 +1,6 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_simple_transaction_consumer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_simple_transaction_consumer.bal' file and execute the
 # 'ballerina run' command as follows.
 $ ballerina run --experimental artemis_simple_transaction_consumer.bal
-# The ActiveMQ Artemis consumer runs as a Ballerina service and listens to the subscribed queue.
-
-[ballerina/http] started HTTP/WS endpoint 172.17.0.1:34985
-2019-04-18 15:19:32,475 INFO  [ballerina/transactions] - Created transaction: 29787878-3686-41b3-9ab3-4bc60ad278f6
+2019-05-17 14:11:29,258 INFO  [ballerina/transactions] - Created transaction: 78aabdba-3994-47cd-b083-80744961f69b

examples/artemis-simple-transaction-producer/artemis_simple_transaction_producer.out

@@ -1,6 +1,6 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_simple_transaction_producer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_simple_transaction_producer.bal' file and execute the
 # 'ballerina run' command as follows.
 $ ballerina run --experimental artemis_simple_transaction_producer.bal
 # The ActiveMQ Artemis producer runs as a Ballerina service and listens to the subscribed queue.

examples/artemis-transaction-producer/artemis_transaction_producer.out

@@ -1,6 +1,6 @@
 # Make sure to have the ActiveMQ Artemis broker running.
 
-# Navigate to the directory that contains the 'artemis_transaction_producer.bal' file and issue the
+# Navigate to the directory that contains the 'artemis_transaction_producer.bal' file and execute the
 # 'ballerina run' command as follows.
 $ ballerina run --experimental artemis_transaction_producer.bal
 # The ActiveMQ Artemis producer runs as a Ballerina service and listens to the subscribed queue.

examples/index.json

@@ -998,6 +998,10 @@
             {
                 "name": "Transaction Producer",
                 "url": "artemis-transaction-producer"
+            },
+            {
+                "name": "Data Binding",
+                "url": "artemis-data-binding"
             }
         ]
     },

stdlib/messaging/activemq-artemis/build.gradle

@@ -24,23 +24,27 @@ dependencies {
 //    implementation project(':ballerina-lang')
     implementation project(':ballerina-io')
     implementation project(':ballerina-builtin')
+    implementation project(':ballerina-time')
+    implementation project(':ballerina-crypto')
+    implementation project(':ballerina-filepath')
+    implementation project(':ballerina-log-api')
+    implementation project(':ballerina-system')
     implementation project(':ballerina-runtime-api')
     implementation project(':ballerina-utils')
     implementation project(':ballerina-launcher')
     implementation 'org.apache.activemq:artemis-core-client'
 
-    baloImplementation project(path: ':ballerina-runtime-api', configuration: 'baloImplementation')
-    baloImplementation project(path: ':ballerina-utils', configuration: 'baloImplementation')
-    baloImplementation project(path: ':ballerina-builtin', configuration: 'baloImplementation')
     baloImplementation project(path: ':ballerina-io', configuration: 'baloImplementation')
+    baloImplementation project(path: ':ballerina-builtin', configuration: 'baloImplementation')
     baloImplementation project(path: ':ballerina-time', configuration: 'baloImplementation')
     baloImplementation project(path: ':ballerina-crypto', configuration: 'baloImplementation')
     baloImplementation project(path: ':ballerina-filepath', configuration: 'baloImplementation')
-    baloImplementation project(path: ':ballerina-system', configuration: 'baloImplementation')
     baloImplementation project(path: ':ballerina-log-api', configuration: 'baloImplementation')
+    baloImplementation project(path: ':ballerina-system', configuration: 'baloImplementation')
+    baloImplementation project(path: ':ballerina-runtime-api', configuration: 'baloImplementation')
+    baloImplementation project(path: ':ballerina-utils', configuration: 'baloImplementation')
 
     testCompile 'org.testng:testng'
-
 }
 
 test {
@@ -50,6 +54,11 @@ test {
             into "$buildDir/lib/repo/ballerina"
         }
     }
+    useTestNG() {
+        suites 'src/test/resources/testng.xml'
+    }
+    systemProperty "java.util.logging.config.file", "src/test/resources/logging.properties"
+    systemProperty "java.util.logging.manager", "org.ballerinalang.logging.BLogManager"
 }
 
 description = 'Ballerina - ActiveMQ Artemis'

stdlib/messaging/activemq-artemis/src/main/ballerina/artemis/artemis_commons.bal

@@ -17,6 +17,7 @@
 import ballerina/crypto;
 import ballerina/filepath;
 import ballerina/io;
+import ballerina/time;
 
 # Constant for the artemis error code.
 public const ARTEMIS_ERROR_CODE = "{ballerina/artemis}ArtemisError";

stdlib/messaging/activemq-artemis/src/main/ballerina/artemis/message.bal

@@ -107,12 +107,20 @@ public type Message client object {
 # + priority - the message priority (between 0 and 9 inclusive)
 # + durable - whether the created message is durable or not
 # + routingType - `RoutingType` of the message
+# + groupId - used to group messages so that the same consumer receives all the messages with a particular groupId
+# + groupSequence - can use to specify a sequence within the group
+# + correlationId - a header for associating the current message with some previous message or application-specific ID
+# + replyTo - indicates which address a JMS consumer should reply to
 public type MessageConfiguration record {|
-    int? expiration = ();
-    int? timeStamp = ();
+    int expiration = 0;
+    int timeStamp = time:currentTime().time;
     byte priority = 0;
     boolean durable = true;
     RoutingType? routingType = ();
+    string? groupId = ();
+    int groupSequence = 0;
+    string? correlationId = ();
+    string? replyTo = ();
 |};
 
 # ActiveMQ Artemis message types.

stdlib/messaging/activemq-artemis/src/main/java/org/ballerinalang/messaging/artemis/ArtemisConnectorException.java

@@ -0,0 +1,90 @@
+/*
+*  Copyright (c) 2019, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+*
+*  WSO2 Inc. 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 org.ballerinalang.messaging.artemis;
+
+import org.ballerinalang.bre.Context;
+import org.ballerinalang.util.exceptions.BallerinaException;
+
+/**
+ * BallerinaException that could occur in Artemis connector.
+ *
+ * @since 0.995
+ */
+public class ArtemisConnectorException extends BallerinaException {
+    private static final long serialVersionUID = 381055783364464822L;
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with the specified detail message.
+     *
+     * @param message Error Message
+     */
+    public ArtemisConnectorException(String message) {
+        super(message);
+    }
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with error message and ballerina context.
+     *
+     * @param message Error message
+     * @param context Ballerina context
+     */
+    public ArtemisConnectorException(String message, Context context) {
+        super(message, context);
+    }
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with the specified detail message and cause.
+     *
+     * @param message Error message
+     * @param cause   Cause
+     */
+    public ArtemisConnectorException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with the specified detail message,
+     * cause and ballerina context.
+     *
+     * @param message Error message
+     * @param cause   Cause
+     * @param context Ballerina context
+     */
+    public ArtemisConnectorException(String message, Throwable cause, Context context) {
+        super(message, cause, context);
+    }
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with the cause.
+     *
+     * @param cause Throwable to be wrap by {@link ArtemisConnectorException}
+     */
+    public ArtemisConnectorException(Throwable cause) {
+        super(cause);
+    }
+
+    /**
+     * Constructs a new {@link ArtemisConnectorException} with ballerina context.
+     *
+     * @param stack Ballerina context
+     */
+    public ArtemisConnectorException(Context stack) {
+        super(stack);
+    }
+}

stdlib/messaging/activemq-artemis/src/main/java/org/ballerinalang/messaging/artemis/ArtemisConstants.java

@@ -34,7 +34,7 @@ public class ArtemisConstants {
     public static final String PROTOCOL_PACKAGE_ARTEMIS = BALLERINA_PACKAGE_PREFIX + ARTEMIS;
 
     // Error related constants
-    static final String ARTEMIS_ERROR_CODE = "{ballerina/artemis}ArtemisError";
+    static final String ARTEMIS_ERROR_CODE = "{" + PROTOCOL_PACKAGE_ARTEMIS + "}ArtemisError";
     static final String ARTEMIS_ERROR_RECORD = "ArtemisError";
     static final String ARTEMIS_ERROR_MESSAGE = "message";
 
@@ -79,7 +79,6 @@ public class ArtemisConstants {
     // Field names for Consumer
     public static final String FILTER = "filter";
     public static final String AUTO_ACK = "autoAck";
-    public static final String BROWSE_ONLY = "browseOnly";
     public static final String QUEUE_CONFIG = "queueConfig";
     public static final String QUEUE_NAME = "queueName";
     public static final String TEMPORARY = "temporary";
@@ -95,6 +94,7 @@ public class ArtemisConstants {
     public static final String EXPIRATION = "expiration";
     public static final String TIME_STAMP = "timeStamp";
     public static final String PRIORITY = "priority";
+    public static final String MESSAGE_CONFIG = "configuration";
 
     // Field names for Producer
     public static final String RATE = "rate";
@@ -103,8 +103,13 @@ public class ArtemisConstants {
     // Common field names
     public static final String DURABLE = "durable";
     public static final String ROUTING_TYPE = "routingType";
+    public static final String GROUP_ID = "groupId";
+    public static final String GROUP_SEQUENCE = "groupSequence";
+    public static final String CORRELATION_ID = "correlationId";
+    public static final String REPLY_TO = "replyTo";
     public static final String AUTO_CREATED = "autoCreated";
-    static final String MULTICAST = "MULTICAST";
+    public static final String MULTICAST = "MULTICAST";
+    public static final String ANYCAST = "ANYCAST";
 
     // Field name for Session
     public static final String USERNAME = "username";