Azure Storage Queue : add consumer integration tests

integration-test-groups/azure/azure-eventhubs/src/main/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsResource.java

@@ -74,6 +74,7 @@ void schedule() {
     @GET
     @Produces(MediaType.APPLICATION_JSON)
     public List<String> receiveEvents() throws Exception {
+
         final MockEndpoint mockEndpoint = context.getEndpoint("mock:azure-consumed", MockEndpoint.class);
         return mockEndpoint.getReceivedExchanges().stream()
                 .map(Exchange::getMessage)

integration-test-groups/azure/azure-storage-queue/pom.xml

@@ -39,6 +39,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy-jackson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-mock</artifactId>
+        </dependency>
 
         <!-- test dependencies -->
         <dependency>
@@ -51,6 +55,11 @@
             <artifactId>rest-assured</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel.quarkus</groupId>
             <artifactId>camel-quarkus-integration-tests-support-azure</artifactId>
@@ -108,6 +117,19 @@
                         </exclusion>
                     </exclusions>
                 </dependency>
+                <dependency>
+                    <groupId>org.apache.camel.quarkus</groupId>
+                    <artifactId>camel-quarkus-mock-deployment</artifactId>
+                    <version>${project.version}</version>
+                    <type>pom</type>
+                    <scope>test</scope>
+                    <exclusions>
+                        <exclusion>
+                            <groupId>*</groupId>
+                            <artifactId>*</artifactId>
+                        </exclusion>
+                    </exclusions>
+                </dependency>
             </dependencies>
         </profile>
         <profile>

integration-test-groups/azure/azure-storage-queue/src/main/java/org/apache/camel/quarkus/component/azure/storage/queue/it/AzureStorageQueueResource.java

@@ -45,21 +45,28 @@
 import com.azure.storage.queue.models.QueueItem;
 import com.azure.storage.queue.models.QueueMessageItem;
 import com.azure.storage.queue.models.QueuesSegmentOptions;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
 import org.apache.camel.ProducerTemplate;
 import org.apache.camel.component.azure.storage.queue.QueueConstants;
 import org.apache.camel.component.azure.storage.queue.QueueOperationDefinition;
+import org.apache.camel.component.mock.MockEndpoint;
 import org.apache.camel.quarkus.component.azure.storage.queue.it.model.ExampleMessage;
+import org.apache.camel.spi.RouteController;
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 @Path("/azure-storage-queue")
 @ApplicationScoped
 public class AzureStorageQueueResource {
 
-    private static final String QUEUE_NAME = "camel-quarkus-" + UUID.randomUUID().toString();
+    protected static final String QUEUE_NAME = "camel-quarkus-" + UUID.randomUUID().toString();
 
     @Inject
     ProducerTemplate producerTemplate;
 
+    @Inject
+    CamelContext context;
+
     @ConfigProperty(name = "azure.storage.account-name")
     String azureStorageAccountName;
 
@@ -190,6 +197,31 @@ public Response clearQueue() throws Exception {
         return Response.noContent().build();
     }
 
+    @Path("/queue/consumer/{action}")
+    @POST
+    public Response modifyConsumerRouteState(@PathParam("action") String action) throws Exception {
+        RouteController controller = context.getRouteController();
+        if (action.equals("start")) {
+            controller.startRoute("queueRoute");
+        } else if (action.equals("stop")) {
+            controller.stopRoute("queueRoute");
+        } else {
+            throw new IllegalArgumentException("Unknown action: " + action);
+        }
+        return Response.noContent().build();
+    }
+
+    @Path("/queue/consumer")
+    @GET
+    @Produces(MediaType.APPLICATION_JSON)
+    public String receiveMessages() throws Exception {
+        final MockEndpoint mockEndpoint = context.getEndpoint("mock:result", MockEndpoint.class);
+        return mockEndpoint.getReceivedExchanges().stream()
+                .map(Exchange::getMessage)
+                .map(m -> m.getBody(String.class))
+                .collect(Collectors.joining("\n"));
+    }
+
     private String componentUri(final QueueOperationDefinition operation) {
         return String.format("azure-storage-queue://%s/%s?operation=%s",
                 azureStorageAccountName, QUEUE_NAME,

integration-test-groups/azure/azure-storage-queue/src/main/java/org/apache/camel/quarkus/component/azure/storage/queue/it/Routes.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.apache.camel.quarkus.component.azure.storage.queue.it;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+
+import static org.apache.camel.quarkus.component.azure.storage.queue.it.AzureStorageQueueResource.QUEUE_NAME;
+
+@ApplicationScoped
+public class Routes extends RouteBuilder {
+
+    @ConfigProperty(name = "azure.storage.account-name")
+    String azureStorageAccountName;
+
+    @Override
+    public void configure() throws Exception {
+        // setting the consumer startup to False in order to differenciate consuming messages by consumer or producer
+        from(String.format("azure-storage-queue://%s/%s?maxMessages=5", azureStorageAccountName, QUEUE_NAME))
+                .routeId("queueRoute").autoStartup(false)
+                .to("mock:result");
+    }
+}

integration-test-groups/azure/azure-storage-queue/src/test/java/org/apache/camel/quarkus/component/azure/storage/queue/it/AzureStorageQueueTest.java

@@ -18,13 +18,15 @@
 
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import io.quarkus.test.common.QuarkusTestResource;
 import io.quarkus.test.junit.QuarkusTest;
 import io.restassured.RestAssured;
 import io.restassured.http.ContentType;
 import org.apache.camel.quarkus.test.support.azure.AzureStorageTestResource;
 import org.junit.jupiter.api.Test;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
 
 import static io.restassured.RestAssured.given;
 import static org.hamcrest.Matchers.containsString;
@@ -123,6 +125,32 @@ public void crud() {
                     .then()
                     .statusCode(204);
 
+            // consumer test
+
+            // start consumer Route
+            given()
+                    .post("/azure-storage-queue/queue/consumer/start")
+                    .then()
+                    .statusCode(204);
+
+            // add message
+            addMessage("Testing consumer");
+
+            // verify message is consumed
+            Awaitility.await().pollInterval(1, TimeUnit.SECONDS).atMost(10, TimeUnit.SECONDS).until(() -> {
+                final String body = RestAssured.given()
+                        .get("/azure-storage-queue/queue/consumer")
+                        .then()
+                        .extract().body().asString();
+                return body != null && body.contains("Testing consumer");
+            });
+
+            // stop consumer Route
+            given()
+                    .post("/azure-storage-queue/queue/consumer/stop")
+                    .then()
+                    .statusCode(204);
+
         } finally {
             // Delete
             RestAssured.delete("/azure-storage-queue/queue/delete")