Skip to content
Permalink
Browse files
Azure Storage Queue : add consumer integration tests
Fixes #3718
  • Loading branch information
zbendhiba committed Apr 15, 2022
1 parent 6888709 commit 7b1af0ce69916892a500347dfe936592d7ad70ae
Showing 5 changed files with 123 additions and 1 deletion.
@@ -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)
@@ -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>
@@ -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,
@@ -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");
}
}
@@ -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")

0 comments on commit 7b1af0c

Please sign in to comment.