diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml index 5d96421c0f2c..02db5bc32031 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml @@ -13,6 +13,11 @@ com.azure.spring 1.0.0 Azure Spring Cloud Sample Event Hubs Binder + + + true + + @@ -80,6 +85,18 @@ spring-boot-starter-test test + + org.junit.jupiter + junit-jupiter-engine + 5.7.1 + test + + + org.junit.vintage + junit-vintage-engine + 5.7.1 + test + @@ -114,6 +131,22 @@ + + org.apache.maven.plugins + maven-failsafe-plugin + 2.22.0 + + + default + + integration-test + + + ${skipSpringITs} + + + + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplicationIT.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplicationIT.java index eb64367c2033..99497841caa7 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplicationIT.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplicationIT.java @@ -10,7 +10,7 @@ import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.system.OutputCaptureRule; -import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.test.web.servlet.MockMvc; @@ -24,7 +24,7 @@ @RunWith(SpringRunner.class) @SpringBootTest(classes = EventHubBinderApplication.class) @AutoConfigureMockMvc -@TestPropertySource(locations = "classpath:application-test.properties") +@ActiveProfiles("manual") public class EventHubBinderApplicationIT { @Rule @@ -34,11 +34,10 @@ public class EventHubBinderApplicationIT { @Test public void testSendAndReceiveMessage() throws Exception { + Thread.sleep(10000); String message = UUID.randomUUID().toString(); - mvc.perform(post("/messages?message=" + message)).andExpect(status().isOk()) .andExpect(content().string(message)); - String messageReceivedLog = String.format("New message received: '%s'", message); String messageCheckpointedLog = String.format("Message '%s' successfully checkpointed", message); boolean messageReceived = false; diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application-test.properties b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application-test.properties deleted file mode 100644 index f99d6422e7b0..000000000000 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application-test.properties +++ /dev/null @@ -1,18 +0,0 @@ -spring.cloud.azure.credential-file-path=file:@credential@ -spring.cloud.azure.resource-group=spring-cloud -spring.cloud.azure.eventhub.namespace=spring-cloud-azure - -spring.cloud.azure.eventhub.checkpoint-storage-account=springcloudcheckpoint -spring.cloud.azure.eventhub.checkpoint-container=checkpointer - -spring.cloud.azure.region=westUS -spring.cloud.azure.auto-create-resources=true - -spring.cloud.stream.bindings.input.destination=eventhub1 -spring.cloud.stream.bindings.input.group=cg1 -spring.cloud.stream.bindings.output.destination=eventhub1 - -spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=MANUAL - -spring.main.banner-mode=off -spring.cloud.azure.msi-enabled=false diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application.yaml new file mode 100644 index 000000000000..49490b5e5a2a --- /dev/null +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/test/resources/application.yaml @@ -0,0 +1,27 @@ +spring: + main: + banner-mode: off + cloud: + azure: + eventhub: + connection-string: ${EVENTHUB_CONNECTION_STRING_BINDER} + checkpoint-storage-account: ${STORAGE_ACCOUNT_BINDER} + checkpoint-access-key: ${STORAGE_ACCOUNT_KEY_BINDER} + checkpoint-container: binder-test + stream: + function: + definition: consume;supply + bindings: + consume-in-0: + destination: binder-test + group: $Default + supply-out-0: + destination: binder-test + eventhub: + bindings: + consume-in-0: + consumer: + checkpoint-mode: MANUAL + default: + producer: + errorChannelEnabled: true diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/test-resources.json b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/test-resources.json new file mode 100644 index 000000000000..0927ad363c26 --- /dev/null +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/test-resources.json @@ -0,0 +1,192 @@ +{ + "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", + "contentVersion": "1.0.0.0", + "parameters": { + "baseName": { + "defaultValue": "[resourceGroup().name]", + "type": "String" + }, + "storageNamePrefix": { + "defaultValue": "[substring(uniqueString(parameters('baseName')), 0, 11)]", + "type": "string" + } + }, + "functions": [], + "variables": { + "eventHubsNamespaceName": "[concat(parameters('baseName'),'-binder-eventhub')]", + "eventHubsNamespaceKeyName": "RootManageSharedAccessKey", + "storageAccountName": "[concat(substring(toLower(concat(parameters('storageNamePrefix'), uniqueString(resourceGroup().id), 'binder')), 0, 18), 'binder')]", + "location": "[resourceGroup().location]" + }, + "resources": [ + { + "type": "Microsoft.EventHub/namespaces", + "apiVersion": "2018-01-01-preview", + "name": "[variables('eventHubsNamespaceName')]", + "location": "[variables('location')]", + "sku": { + "name": "Standard", + "tier": "Standard", + "capacity": 1 + }, + "properties": { + "zoneRedundant": false, + "isAutoInflateEnabled": false, + "maximumThroughputUnits": 0, + "kafkaEnabled": true + } + }, + { + "type": "Microsoft.EventHub/namespaces/AuthorizationRules", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/RootManageSharedAccessKey')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "rights": [ + "Listen", + "Manage", + "Send" + ] + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/binder-test')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/networkRuleSets", + "apiVersion": "2018-01-01-preview", + "name": "[concat(variables('eventHubsNamespaceName'), '/default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "defaultAction": "Deny", + "virtualNetworkRules": [], + "ipRules": [] + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/binder-test/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'binder-test')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + }, + { + "type": "Microsoft.Storage/storageAccounts", + "apiVersion": "2019-06-01", + "name": "[variables('storageAccountName')]", + "location": "[variables('location')]", + "sku": { + "name": "Standard_RAGRS", + "tier": "Standard" + }, + "kind": "StorageV2", + "properties": { + "networkAcls": { + "bypass": "AzureServices", + "virtualNetworkRules": [], + "ipRules": [], + "defaultAction": "Allow" + }, + "supportsHttpsTrafficOnly": true, + "encryption": { + "services": { + "file": { + "keyType": "Account", + "enabled": true + }, + "blob": { + "keyType": "Account", + "enabled": true + } + }, + "keySource": "Microsoft.Storage" + }, + "accessTier": "Hot" + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "sku": { + "name": "Standard_RAGRS", + "tier": "Standard" + }, + "properties": { + "cors": { + "corsRules": [] + }, + "deleteRetentionPolicy": { + "enabled": false + } + } + }, + { + "type": "Microsoft.Storage/storageAccounts/fileServices", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "sku": { + "name": "Standard_RAGRS", + "tier": "Standard" + }, + "properties": { + "cors": { + "corsRules": [] + } + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default/binder-test')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "properties": { + "publicAccess": "None" + } + } + ], + "outputs": { + "STORAGE_ACCOUNT_BINDER": { + "type": "string", + "value": "[variables('storageAccountName')]" + }, + "STORAGE_ACCOUNT_KEY_BINDER": { + "type": "string", + "value": "[listKeys(resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName')), '2019-06-01').keys[0].value]" + }, + "EVENTHUB_CONNECTION_STRING_BINDER": { + "type": "string", + "value": "[listKeys(resourceId('Microsoft.EventHub/namespaces/authorizationRules', variables('eventHubsNamespaceName'), variables('eventHubsNamespaceKeyName')), '2017-04-01').primaryConnectionString]" + } + } +} \ No newline at end of file diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml index 1baf95c141ca..192fa07c4b97 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml @@ -14,6 +14,10 @@ 1.0.0 Azure Spring Cloud Sample Event Hubs Kafka + + true + + @@ -76,5 +80,38 @@ spring-boot-starter-test test + + org.junit.jupiter + junit-jupiter-engine + 5.7.1 + test + + + org.junit.vintage + junit-vintage-engine + 5.7.1 + test + + + + + + org.apache.maven.plugins + maven-failsafe-plugin + 2.22.0 + + + default + + integration-test + + + ${skipSpringITs} + + + + + + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplicationIT.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplicationIT.java index c35438801f13..807df8b83199 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplicationIT.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplicationIT.java @@ -10,7 +10,6 @@ import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.boot.test.system.OutputCaptureRule; -import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.test.web.servlet.MockMvc; @@ -24,7 +23,6 @@ @RunWith(SpringRunner.class) @SpringBootTest(classes = EventHubKafkaBinderApplication.class) @AutoConfigureMockMvc -@TestPropertySource(locations = "classpath:application-test.properties") public class EventHubKafkaBinderApplicationIT { @Rule @@ -34,11 +32,10 @@ public class EventHubKafkaBinderApplicationIT { @Test public void testSendAndReceiveMessage() throws Exception { + Thread.sleep(10000); String message = UUID.randomUUID().toString(); - mvc.perform(post("/messages?message=" + message)).andExpect(status().isOk()) .andExpect(content().string(message)); - String messageReceivedLog = String.format("New message received: '%s'", message); boolean messageReceived = false; diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application-test.properties b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application-test.properties deleted file mode 100644 index 4d32ee772e1f..000000000000 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application-test.properties +++ /dev/null @@ -1,9 +0,0 @@ -spring.cloud.azure.credential-file-path=file:@credential@ -spring.cloud.azure.resource-group=spring-cloud -spring.cloud.azure.eventhub.namespace=spring-cloud-kafka - -spring.cloud.stream.bindings.input.destination=eventhub2 -spring.cloud.stream.bindings.input.group=cg1 -spring.cloud.stream.bindings.output.destination=eventhub2 - -spring.main.banner-mode=off diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application.yaml new file mode 100644 index 000000000000..55dfdaf9d9a4 --- /dev/null +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/test/resources/application.yaml @@ -0,0 +1,21 @@ +spring: + main: + banner-mode: off + cloud: + azure: + client-id: ${SPRING_CLIENT_ID_KAFKA} + client-secret: ${SPRING_CLIENT_SECRET_KAFKA} + tenant-id: ${SPRING_TENANT_ID_KAFKA} + resource-group: ${SPRING_RESOURCE_GROUP_KAFKA} + subscription-id: ${SPRING_SUBSCRIPTION_ID_KAFKA} + eventhub: + namespace: ${EVENTHUB_NAMESPACE_NAME_KAFKA} + stream: + function: + definition: consume;supply + bindings: + consume-in-0: + destination: kafka-test + group: $Default + supply-out-0: + destination: kafka-test diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/test-resources.json b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/test-resources.json new file mode 100644 index 000000000000..473df14b7e09 --- /dev/null +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/test-resources.json @@ -0,0 +1,109 @@ +{ + "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#", + "contentVersion": "1.0.0.0", + "parameters": { + "baseName": { + "defaultValue": "[resourceGroup().name]", + "type": "String" + }, + "tenantId": { + "type": "String" + }, + "testApplicationId": { + "type": "String" + }, + "testApplicationSecret": { + "type": "String" + } + }, + "functions": [], + "variables": { + "eventHubsNamespaceName": "[concat(parameters('baseName'),'-kafka-eventhub')]", + "eventHubsNamespaceKeyName": "RootManageSharedAccessKey", + "location": "[resourceGroup().location]" + }, + "resources": [ + { + "type": "Microsoft.EventHub/namespaces", + "apiVersion": "2018-01-01-preview", + "name": "[variables('eventHubsNamespaceName')]", + "location": "[variables('location')]", + "sku": { + "name": "Standard", + "tier": "Standard", + "capacity": 1 + }, + "properties": { + "zoneRedundant": false, + "isAutoInflateEnabled": false, + "maximumThroughputUnits": 0, + "kafkaEnabled": true + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/kafka-test')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/networkRuleSets", + "apiVersion": "2018-01-01-preview", + "name": "[concat(variables('eventHubsNamespaceName'), '/default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "defaultAction": "Deny", + "virtualNetworkRules": [], + "ipRules": [] + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/kafka-test/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'kafka-test')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + } + ], + "outputs": { + "EVENTHUB_NAMESPACE_NAME_KAFKA": { + "type": "string", + "value": "[variables('eventHubsNamespaceName')]" + }, + "SPRING_TENANT_ID_KAFKA": { + "type": "string", + "value": "[parameters('tenantId')]" + }, + "SPRING_CLIENT_ID_KAFKA": { + "type": "string", + "value": "[parameters('testApplicationId')]" + }, + "SPRING_CLIENT_SECRET_KAFKA": { + "type": "string", + "value": "[parameters('testApplicationSecret')]" + }, + "SPRING_RESOURCE_GROUP_KAFKA": { + "type": "string", + "value": "[resourceGroup().name]" + }, + "SPRING_SUBSCRIPTION_ID_KAFKA": { + "type": "string", + "value": "[subscription().subscriptionId]" + } + } +} \ No newline at end of file diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderBatchModeIT.java b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderBatchModeIT.java deleted file mode 100644 index a3a540afe833..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderBatchModeIT.java +++ /dev/null @@ -1,49 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.spring.eventhub.stream.binder.integration; - -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.junit4.SpringRunner; - -import java.util.UUID; - -import static org.assertj.core.api.Assertions.assertThat; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = EventHubBinderBatchModeIT.TestConfig.class) -@TestPropertySource(locations = "classpath:application-test.properties", - properties = "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=BATCH") -public class EventHubBinderBatchModeIT { - - private static String message = UUID.randomUUID().toString(); - - @Autowired - Source source; - - @Test - public void testSendAndReceiveMessage() { - this.source.output().send(new GenericMessage<>(message)); - } - - @EnableBinding({Source.class, Sink.class}) - @EnableAutoConfiguration - public static class TestConfig { - - @StreamListener(Sink.INPUT) - public void handleMessage(String message) { - assertThat(message.equals(EventHubBinderBatchModeIT.message)).isTrue(); - } - - } -} diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderManualModeIT.java b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderManualModeIT.java deleted file mode 100644 index 4cab24ffcdeb..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderManualModeIT.java +++ /dev/null @@ -1,55 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.spring.eventhub.stream.binder.integration; - -import com.azure.spring.integration.core.AzureHeaders; -import com.azure.spring.integration.core.api.Checkpointer; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.handler.annotation.Header; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.junit4.SpringRunner; - -import java.util.UUID; - -import static org.assertj.core.api.Assertions.assertThat; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = EventHubBinderManualModeIT.TestConfig.class) -@TestPropertySource(locations = "classpath:application-test.properties", - properties = "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=MANUAL") -public class EventHubBinderManualModeIT { - - private static String message = UUID.randomUUID().toString(); - @Autowired - Source source; - - @Test - public void testSendAndReceiveMessage() { - this.source.output().send(new GenericMessage<>(message)); - } - - @EnableBinding({Source.class, Sink.class}) - @EnableAutoConfiguration - public static class TestConfig { - - @StreamListener(Sink.INPUT) - public void handleMessage(String message, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) { - assertThat(message.equals(EventHubBinderManualModeIT.message)).isTrue(); - checkpointer.success().handle((r, ex) -> { - assertThat(ex == null).isTrue(); - return null; - }); - } - - } -} diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderRecordModeIT.java b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderRecordModeIT.java deleted file mode 100644 index c61acd7999ed..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderRecordModeIT.java +++ /dev/null @@ -1,48 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.spring.eventhub.stream.binder.integration; - -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.junit4.SpringRunner; - -import java.util.UUID; - -import static org.assertj.core.api.Assertions.assertThat; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = EventHubBinderRecordModeIT.TestConfig.class) -@TestPropertySource(locations = "classpath:application-test.properties", - properties = "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=RECORD") -public class EventHubBinderRecordModeIT { - - private static String message = UUID.randomUUID().toString(); - @Autowired - Source source; - - @Test - public void testSendAndReceiveMessage() { - this.source.output().send(new GenericMessage<>(message)); - } - - @EnableBinding({Source.class, Sink.class}) - @EnableAutoConfiguration - public static class TestConfig { - - @StreamListener(Sink.INPUT) - public void handleMessage(String message) { - assertThat(message.equals(EventHubBinderRecordModeIT.message)).isTrue(); - } - - } -} diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderSyncModeIT.java b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderSyncModeIT.java deleted file mode 100644 index 1ce2935ed7f0..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/java/com/azure/spring/eventhub/stream/binder/integration/EventHubBinderSyncModeIT.java +++ /dev/null @@ -1,48 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.spring.eventhub.stream.binder.integration; - -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.junit4.SpringRunner; - -import java.util.UUID; - -import static org.assertj.core.api.Assertions.assertThat; - -@RunWith(SpringRunner.class) -@SpringBootTest(classes = EventHubBinderSyncModeIT.TestConfig.class) -@TestPropertySource(locations = "classpath:application-test.properties", - properties = "spring.cloud.stream.eventhub.bindings.input.producer.sync=true") -public class EventHubBinderSyncModeIT { - - private static String message = UUID.randomUUID().toString(); - @Autowired - Source source; - - @Test - public void testSendAndReceiveMessage() { - this.source.output().send(new GenericMessage<>(message)); - } - - @EnableBinding({Source.class, Sink.class}) - @EnableAutoConfiguration - public static class TestConfig { - - @StreamListener(Sink.INPUT) - public void handleMessage(String message) { - assertThat(message.equals(EventHubBinderSyncModeIT.message)).isTrue(); - } - - } -} diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/application-test.properties b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/application-test.properties deleted file mode 100644 index b3fd93bb5227..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/application-test.properties +++ /dev/null @@ -1,14 +0,0 @@ -spring.cloud.azure.credential-file-path=file:@credential@ -spring.cloud.azure.resource-group=spring-cloud -spring.cloud.azure.eventhub.namespace=spring-cloud-azure - -spring.cloud.azure.eventhub.checkpoint-storage-account=springcloudcheckpoint - -spring.cloud.azure.region=westUS -spring.cloud.azure.auto-create-resources=true - -spring.cloud.stream.bindings.input.destination=eventhub1 -spring.cloud.stream.bindings.input.group=cg1 -spring.cloud.stream.bindings.output.destination=eventhub1 - -spring.main.banner-mode=off diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/logback-test.xml b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/logback-test.xml deleted file mode 100644 index 300f34b41c27..000000000000 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/src/test/resources/logback-test.xml +++ /dev/null @@ -1,12 +0,0 @@ - - - - - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n - - - - - - - diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/pom.xml b/sdk/spring/azure-spring-cloud-test-eventhubs/pom.xml index 050b627e1a37..06a354d9eefd 100644 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/pom.xml +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/pom.xml @@ -34,15 +34,9 @@ 2.8.6 - org.springframework.boot - spring-boot-starter-test - test - - - org.junit.vintage - junit-vintage-engine - - + com.azure.spring + azure-spring-cloud-stream-binder-test + 2.4.0-beta.1 diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/EventHubBinderConsumeErrorIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/EventHubBinderConsumeErrorIT.java new file mode 100644 index 000000000000..b0d249d3e18a --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/EventHubBinderConsumeErrorIT.java @@ -0,0 +1,93 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.spring.test.eventhubs; + +import com.azure.spring.integration.core.AzureHeaders; +import com.azure.spring.integration.core.api.reactor.Checkpointer; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +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.integration.annotation.ServiceActivator; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.TestPropertySource; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; + +import java.util.concurrent.Exchanger; +import java.util.function.Consumer; +import java.util.function.Supplier; + +@SpringBootTest(classes = EventHubBinderConsumeErrorIT.TestConfig.class) +@TestPropertySource(properties = + { + "spring.cloud.stream.bindings.consume-in-0.destination=test-eventhub", + "spring.cloud.stream.bindings.supply-out-0.destination=test-eventhub", + "spring.cloud.azure.eventhub.checkpoint-container=test-eventhub" + }) +public class EventHubBinderConsumeErrorIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderConsumeErrorIT.class); + private static final String MESSAGE = "Azure Spring Cloud EventHub Test"; + + @Autowired + private Sinks.One> one; + + @EnableAutoConfiguration + public static class TestConfig { + + public static final Exchanger EXCHANGER = new Exchanger<>(); + + @Bean + public Sinks.One> one() { + return Sinks.one(); + } + + @Bean + public Supplier>> supply(Sinks.One> one) { + return () -> one.asMono() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> { + LOGGER.info("New message received: '{}'", message.getPayload()); + Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(AzureHeaders.CHECKPOINTER); + checkpointer.success(); + try { + String exchange = EXCHANGER.exchange(message.getPayload()); + LOGGER.info("Consume exchange: {}", exchange); + } catch (InterruptedException e) { + e.printStackTrace(); + } + throw new RuntimeException(); + }; + } + + @ServiceActivator(inputChannel = "test-eventhub.$Default.errors") + public void consumeError(Message message) throws InterruptedException { + EXCHANGER.exchange("ERROR!"); + } + } + + @Test + @Timeout(70) + void integrationTest() throws InterruptedException { + // Wait for eventhub initialization to complete + Thread.sleep(15000); + one.emitValue(new GenericMessage<>(MESSAGE), Sinks.EmitFailureHandler.FAIL_FAST); + String msg = TestConfig.EXCHANGER.exchange(MESSAGE); + Assertions.assertEquals(MESSAGE, msg); + msg = TestConfig.EXCHANGER.exchange(""); + Assertions.assertEquals("ERROR!", msg); + } +} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderBatchModeIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderBatchModeIT.java new file mode 100644 index 000000000000..8b708eff5ded --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderBatchModeIT.java @@ -0,0 +1,75 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.spring.test.eventhubs.stream.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 = EventHubBinderBatchModeIT.TestConfig.class) +@TestPropertySource(properties = + { + "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=BATCH", + "spring.cloud.stream.bindings.consume-in-0.destination=test-eventhub-batch", + "spring.cloud.stream.bindings.supply-out-0.destination=test-eventhub-batch", + "spring.cloud.azure.eventhub.checkpoint-container=test-eventhub-batch" + }) +public class EventHubBinderBatchModeIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderBatchModeIT.class); + + private static String message = UUID.randomUUID().toString(); + private static final AtomicInteger count = new AtomicInteger(0); + + @Autowired + private Sinks.Many> many; + + @EnableAutoConfiguration + public static class TestConfig { + + @Bean + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); + } + + @Bean + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m.getPayload())) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> { + LOGGER.info("New message received: '{}'", message.getPayload()); + Assertions.assertEquals(message.getPayload(), EventHubBinderBatchModeIT.message); + count.addAndGet(1); + }; + } + } + + @Test + public void testSendAndReceiveMessage() throws InterruptedException { + Thread.sleep(15000); + many.emitNext(new GenericMessage<>(message), Sinks.EmitFailureHandler.FAIL_FAST); + Thread.sleep(6000); + Assertions.assertEquals(1, count.get()); + } +} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderManualModeIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderManualModeIT.java new file mode 100644 index 000000000000..8d7e3eb48819 --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderManualModeIT.java @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.spring.test.eventhubs.stream.binder; + +import com.azure.spring.integration.core.AzureHeaders; +import com.azure.spring.integration.core.api.reactor.Checkpointer; +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 = EventHubBinderManualModeIT.TestConfig.class) +@TestPropertySource(properties = + { + "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=MANUAL", + "spring.cloud.stream.bindings.consume-in-0.destination=test-eventhub-manual", + "spring.cloud.stream.bindings.supply-out-0.destination=test-eventhub-manual", + "spring.cloud.azure.eventhub.checkpoint-container=test-eventhub-manual" + }) +public class EventHubBinderManualModeIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderManualModeIT.class); + private static String message = UUID.randomUUID().toString(); + private static final AtomicInteger count = new AtomicInteger(0); + + @Autowired + private Sinks.Many> many; + + @EnableAutoConfiguration + public static class TestConfig { + + @Bean + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); + } + + @Bean + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> { + LOGGER.info("New message received: '{}'", message.getPayload()); + Assertions.assertEquals(message.getPayload(), EventHubBinderManualModeIT.message); + Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(AzureHeaders.CHECKPOINTER); + checkpointer.success().handle((r, ex) -> { + Assertions.assertNull(ex); + }); + count.addAndGet(1); + }; + } + } + + @Test + public void testSendAndReceiveMessage() throws InterruptedException { + Thread.sleep(15000); + many.emitNext(new GenericMessage<>(message), Sinks.EmitFailureHandler.FAIL_FAST); + Thread.sleep(6000); + Assertions.assertEquals(1, count.get()); + } +} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderRecordModeIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderRecordModeIT.java new file mode 100644 index 000000000000..75f5df4fa593 --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderRecordModeIT.java @@ -0,0 +1,74 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.spring.test.eventhubs.stream.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 = EventHubBinderRecordModeIT.TestConfig.class) +@TestPropertySource(properties = + { + "spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=RECORD", + "spring.cloud.stream.bindings.consume-in-0.destination=test-eventhub-record", + "spring.cloud.stream.bindings.supply-out-0.destination=test-eventhub-record", + "spring.cloud.azure.eventhub.checkpoint-container=test-eventhub-record" + }) +public class EventHubBinderRecordModeIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderManualModeIT.class); + private static String message = UUID.randomUUID().toString(); + private static final AtomicInteger count = new AtomicInteger(0); + + @Autowired + private Sinks.Many> many; + + @EnableAutoConfiguration + public static class TestConfig { + + @Bean + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); + } + + @Bean + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m.getPayload())) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> { + LOGGER.info("New message received: '{}'", message.getPayload()); + Assertions.assertEquals(message.getPayload(), EventHubBinderRecordModeIT.message); + count.addAndGet(1); + }; + } + } + + @Test + public void testSendAndReceiveMessage() throws InterruptedException { + Thread.sleep(15000); + many.emitNext(new GenericMessage<>(message), Sinks.EmitFailureHandler.FAIL_FAST); + Thread.sleep(6000); + Assertions.assertEquals(1, count.get()); + } +} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderSyncModeIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderSyncModeIT.java new file mode 100644 index 000000000000..c76b91822bcc --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/azure/spring/test/eventhubs/stream/binder/EventHubBinderSyncModeIT.java @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.spring.test.eventhubs.stream.binder; + +import org.junit.Rule; +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.boot.test.system.OutputCaptureRule; +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 = EventHubBinderSyncModeIT.TestConfig.class) +@TestPropertySource(properties = + { + "spring.cloud.stream.eventhub.bindings.input.producer.sync=true", + "spring.cloud.stream.bindings.consume-in-0.destination=test-eventhub-sync", + "spring.cloud.stream.bindings.supply-out-0.destination=test-eventhub-sync", + "spring.cloud.azure.eventhub.checkpoint-container=test-eventhub-sync" + }) +public class EventHubBinderSyncModeIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderManualModeIT.class); + private static String message = UUID.randomUUID().toString(); + + @Autowired + private Sinks.Many> many; + + @Rule + public OutputCaptureRule capture = new OutputCaptureRule(); + + private static final AtomicInteger count = new AtomicInteger(0); + + @EnableAutoConfiguration + public static class TestConfig { + + @Bean + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); + } + + @Bean + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m.getPayload())) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> { + LOGGER.info("New message received: '{}'", message.getPayload()); + Assertions.assertEquals(message.getPayload(), EventHubBinderSyncModeIT.message); + count.addAndGet(1); + }; + } + } + + @Test + public void testSendAndReceiveMessage() throws InterruptedException { + Thread.sleep(15000); + many.emitNext(new GenericMessage<>(message), Sinks.EmitFailureHandler.FAIL_FAST); + Thread.sleep(6000); + Assertions.assertEquals(1, count.get()); + } +} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Application.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Application.java deleted file mode 100644 index 3e0699628252..000000000000 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Application.java +++ /dev/null @@ -1,10 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.microsoft.azure.eventhubs; - -import org.springframework.boot.autoconfigure.SpringBootApplication; - -@SpringBootApplication -public class Application { -} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/EventHubIT.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/EventHubIT.java deleted file mode 100644 index 33b737f1a081..000000000000 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/EventHubIT.java +++ /dev/null @@ -1,38 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.microsoft.azure.eventhubs; - -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.messaging.support.GenericMessage; - -@EnableBinding(Source.class) -@SpringBootTest -class EventHubIT { - - private static final String MESSAGE = "Azure Spring Cloud EventHub Test"; - - private final Source source; - - @Autowired - EventHubIT(Source source) { - this.source = source; - } - - @Test - void integrationTest() throws InterruptedException { - // Wait for eventhub initialization to complete - Thread.sleep(15000); - this.source.output().send(new GenericMessage<>(MESSAGE)); - String msg = Receiver.EXCHANGER.exchange(MESSAGE); - Assertions.assertEquals(MESSAGE, msg); - msg = Receiver.EXCHANGER.exchange(""); - Assertions.assertEquals("ERROR!", msg); - } - -} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Receiver.java b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Receiver.java deleted file mode 100644 index 2c1b7e3dc017..000000000000 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/java/com/microsoft/azure/eventhubs/Receiver.java +++ /dev/null @@ -1,34 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.microsoft.azure.eventhubs; - -import com.azure.spring.integration.core.AzureHeaders; -import com.azure.spring.integration.core.api.reactor.Checkpointer; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.messaging.Message; -import org.springframework.messaging.handler.annotation.Header; - -import java.util.concurrent.Exchanger; - -@EnableBinding(Sink.class) -public class Receiver { - - public static final Exchanger EXCHANGER = new Exchanger<>(); - - @StreamListener(Sink.INPUT) - public void handleMessage(String message, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) throws InterruptedException { - checkpointer.success(); - EXCHANGER.exchange(message); - throw new RuntimeException(); - } - - @ServiceActivator(inputChannel = "test.$Default.errors") - public void consumerError(Message message) throws InterruptedException { - EXCHANGER.exchange("ERROR!"); - } - -} diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.properties b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.properties deleted file mode 100644 index bfe38b213817..000000000000 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.properties +++ /dev/null @@ -1,8 +0,0 @@ -spring.cloud.azure.eventhub.checkpoint-storage-account=${STORAGE_ACCOUNT} -spring.cloud.azure.eventhub.connection-string=${EVENTHUB_CONNECTION_STRING} -spring.cloud.azure.eventhub.checkpoint-access-key=${STORAGE_ACCOUNT_KEY} -spring.cloud.stream.bindings.input.destination=test -spring.cloud.stream.bindings.input.group=$Default -spring.cloud.stream.bindings.output.destination=test -spring.cloud.stream.default.producer.errorChannelEnabled=true -spring.cloud.stream.eventhub.bindings.input.consumer.checkpoint-mode=MANUAL diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.yaml b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.yaml new file mode 100644 index 000000000000..ce258d2e48b9 --- /dev/null +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/src/test/resources/application.yaml @@ -0,0 +1,21 @@ +spring: + cloud: + azure: + eventhub: + connection-string: ${EVENTHUB_CONNECTION_STRING_TEST_EVENTHUBS} + checkpoint-storage-account: ${STORAGE_ACCOUNT_TEST_EVENTHUBS} + checkpoint-access-key: ${STORAGE_ACCOUNT_KEY_TEST_EVENTHUBS} + stream: + function: + definition: consume;supply + bindings: + consume-in-0: + group: $Default + eventhub: + bindings: + consume-in-0: + consumer: + checkpoint-mode: MANUAL + default: + producer: + errorChannelEnabled: true diff --git a/sdk/spring/azure-spring-cloud-test-eventhubs/test-resources.json b/sdk/spring/azure-spring-cloud-test-eventhubs/test-resources.json index 01d7e9fd4874..41445e5cc01f 100644 --- a/sdk/spring/azure-spring-cloud-test-eventhubs/test-resources.json +++ b/sdk/spring/azure-spring-cloud-test-eventhubs/test-resources.json @@ -13,9 +13,9 @@ }, "functions": [], "variables": { - "eventHubsNamespaceName": "[concat(parameters('baseName'),'-eventhub')]", + "eventHubsNamespaceName": "[concat(parameters('baseName'),'-test-eventhub')]", "eventHubsNamespaceKeyName": "RootManageSharedAccessKey", - "storageAccountName": "[substring(toLower(concat(parameters('storageNamePrefix'), uniqueString(resourceGroup().id))), 0, 24)]", + "storageAccountName": "[concat(substring(toLower(concat(parameters('storageNamePrefix'), uniqueString(resourceGroup().id))), 0, 20), 'test')]", "location": "[resourceGroup().location]" }, "resources": [ @@ -55,7 +55,7 @@ { "type": "Microsoft.EventHub/namespaces/eventhubs", "apiVersion": "2017-04-01", - "name": "[concat(variables('eventHubsNamespaceName'), '/test')]", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub')]", "location": "[variables('location')]", "dependsOn": [ "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" @@ -80,13 +80,113 @@ "ipRules": [] } }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-batch')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-manual')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-record')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-sync')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": { + "messageRetentionInDays": 1, + "partitionCount": 1, + "status": "Active" + } + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test-eventhub')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-batch/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test-eventhub')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + }, { "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", "apiVersion": "2017-04-01", - "name": "[concat(variables('eventHubsNamespaceName'), '/test/$Default')]", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-manual/$Default')]", "location": "[variables('location')]", "dependsOn": [ - "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test')]", + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test-eventhub')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-record/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test-eventhub')]", + "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" + ], + "properties": {} + }, + { + "type": "Microsoft.EventHub/namespaces/eventhubs/consumergroups", + "apiVersion": "2017-04-01", + "name": "[concat(variables('eventHubsNamespaceName'), '/test-eventhub-sync/$Default')]", + "location": "[variables('location')]", + "dependsOn": [ + "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubsNamespaceName'), 'test-eventhub')]", "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubsNamespaceName'))]" ], "properties": {} @@ -165,7 +265,55 @@ { "type": "Microsoft.Storage/storageAccounts/blobServices/containers", "apiVersion": "2019-06-01", - "name": "[concat(variables('storageAccountName'), '/default/test')]", + "name": "[concat(variables('storageAccountName'), '/default/test-eventhub')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "properties": { + "publicAccess": "None" + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default/test-eventhub-batch')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "properties": { + "publicAccess": "None" + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default/test-eventhub-manual')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "properties": { + "publicAccess": "None" + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default/test-eventhub-record')]", + "dependsOn": [ + "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", + "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" + ], + "properties": { + "publicAccess": "None" + } + }, + { + "type": "Microsoft.Storage/storageAccounts/blobServices/containers", + "apiVersion": "2019-06-01", + "name": "[concat(variables('storageAccountName'), '/default/test-eventhub-sync')]", "dependsOn": [ "[resourceId('Microsoft.Storage/storageAccounts/blobServices', variables('storageAccountName'), 'default')]", "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]" @@ -176,15 +324,15 @@ } ], "outputs": { - "STORAGE_ACCOUNT": { + "STORAGE_ACCOUNT_TEST_EVENTHUBS": { "type": "string", "value": "[variables('storageAccountName')]" }, - "STORAGE_ACCOUNT_KEY": { + "STORAGE_ACCOUNT_KEY_TEST_EVENTHUBS": { "type": "string", "value": "[listKeys(resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName')), '2019-06-01').keys[0].value]" }, - "EVENTHUB_CONNECTION_STRING": { + "EVENTHUB_CONNECTION_STRING_TEST_EVENTHUBS": { "type": "string", "value": "[listKeys(resourceId('Microsoft.EventHub/namespaces/authorizationRules', variables('eventHubsNamespaceName'), variables('eventHubsNamespaceKeyName')), '2017-04-01').primaryConnectionString]" } diff --git a/sdk/spring/spring-test-template.yml b/sdk/spring/spring-test-template.yml index 76261158dfdd..a0465525cc98 100644 --- a/sdk/spring/spring-test-template.yml +++ b/sdk/spring/spring-test-template.yml @@ -8,6 +8,8 @@ parameters: - spring/azure-spring-boot-test-servicebus-jms - spring/azure-spring-boot-test-storage - spring/azure-spring-cloud-test-eventhubs + - spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder + - spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka Artifacts: - name: azure-spring-boot-test-application groupId: com.azure.spring @@ -51,6 +53,12 @@ parameters: - name: azure-spring-boot-test-storage groupId: com.azure.spring safeName: azurespringbootteststorage + - name: azure-spring-cloud-sample-eventhubs-binder + groupId: com.azure.spring + safeName: azurespringcloudsampleeventhubsbinder + - name: azure-spring-cloud-sample-eventhubs-kafka + groupId: com.azure.spring + safeName: azurespringcloudsampleeventhubskafka EnvVars: AAD_TENANT_ID_1: $(java-spring-aad-tenant-id-1) AAD_USER_NAME_1: $(java-spring-aad-user-name-1)