Merge pull request #136 from ayeshLK/2201.3.x-dev

[2201.3.x] Update value type of the application properties to `map<anydata>` in `asb:Message`

asb-ballerina/Ballerina.toml

@@ -2,7 +2,7 @@
 distribution = "2201.3.1"
 org = "ballerinax"
 name = "asb"
-version = "3.0.3"
+version = "3.0.4"
 license= ["Apache-2.0"]
 authors = ["Ballerina"]
 keywords = ["IT Operations/Message Brokers", "Cost/Paid", "Vendor/Microsoft"]
@@ -13,9 +13,9 @@ repository = "https://github.com/ballerina-platform/module-ballerinax-azure-serv
 observabilityIncluded = true
 
 [[platform.java11.dependency]]
-path = "../asb-native/build/libs/asb-native-3.0.3.jar"
+path = "../asb-native/build/libs/asb-native-3.0.4.jar"
 groupId = "org.ballerinax"
 artifactId = "asb-native"
 module = "asb-native" 
-version = "3.0.3"
+version = "3.0.4"
 

asb-ballerina/Dependencies.toml

@@ -104,7 +104,7 @@ modules = [
 [[package]]
 org = "ballerinax"
 name = "asb"
-version = "3.0.3"
+version = "3.0.4"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"},
 	{org = "ballerina", name = "log"},

asb-ballerina/message.bal

@@ -91,5 +91,5 @@ public type Message record {|
 @display {label: "Application Properties"}
 public type ApplicationProperties record {|
     @display {label: "Properties"}
-    map<any> properties?;
+    map<anydata> properties?;
 |};

asb-ballerina/tests/asb_test.bal

@@ -31,7 +31,7 @@ string stringContent = "This is ASB connector test-Message Body";
 byte[] byteContent = stringContent.toBytes();
 json jsonContent = {name: "wso2", color: "orange", price: 5.36};
 byte[] byteContentFromJson = jsonContent.toJsonString().toBytes();
-map<any> properties = {a: "propertyValue1", b: "propertyValue2", c: 1, d: "true", f: 1.345, s: false, k:1020202, g: jsonContent};
+map<anydata> properties = {a: "propertyValue1", b: "propertyValue2", c: 1, d: "true", f: 1.345, s: false, k:1020202, g: jsonContent};
 int timeToLive = 60; // In seconds
 int serverWaitTime = 60; // In seconds
 int maxMessageCount = 2;
@@ -94,7 +94,7 @@ function testSendAndReceiveMessageFromQueueOperation() returns error? {
     if (messageReceived is Message) {
         var result = check messageReceiver->complete(messageReceived);
         test:assertEquals(result, (), msg = "Complete message not successful.");
-        float mapValue = <float> check getApplicationPropertyByName(messageReceived, "f");
+        float mapValue = check getApplicationPropertyByName(messageReceived, "f").ensureType();
         test:assertEquals(mapValue, <float>properties["f"], "Retrieving application properties failed");
     } else if (messageReceived is ()) {
         test:assertFail("No message in the queue.");

asb-ballerina/utils.bal

@@ -19,8 +19,8 @@
 # + message - Represents a message that contains application properties
 # + name - Represents the name of the application property that the user wants to retrieve.
 # + return - Returns null if the requested property does not exist
-public isolated function getApplicationPropertyByName(Message message, string name) returns any|error? {
+public isolated function getApplicationPropertyByName(Message message, string name) returns anydata|error? {
         ApplicationProperties applicationPropertiesResult = check message.applicationProperties.ensureType();
-        map<any> properties = check applicationPropertiesResult.properties.ensureType();
+        map<anydata> properties = check applicationPropertiesResult.properties.ensureType();
         return properties[name];
 }

asb-native/src/main/java/org/ballerinax/asb/receiver/MessageReceiver.java

@@ -28,7 +28,11 @@
 import com.azure.messaging.servicebus.ServiceBusClientBuilder.ServiceBusReceiverClientBuilder;
 import com.azure.messaging.servicebus.models.DeadLetterOptions;
 import com.azure.messaging.servicebus.models.ServiceBusReceiveMode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.ballerina.runtime.api.PredefinedTypes;
+import io.ballerina.runtime.api.creators.TypeCreator;
 import io.ballerina.runtime.api.creators.ValueCreator;
+import io.ballerina.runtime.api.types.MapType;
 import io.ballerina.runtime.api.utils.StringUtils;
 import io.ballerina.runtime.api.utils.TypeUtils;
 import io.ballerina.runtime.api.values.BMap;
@@ -58,6 +62,7 @@
  */
 public class MessageReceiver {
     private static final Logger log = Logger.getLogger(MessageReceiver.class);
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
     private ServiceBusReceiverClient receiver;
 
     /**
@@ -256,16 +261,49 @@ private BMap<BString, Object> getReceivedMessage(BObject endpointClient, Service
         map.put("deadLetterReason", StringUtils.fromString(receivedMessage.getDeadLetterReason()));
         map.put("deadLetterSource", StringUtils.fromString(receivedMessage.getDeadLetterSource()));
         map.put("state", StringUtils.fromString(receivedMessage.getState().toString()));
-        BMap<BString, Object> applicationProperties = ValueCreator.createRecordValue(ModuleUtils.getModule(),
-                ASBConstants.APPLICATION_PROPERTY_TYPE);
-        Object appProperties = ASBUtils.toBMap(receivedMessage.getApplicationProperties());
-        map.put("applicationProperties", ValueCreator.createRecordValue(applicationProperties, appProperties));
+        map.put("applicationProperties", getApplicationProperties(receivedMessage));
         BMap<BString, Object> createRecordValue = ValueCreator.createRecordValue(ModuleUtils.getModule(),
                 ASBConstants.MESSAGE_RECORD, map);
         endpointClient.addNativeData(receivedMessage.getLockToken(), receivedMessage);
         return createRecordValue;
     }
 
+    private static BMap<BString, Object> getApplicationProperties(ServiceBusReceivedMessage message) {
+        BMap<BString, Object> applicationPropertiesRecord = ValueCreator.createRecordValue(ModuleUtils.getModule(),
+                ASBConstants.APPLICATION_PROPERTY_TYPE);
+        MapType mapType = TypeCreator.createMapType(PredefinedTypes.TYPE_ANYDATA);
+        BMap<BString, Object> applicationProperties = ValueCreator.createMapValue(mapType);
+        for (Map.Entry<String, Object> property: message.getApplicationProperties().entrySet()) {
+            populateApplicationProperty(applicationProperties, property.getKey(), property.getValue());
+        }
+        return ValueCreator.createRecordValue(applicationPropertiesRecord, applicationProperties);
+    }
+    private static void populateApplicationProperty(BMap<BString, Object> applicationProperties,
+                                                    String key, Object value) {
+        BString propertyKey = StringUtils.fromString(key);
+        if (value instanceof String) {
+            applicationProperties.put(propertyKey, StringUtils.fromString((String) value));
+        } else if (value instanceof Integer) {
+            applicationProperties.put(propertyKey, (Integer) value);
+        } else if (value instanceof Long) {
+            applicationProperties.put(propertyKey, (Long) value);
+        } else if (value instanceof Float) {
+            applicationProperties.put(propertyKey, (Float) value);
+        } else if (value instanceof Double) {
+            applicationProperties.put(propertyKey, (Double) value);
+        } else if (value instanceof Boolean) {
+            applicationProperties.put(propertyKey, (Boolean) value);
+        } else if (value instanceof Character) {
+            applicationProperties.put(propertyKey, (Character) value);
+        } else if (value instanceof Byte) {
+            applicationProperties.put(propertyKey, (Byte) value);
+        } else if (value instanceof Short) {
+            applicationProperties.put(propertyKey, (Short) value);
+        } else {
+            applicationProperties.put(propertyKey, StringUtils.fromString(value.toString()));
+        }
+    }
+
     /**
      * Receive Batch of Messages with configurable parameters as Map when Receiver
      * Connection is given as a parameter,

gradle.properties

@@ -1,4 +1,4 @@
 org.gradle.caching=true
 group=org.ballerinax.azure.servicebus
-version=3.0.3
+version=3.0.4
 ballerinaLangVersion=2201.3.1