You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "jamesnetherton (via GitHub)" <gi...@apache.org> on 2024/08/22 06:16:14 UTC
[PR] Increase azure-eventhubs test coverage [camel-quarkus]
jamesnetherton opened a new pull request, #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "aldettinger (via GitHub)" <gi...@apache.org>.
aldettinger commented on code in PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#discussion_r1726469052
##########
integration-test-groups/azure/azure-eventhubs/src/test/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsTest.java:
##########
@@ -16,45 +16,422 @@
*/
package org.apache.camel.quarkus.component.azure.eventhubs.it;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
import java.util.concurrent.TimeUnit;
import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
-import org.apache.commons.lang3.RandomStringUtils;
+import io.restassured.response.Response;
import org.awaitility.Awaitility;
-import org.jboss.logging.Logger;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
@EnabledIfEnvironmentVariable(named = "AZURE_STORAGE_ACCOUNT_NAME", matches = ".+")
@EnabledIfEnvironmentVariable(named = "AZURE_STORAGE_ACCOUNT_KEY", matches = ".+")
@EnabledIfEnvironmentVariable(named = "AZURE_EVENT_HUBS_BLOB_CONTAINER_NAME", matches = ".+")
@EnabledIfEnvironmentVariable(named = "AZURE_EVENT_HUBS_CONNECTION_STRING", matches = ".+")
@QuarkusTest
class AzureEventhubsTest {
+ // NOTE: Consumer endpoints are started / stopped manually to prevent them from inferring with each other
- private static final Logger LOG = Logger.getLogger(AzureEventhubsTest.class);
+ @AfterEach
+ public void afterEach() {
+ RestAssured.post("/azure-eventhubs/reset/mocks")
Review Comment:
Same way, don't we have reasonable way of avoiding this ? Like an option to restart quarkus context on every test ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "aldettinger (via GitHub)" <gi...@apache.org>.
aldettinger commented on code in PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#discussion_r1726467488
##########
integration-test-groups/azure/azure-eventhubs/src/main/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsResource.java:
##########
@@ -17,77 +17,142 @@
package org.apache.camel.quarkus.component.azure.eventhubs.it;
import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
-import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
-import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.mock.MockEndpoint;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.util.ObjectHelper;
+import org.jboss.logging.Logger;
@Path("/azure-eventhubs")
@ApplicationScoped
public class AzureEventhubsResource {
+ private static final Logger LOG = Logger.getLogger(AzureEventhubsResource.class);
@Inject
ProducerTemplate producerTemplate;
- @Inject
- ConsumerTemplate consumerTemplate;
-
@Inject
CamelContext context;
- @ConfigProperty(name = "azure.event.hubs.connection.string")
- Optional<String> connectionString;
-
- private volatile String message;
- private int counter = 0;
-
- /**
- * For some reason if we send just a single message, it is not always received by the consumer.
- * Sending multiple messages seems to be more reliable.
- */
- @Scheduled(every = "1s")
- void schedule() {
- if (message != null) {
- final String endpointUri = "azure-eventhubs:?connectionString=RAW(" + connectionString.get() + ")";
- producerTemplate.sendBody(endpointUri, message + (counter++));
+ @Path("/receive-event/{partitionId}")
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map<String, Object> receiveEvent(@PathParam("partitionId") String partitionId, String match) throws Exception {
+ final MockEndpoint mockEndpoint = context.getEndpoint("mock:partition-%s-results".formatted(partitionId),
+ MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ Optional<Exchange> optionalExchange = receivedExchanges.stream()
+ .filter(exchange -> exchange.getMessage().getBody(String.class).equals(match))
+ .findFirst();
+
+ if (optionalExchange.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ mockEndpoint.reset();
Review Comment:
There was flaky and debugging issues in the past with mock.resetting approach.
Isn't there a reasonable way to avoid this ? Like using a distinct mock per use case ? Actually, do we have multiple call for the same `mock:partition-%s-results` here ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#discussion_r1726482151
##########
integration-test-groups/azure/azure-eventhubs/src/main/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsResource.java:
##########
@@ -17,77 +17,142 @@
package org.apache.camel.quarkus.component.azure.eventhubs.it;
import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
-import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
-import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.mock.MockEndpoint;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.util.ObjectHelper;
+import org.jboss.logging.Logger;
@Path("/azure-eventhubs")
@ApplicationScoped
public class AzureEventhubsResource {
+ private static final Logger LOG = Logger.getLogger(AzureEventhubsResource.class);
@Inject
ProducerTemplate producerTemplate;
- @Inject
- ConsumerTemplate consumerTemplate;
-
@Inject
CamelContext context;
- @ConfigProperty(name = "azure.event.hubs.connection.string")
- Optional<String> connectionString;
-
- private volatile String message;
- private int counter = 0;
-
- /**
- * For some reason if we send just a single message, it is not always received by the consumer.
- * Sending multiple messages seems to be more reliable.
- */
- @Scheduled(every = "1s")
- void schedule() {
- if (message != null) {
- final String endpointUri = "azure-eventhubs:?connectionString=RAW(" + connectionString.get() + ")";
- producerTemplate.sendBody(endpointUri, message + (counter++));
+ @Path("/receive-event/{partitionId}")
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map<String, Object> receiveEvent(@PathParam("partitionId") String partitionId, String match) throws Exception {
+ final MockEndpoint mockEndpoint = context.getEndpoint("mock:partition-%s-results".formatted(partitionId),
+ MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ Optional<Exchange> optionalExchange = receivedExchanges.stream()
+ .filter(exchange -> exchange.getMessage().getBody(String.class).equals(match))
+ .findFirst();
+
+ if (optionalExchange.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ mockEndpoint.reset();
Review Comment:
Looks like there's one place that is not using a unique mock endpoint. So I will change that and remove the mock resetting entirely.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton merged PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "aldettinger (via GitHub)" <gi...@apache.org>.
aldettinger commented on code in PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#discussion_r1726538556
##########
integration-test-groups/azure/azure-eventhubs/src/main/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsResource.java:
##########
@@ -17,77 +17,139 @@
package org.apache.camel.quarkus.component.azure.eventhubs.it;
import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
-import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
-import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.mock.MockEndpoint;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.util.ObjectHelper;
+import org.jboss.logging.Logger;
@Path("/azure-eventhubs")
@ApplicationScoped
public class AzureEventhubsResource {
+ private static final Logger LOG = Logger.getLogger(AzureEventhubsResource.class);
@Inject
ProducerTemplate producerTemplate;
- @Inject
- ConsumerTemplate consumerTemplate;
-
@Inject
CamelContext context;
- @ConfigProperty(name = "azure.event.hubs.connection.string")
- Optional<String> connectionString;
-
- private volatile String message;
- private int counter = 0;
-
- /**
- * For some reason if we send just a single message, it is not always received by the consumer.
- * Sending multiple messages seems to be more reliable.
- */
- @Scheduled(every = "1s")
- void schedule() {
- if (message != null) {
- final String endpointUri = "azure-eventhubs:?connectionString=RAW(" + connectionString.get() + ")";
- producerTemplate.sendBody(endpointUri, message + (counter++));
+ @Path("/receive-event")
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map<String, Object> receiveEvent(@QueryParam("endpointUri") String endpointUri, String match) {
+ final MockEndpoint mockEndpoint = context.getEndpoint(endpointUri, MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ Optional<Exchange> optionalExchange = receivedExchanges.stream()
+ .filter(exchange -> exchange.getMessage().getBody(String.class).equals(match))
+ .findFirst();
+
+ if (optionalExchange.isEmpty()) {
+ return Collections.emptyMap();
}
+
+ Exchange exchange = optionalExchange.get();
+ Message message = exchange.getMessage();
+ return Map.of(
+ "body", message.getBody(String.class),
+ "headers", message.getHeaders());
+ }
+
+ @Path("/send-event/{partitionId}")
+ @POST
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ public Response sendEvent(
+ @PathParam("partitionId") String partitionId,
+ @QueryParam("endpointUri") String endpointUri,
+ String message) throws Exception {
+
+ if (ObjectHelper.isEmpty(endpointUri)) {
+ endpointUri = "direct:sendEvent";
+ }
+
+ LOG.infof("Producing event to endpoint uri: %s", endpointUri);
+
+ producerTemplate.sendBodyAndHeader(endpointUri, message, EventHubsConstants.PARTITION_ID, partitionId);
+ return Response.created(new URI("https://camel.apache.org/")).build();
}
@Path("/receive-events")
@GET
@Produces(MediaType.APPLICATION_JSON)
- public List<String> receiveEvents() throws Exception {
+ public List<Map<String, Object>> receiveEvents(@QueryParam("endpointUri") String endpointUri, List<String> matches) {
+ final MockEndpoint mockEndpoint = context.getEndpoint(endpointUri, MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ List<Exchange> exchanges = receivedExchanges.stream()
+ .filter(exchange -> matches.contains(exchange.getMessage().getBody(String.class)))
+ .toList();
+
+ if (exchanges.isEmpty()) {
+ return Collections.emptyList();
+ }
- final MockEndpoint mockEndpoint = context.getEndpoint("mock:azure-consumed", MockEndpoint.class);
- return mockEndpoint.getReceivedExchanges().stream()
- .map(Exchange::getMessage)
- .map(m -> m.getBody(String.class))
- .collect(Collectors.toList());
+ List<Map<String, Object>> result = new ArrayList<>();
+ for (Exchange exchange : exchanges) {
+ Message message = exchange.getMessage();
+ result.add(Map.of(
+ "body", message.getBody(String.class),
+ "headers", message.getHeaders()));
+ }
+
+ return result;
}
- @Path("/send-events")
+ @Path("/send-events/{partitionId}")
@POST
+ @Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.TEXT_PLAIN)
- @Consumes(MediaType.TEXT_PLAIN)
- public Response sendEvents(String body) throws Exception {
- this.message = body; // start sending the messages via schedule()
+ public Response sendEvents(@PathParam("partitionId") String partitionId, List<String> messages) throws Exception {
+ producerTemplate.sendBodyAndHeader("direct:sendEvent", messages, EventHubsConstants.PARTITION_ID, partitionId);
return Response.created(new URI("https://camel.apache.org/")).build();
}
+ @Path("/route/{routeId}/start")
+ @POST
+ public void startRoute(@PathParam("routeId") String routeId) throws Exception {
+ LOG.infof("Starting route: %s", routeId);
+ context.getRouteController().startRoute(routeId);
+ // A random jitter value is applied in the Event Hubs client before its message listener is active.
+ // In addition, claiming ownership of partitions seems to take an indeterminate amount of time.
+ // Therefore, we need to wait until it's safe to produce events
+ Thread.sleep(5000);
+ }
+
+ @Path("/route/{routeId}/stop")
+ @POST
+ public void stopRoute(@PathParam("routeId") String routeId) throws Exception {
+ context.getRouteController().stopRoute(routeId);
+ }
+
+ @Path("/reset/mocks")
Review Comment:
We could probably remove this one if unused ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#issuecomment-2304008524
Latest commit uses all unique mock endpoints as per @aldettinger feedback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Increase azure-eventhubs test coverage [camel-quarkus]
Posted by "jamesnetherton (via GitHub)" <gi...@apache.org>.
jamesnetherton commented on code in PR #6371:
URL: https://github.com/apache/camel-quarkus/pull/6371#discussion_r1726569976
##########
integration-test-groups/azure/azure-eventhubs/src/main/java/org/apache/camel/quarkus/component/azure/eventhubs/it/AzureEventhubsResource.java:
##########
@@ -17,77 +17,139 @@
package org.apache.camel.quarkus.component.azure.eventhubs.it;
import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
-import io.quarkus.scheduler.Scheduled;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.apache.camel.CamelContext;
-import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
+import org.apache.camel.Message;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.azure.eventhubs.EventHubsConstants;
import org.apache.camel.component.mock.MockEndpoint;
-import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.apache.camel.util.ObjectHelper;
+import org.jboss.logging.Logger;
@Path("/azure-eventhubs")
@ApplicationScoped
public class AzureEventhubsResource {
+ private static final Logger LOG = Logger.getLogger(AzureEventhubsResource.class);
@Inject
ProducerTemplate producerTemplate;
- @Inject
- ConsumerTemplate consumerTemplate;
-
@Inject
CamelContext context;
- @ConfigProperty(name = "azure.event.hubs.connection.string")
- Optional<String> connectionString;
-
- private volatile String message;
- private int counter = 0;
-
- /**
- * For some reason if we send just a single message, it is not always received by the consumer.
- * Sending multiple messages seems to be more reliable.
- */
- @Scheduled(every = "1s")
- void schedule() {
- if (message != null) {
- final String endpointUri = "azure-eventhubs:?connectionString=RAW(" + connectionString.get() + ")";
- producerTemplate.sendBody(endpointUri, message + (counter++));
+ @Path("/receive-event")
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Map<String, Object> receiveEvent(@QueryParam("endpointUri") String endpointUri, String match) {
+ final MockEndpoint mockEndpoint = context.getEndpoint(endpointUri, MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ Optional<Exchange> optionalExchange = receivedExchanges.stream()
+ .filter(exchange -> exchange.getMessage().getBody(String.class).equals(match))
+ .findFirst();
+
+ if (optionalExchange.isEmpty()) {
+ return Collections.emptyMap();
}
+
+ Exchange exchange = optionalExchange.get();
+ Message message = exchange.getMessage();
+ return Map.of(
+ "body", message.getBody(String.class),
+ "headers", message.getHeaders());
+ }
+
+ @Path("/send-event/{partitionId}")
+ @POST
+ @Consumes(MediaType.TEXT_PLAIN)
+ @Produces(MediaType.TEXT_PLAIN)
+ public Response sendEvent(
+ @PathParam("partitionId") String partitionId,
+ @QueryParam("endpointUri") String endpointUri,
+ String message) throws Exception {
+
+ if (ObjectHelper.isEmpty(endpointUri)) {
+ endpointUri = "direct:sendEvent";
+ }
+
+ LOG.infof("Producing event to endpoint uri: %s", endpointUri);
+
+ producerTemplate.sendBodyAndHeader(endpointUri, message, EventHubsConstants.PARTITION_ID, partitionId);
+ return Response.created(new URI("https://camel.apache.org/")).build();
}
@Path("/receive-events")
@GET
@Produces(MediaType.APPLICATION_JSON)
- public List<String> receiveEvents() throws Exception {
+ public List<Map<String, Object>> receiveEvents(@QueryParam("endpointUri") String endpointUri, List<String> matches) {
+ final MockEndpoint mockEndpoint = context.getEndpoint(endpointUri, MockEndpoint.class);
+ List<Exchange> receivedExchanges = mockEndpoint.getReceivedExchanges();
+
+ List<Exchange> exchanges = receivedExchanges.stream()
+ .filter(exchange -> matches.contains(exchange.getMessage().getBody(String.class)))
+ .toList();
+
+ if (exchanges.isEmpty()) {
+ return Collections.emptyList();
+ }
- final MockEndpoint mockEndpoint = context.getEndpoint("mock:azure-consumed", MockEndpoint.class);
- return mockEndpoint.getReceivedExchanges().stream()
- .map(Exchange::getMessage)
- .map(m -> m.getBody(String.class))
- .collect(Collectors.toList());
+ List<Map<String, Object>> result = new ArrayList<>();
+ for (Exchange exchange : exchanges) {
+ Message message = exchange.getMessage();
+ result.add(Map.of(
+ "body", message.getBody(String.class),
+ "headers", message.getHeaders()));
+ }
+
+ return result;
}
- @Path("/send-events")
+ @Path("/send-events/{partitionId}")
@POST
+ @Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.TEXT_PLAIN)
- @Consumes(MediaType.TEXT_PLAIN)
- public Response sendEvents(String body) throws Exception {
- this.message = body; // start sending the messages via schedule()
+ public Response sendEvents(@PathParam("partitionId") String partitionId, List<String> messages) throws Exception {
+ producerTemplate.sendBodyAndHeader("direct:sendEvent", messages, EventHubsConstants.PARTITION_ID, partitionId);
return Response.created(new URI("https://camel.apache.org/")).build();
}
+ @Path("/route/{routeId}/start")
+ @POST
+ public void startRoute(@PathParam("routeId") String routeId) throws Exception {
+ LOG.infof("Starting route: %s", routeId);
+ context.getRouteController().startRoute(routeId);
+ // A random jitter value is applied in the Event Hubs client before its message listener is active.
+ // In addition, claiming ownership of partitions seems to take an indeterminate amount of time.
+ // Therefore, we need to wait until it's safe to produce events
+ Thread.sleep(5000);
+ }
+
+ @Path("/route/{routeId}/stop")
+ @POST
+ public void stopRoute(@PathParam("routeId") String routeId) throws Exception {
+ context.getRouteController().stopRoute(routeId);
+ }
+
+ @Path("/reset/mocks")
Review Comment:
Yep - I forgot 🤦 .
It's removed now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org