Integration test scenario for Service Bus binder

eng/versioning/version_client.txt

@@ -179,6 +179,7 @@ com.azure.spring:azure-spring-boot-test-keyvault;1.0.0;1.0.0
 com.azure.spring:azure-spring-boot-test-keyvault-reactive;1.0.0;1.0.0
 com.azure.spring:azure-spring-boot-test-parent;1.0.0;1.0.0
 com.azure.spring:azure-spring-boot-test-servicebus-jms;1.0.0;1.0.0
+com.azure.spring:azure-spring-cloud-test-servicebus-binder;1.0.0;1.0.0
 com.azure.spring:azure-spring-cloud-test-eventhubs;1.0.0;1.0.0
 com.azure.spring:azure-spring-cloud-test-storage;1.0.0;1.0.0
 com.azure.spring:azure-spring-cloud-test-appconfiguration-config;1.0.0;1.0.0

sdk/spring/azure-spring-cloud-test-servicebus-binder/CHANGELOG.md

@@ -0,0 +1,5 @@
+# Release History
+
+## 1.0.0 (Unreleased)
+
+

sdk/spring/azure-spring-cloud-test-servicebus-binder/README.md

@@ -0,0 +1,8 @@
+# Azure Spring Cloud Integration tests client library for Java
+
+## Key concepts
+## Getting started
+## Examples
+## Troubleshooting
+## Next steps
+## Contributing

sdk/spring/azure-spring-cloud-test-servicebus-binder/pom.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>com.azure.spring</groupId>
+    <artifactId>azure-spring-boot-test-parent</artifactId>
+    <version>1.0.0</version> <!-- {x-version-update;com.azure.spring:azure-spring-boot-test-parent;current} -->
+    <relativePath>../azure-spring-boot-test-parent</relativePath>
+  </parent>
+
+  <groupId>com.azure.spring</groupId>
+  <artifactId>azure-spring-cloud-test-servicebus-binder</artifactId>
+  <version>1.0.0</version> <!-- {x-version-update;com.azure.spring:azure-spring-cloud-test-servicebus-binder;current} -->
+
+  <dependencies>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>azure-spring-cloud-stream-binder-servicebus-queue</artifactId>
+      <version>2.5.0-beta.1</version> <!-- {x-version-update;com.azure.spring:azure-spring-cloud-stream-binder-servicebus-queue;current} -->
+    </dependency>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>azure-spring-cloud-stream-binder-servicebus-topic</artifactId>
+      <version>2.5.0-beta.1</version> <!-- {x-version-update;com.azure.spring:azure-spring-cloud-stream-binder-servicebus-topic;current} -->
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.junit.vintage</groupId>
+          <artifactId>junit-vintage-engine</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>com.azure.spring</groupId>
+      <artifactId>azure-spring-boot-test-core</artifactId>
+      <version>1.0.0</version> <!-- {x-version-update;com.azure.spring:azure-spring-boot-test-core;current} -->
+    </dependency>
+  </dependencies>
+</project>

sdk/spring/azure-spring-cloud-test-servicebus-binder/src/test/java/com/azure/spring/sample/servicebus/binder/ServiceBusQueueAndTopicBinderIT.java

@@ -0,0 +1,104 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.spring.sample.servicebus.binder;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.TestPropertySource;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+@SpringBootTest(classes = { ServiceBusQueueAndTopicBinderIT.TestQueueConfig.class,
+    ServiceBusQueueAndTopicBinderIT.TestTopicConfig.class })
+@TestPropertySource(locations = "classpath:application.yml")
+public class ServiceBusQueueAndTopicBinderIT {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueAndTopicBinderIT.class);
+
+    private static String message = UUID.randomUUID().toString();
+
+    private static final AtomicInteger count = new AtomicInteger(0);
+
+    @Autowired
+    private Sinks.Many<Message<String>> manyQueue;
+
+    @Autowired
+    private Sinks.Many<Message<String>> manyTopic;
+
+    @EnableAutoConfiguration
+    public static class TestQueueConfig {
+
+        @Bean
+        public Sinks.Many<Message<String>> manyQueue() {
+            return Sinks.many().unicast().onBackpressureBuffer();
+        }
+
+        @Bean
+        public Supplier<Flux<Message<String>>> queueSupply(Sinks.Many<Message<String>> manyQueue) {
+            return () -> manyQueue.asFlux()
+                                  .doOnNext(m -> LOGGER.info("Manually sending message {}", m))
+                                  .doOnError(t -> LOGGER.error("Error encountered", t));
+        }
+
+        @Bean
+        public Consumer<Message<String>> queueConsume() {
+            return message -> {
+                LOGGER.info("New message received: '{}'", message);
+                Assertions.assertEquals(message.getPayload(), ServiceBusQueueAndTopicBinderIT.message);
+                count.addAndGet(1);
+            };
+        }
+    }
+
+    @EnableAutoConfiguration
+    public static class TestTopicConfig {
+
+        @Bean
+        public Sinks.Many<Message<String>> manyTopic() {
+            return Sinks.many().unicast().onBackpressureBuffer();
+        }
+
+        @Bean
+        public Supplier<Flux<Message<String>>> topicSupply(Sinks.Many<Message<String>> manyTopic) {
+            return () -> manyTopic.asFlux()
+                                  .doOnNext(m -> LOGGER.info("Manually sending message {}", m))
+                                  .doOnError(t -> LOGGER.error("Error encountered", t));
+        }
+
+        @Bean
+        public Consumer<Message<String>> topicConsume() {
+            return message -> {
+                LOGGER.info("New message received: '{}'", message);
+                Assertions.assertEquals(message.getPayload(), ServiceBusQueueAndTopicBinderIT.message);
+                count.addAndGet(1);
+            };
+        }
+    }
+
+    @Test
+    public void testSendAndReceiveMessage() throws InterruptedException {
+        Thread.sleep(1500);
+        GenericMessage<String> genericMessage = new GenericMessage<>(message);
+        manyQueue.emitNext(genericMessage, Sinks.EmitFailureHandler.FAIL_FAST);
+        Thread.sleep(5000);
+        Assertions.assertEquals(1, count.get());
+        Thread.sleep(1500);
+        manyTopic.emitNext(genericMessage, Sinks.EmitFailureHandler.FAIL_FAST);
+        Thread.sleep(5000);
+        Assertions.assertEquals(2, count.get());
+    }
+
+}

sdk/spring/azure-spring-cloud-test-servicebus-binder/src/test/resources/application.yml

@@ -0,0 +1,36 @@
+spring:
+  cloud:
+    stream:
+      function:
+        definition: queueConsume;queueSupply;topicConsume;topicSupply;
+      bindings:
+        topicConsume-in-0:
+          destination: topic1
+          group: topicSub
+        topicSupply-out-0:
+          destination: topic1
+        queueConsume-in-0:
+          binder: servicebus-2
+          destination: queue1
+        queueSupply-out-0:
+          binder: servicebus-2
+          destination: queue1
+      binders:
+        servicebus-1:
+          type: servicebus-topic
+          default-candidate: true
+          environment:
+            spring:
+              cloud:
+                azure:
+                  servicebus:
+                    connection-string: ${SERVICEBUS_BINDER_TEST_CONNECTION_STRING}
+        servicebus-2:
+          type: servicebus-queue
+          default-candidate: false
+          environment:
+            spring:
+              cloud:
+                azure:
+                  servicebus:
+                    connection-string: ${SERVICEBUS_BINDER_TEST_CONNECTION_STRING}

sdk/spring/azure-spring-cloud-test-servicebus-binder/test-resources.json

@@ -0,0 +1,117 @@
+{
+  "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
+  "contentVersion": "1.0.0.0",
+  "parameters": {
+    "baseName": {
+      "defaultValue": "[resourceGroup().name]",
+      "type": "String"
+    }
+  },
+  "variables": {
+    "namespaces_name_standard": "[concat(parameters('baseName'), '-Standard')]",
+    "location": "[resourceGroup().location]"
+  },
+  "resources": [
+    {
+      "type": "Microsoft.ServiceBus/namespaces",
+      "apiVersion": "2018-01-01-preview",
+      "name": "[variables('namespaces_name_standard')]",
+      "location": "[variables('location')]",
+      "sku": {
+        "name": "Standard",
+        "tier": "Standard"
+      },
+      "properties": {
+        "zoneRedundant": false
+      }
+    },
+    {
+      "type": "Microsoft.ServiceBus/namespaces/AuthorizationRules",
+      "apiVersion": "2017-04-01",
+      "name": "[concat(variables('namespaces_name_standard'), '/RootManageSharedAccessKey')]",
+      "location": "[variables('location')]",
+      "dependsOn": [
+        "[resourceId('Microsoft.ServiceBus/namespaces', variables('namespaces_name_standard'))]"
+      ],
+      "properties": {
+        "rights": [
+          "Listen",
+          "Manage",
+          "Send"
+        ]
+      }
+    },
+    {
+      "type": "Microsoft.ServiceBus/namespaces/queues",
+      "apiVersion": "2017-04-01",
+      "name": "[concat(variables('namespaces_name_standard'), '/queue1')]",
+      "location": "[variables('location')]",
+      "dependsOn": [
+        "[resourceId('Microsoft.ServiceBus/namespaces', variables('namespaces_name_standard'))]"
+      ],
+      "properties": {
+        "lockDuration": "PT30S",
+        "maxSizeInMegabytes": 1024,
+        "requiresDuplicateDetection": false,
+        "requiresSession": false,
+        "defaultMessageTimeToLive": "P14D",
+        "deadLetteringOnMessageExpiration": false,
+        "enableBatchedOperations": true,
+        "duplicateDetectionHistoryTimeWindow": "PT10M",
+        "maxDeliveryCount": 10,
+        "status": "Active",
+        "autoDeleteOnIdle": "P10675199DT2H48M5.4775807S",
+        "enablePartitioning": false,
+        "enableExpress": false
+      }
+    },
+    {
+      "type": "Microsoft.ServiceBus/namespaces/topics",
+      "apiVersion": "2017-04-01",
+      "name": "[concat(variables('namespaces_name_standard'), '/topic1')]",
+      "location": "[variables('location')]",
+      "dependsOn": [
+        "[resourceId('Microsoft.ServiceBus/namespaces', variables('namespaces_name_standard'))]"
+      ],
+      "properties": {
+        "defaultMessageTimeToLive": "P14D",
+        "maxSizeInMegabytes": 1024,
+        "requiresDuplicateDetection": false,
+        "duplicateDetectionHistoryTimeWindow": "PT10M",
+        "enableBatchedOperations": true,
+        "status": "Active",
+        "supportOrdering": true,
+        "autoDeleteOnIdle": "P10675199DT2H48M5.4775807S",
+        "enablePartitioning": false,
+        "enableExpress": false
+      }
+    },
+    {
+      "type": "Microsoft.ServiceBus/namespaces/topics/subscriptions",
+      "apiVersion": "2018-01-01-preview",
+      "name": "[concat(variables('namespaces_name_standard'), '/topic1/topicSub')]",
+      "location": "[variables('location')]",
+      "dependsOn": [
+        "[resourceId('Microsoft.ServiceBus/namespaces/topics', variables('namespaces_name_standard'), 'topic1')]",
+        "[resourceId('Microsoft.ServiceBus/namespaces', variables('namespaces_name_standard'))]"
+      ],
+      "properties": {
+        "lockDuration": "PT30S",
+        "requiresSession": false,
+        "defaultMessageTimeToLive": "P14D",
+        "deadLetteringOnMessageExpiration": false,
+        "deadLetteringOnFilterEvaluationExceptions": false,
+        "maxDeliveryCount": 1,
+        "status": "Active",
+        "enableBatchedOperations": true,
+        "autoDeleteOnIdle": "P14D"
+      }
+    }
+  ],
+  "outputs": {
+    "SERVICEBUS_BINDER_TEST_CONNECTION_STRING": {
+      "type": "string",
+      "value": "[listKeys(resourceId('Microsoft.ServiceBus/namespaces/authorizationRules', variables('namespaces_name_standard'), 'RootManageSharedAccessKey'), '2017-04-01').primaryConnectionString]"
+    }
+  }
+}

sdk/spring/pom.xml

@@ -213,6 +213,7 @@
         <module>azure-spring-cloud-stream-binder-test</module>
         <module>azure-spring-cloud-telemetry</module>
         <module>azure-spring-cloud-test-eventhubs</module>
+        <module>azure-spring-cloud-test-servicebus-binder</module>
         <module>azure-spring-integration-core</module>
         <module>azure-spring-integration-eventhubs</module>
         <module>azure-spring-integration-servicebus</module>

sdk/spring/spring-test-template.yml

@@ -8,6 +8,7 @@ parameters:
     - spring/azure-spring-boot-test-servicebus-jms
     - spring/azure-spring-boot-test-storage
     - spring/azure-spring-cloud-test-eventhubs
+    - spring/azure-spring-cloud-test-servicebus-binder
     - spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder
     - spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka
   Artifacts:
@@ -59,6 +60,9 @@ parameters:
     - name: azure-spring-cloud-sample-eventhubs-kafka
       groupId: com.azure.spring
       safeName: azurespringcloudsampleeventhubskafka
+    - name: azure-spring-cloud-test-servicebus-binder
+      groupId: com.azure.spring
+      safeName: azurespringcloudtestservicebusbinder
   EnvVars:
     AAD_TENANT_ID_1: $(java-spring-aad-tenant-id-1)
     AAD_USER_NAME_1: $(java-spring-aad-user-name-1)