-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
0aa38bc
commit cd0c544
Showing
8 changed files
with
514 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
212 changes: 212 additions & 0 deletions
212
pubsub-plus-connector/src/test/java/io/quarkiverse/solace/perf/EndToEndPerformanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,212 @@ | ||
| package io.quarkiverse.solace.perf; | ||
|
|
||
| import static org.awaitility.Awaitility.await; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.CopyOnWriteArrayList; | ||
|
|
||
| import jakarta.enterprise.context.ApplicationScoped; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Incoming; | ||
| import org.eclipse.microprofile.reactive.messaging.Message; | ||
| import org.eclipse.microprofile.reactive.messaging.Outgoing; | ||
| import org.junit.jupiter.api.Disabled; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import com.solace.messaging.publisher.PersistentMessagePublisher; | ||
| import com.solace.messaging.receiver.PersistentMessageReceiver; | ||
| import com.solace.messaging.resources.Queue; | ||
| import com.solace.messaging.resources.Topic; | ||
| import com.solace.messaging.resources.TopicSubscription; | ||
|
|
||
| import io.quarkiverse.solace.base.WeldTestBase; | ||
| import io.quarkiverse.solace.incoming.SolaceInboundMessage; | ||
| import io.smallrye.mutiny.Multi; | ||
| import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig; | ||
|
|
||
| @Disabled | ||
| public class EndToEndPerformanceTest extends WeldTestBase { | ||
|
|
||
| private static final int COUNT = 100000; | ||
|
|
||
| private static final int TIMEOUT_IN_SECONDS = 400; | ||
|
|
||
| @Test | ||
| public void endToEndPerformanceTesttWithBackPressureWaitAndWaitForPublishReceipt() { | ||
| String processedTopic = topic + "/processed"; | ||
| MapBasedConfig config = new MapBasedConfig() | ||
| .with("mp.messaging.incoming.in.connector", "quarkus-solace") | ||
| .with("mp.messaging.incoming.in.consumer.queue.name", queue) | ||
| .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true") | ||
| .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start") | ||
| .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic) | ||
| .with("mp.messaging.outgoing.out.connector", "quarkus-solace") | ||
| .with("mp.messaging.outgoing.out.producer.topic", processedTopic); | ||
|
|
||
| // Run app that consumes messages | ||
| runApplication(config, MyProcessor.class); | ||
|
|
||
| List<String> received = new CopyOnWriteArrayList<>(); | ||
|
|
||
| // Start listening processed messages | ||
| PersistentMessageReceiver receiver = messagingService.createPersistentMessageReceiverBuilder() | ||
| .withMessageAutoAcknowledgement() | ||
| .withSubscriptions(TopicSubscription.of(processedTopic)) | ||
| .build(Queue.nonDurableExclusiveQueue()); | ||
| receiver.receiveAsync(inboundMessage -> received.add(inboundMessage.getPayloadAsString())); | ||
| receiver.start(); | ||
|
|
||
| // Produce messages | ||
| PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder() | ||
| .build() | ||
| .start(); | ||
| Topic tp = Topic.of(topic); | ||
| for (int i = 0; i < COUNT; i++) { | ||
| publisher.publish(String.valueOf(i + 1), tp); | ||
| } | ||
|
|
||
| await() | ||
| .atMost(Duration.ofSeconds(TIMEOUT_IN_SECONDS)) | ||
| .until(() -> received.size() == COUNT); | ||
| } | ||
|
|
||
| @Test | ||
| public void endToEndPerformanceTesttWithBackPressureWaitAndNoWaitForPublishReceipt() { | ||
| String processedTopic = topic + "/processed"; | ||
| MapBasedConfig config = new MapBasedConfig() | ||
| .with("mp.messaging.incoming.in.connector", "quarkus-solace") | ||
| .with("mp.messaging.incoming.in.consumer.queue.name", queue) | ||
| .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true") | ||
| .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start") | ||
| .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic) | ||
| .with("mp.messaging.outgoing.out.connector", "quarkus-solace") | ||
| .with("mp.messaging.outgoing.out.producer.topic", processedTopic) | ||
| .with("mp.messaging.outgoing.out.producer.waitForPublishReceipt", false); | ||
|
|
||
| // Run app that consumes messages | ||
| runApplication(config, MyProcessor.class); | ||
|
|
||
| List<String> received = new CopyOnWriteArrayList<>(); | ||
|
|
||
| // Start listening processed messages | ||
| PersistentMessageReceiver receiver = messagingService.createPersistentMessageReceiverBuilder() | ||
| .withMessageAutoAcknowledgement() | ||
| .withSubscriptions(TopicSubscription.of(processedTopic)) | ||
| .build(Queue.nonDurableExclusiveQueue()); | ||
| receiver.receiveAsync(inboundMessage -> received.add(inboundMessage.getPayloadAsString())); | ||
| receiver.start(); | ||
|
|
||
| // Produce messages | ||
| PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder() | ||
| .build() | ||
| .start(); | ||
| Topic tp = Topic.of(topic); | ||
| for (int i = 0; i < COUNT; i++) { | ||
| publisher.publish(String.valueOf(i + 1), tp); | ||
| } | ||
|
|
||
| await() | ||
| .atMost(Duration.ofSeconds(TIMEOUT_IN_SECONDS)) | ||
| .until(() -> received.size() == COUNT); | ||
| } | ||
|
|
||
| // | ||
| @Test | ||
| public void endToEndPerformanceTesttWithBackPressureElasticAndWaitForPublishReceipt() { | ||
| String processedTopic = topic + "/processed"; | ||
| MapBasedConfig config = new MapBasedConfig() | ||
| .with("mp.messaging.incoming.in.connector", "quarkus-solace") | ||
| .with("mp.messaging.incoming.in.consumer.queue.name", queue) | ||
| .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true") | ||
| .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start") | ||
| .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic) | ||
| .with("mp.messaging.outgoing.out.connector", "quarkus-solace") | ||
| .with("mp.messaging.outgoing.out.producer.topic", processedTopic) | ||
| .with("mp.messaging.outgoing.out.producer.back-pressure.strategy", "elastic"); | ||
|
|
||
| // Run app that consumes messages | ||
| runApplication(config, MyProcessor.class); | ||
|
|
||
| List<String> received = new CopyOnWriteArrayList<>(); | ||
|
|
||
| // Start listening processed messages | ||
| PersistentMessageReceiver receiver = messagingService.createPersistentMessageReceiverBuilder() | ||
| .withMessageAutoAcknowledgement() | ||
| .withSubscriptions(TopicSubscription.of(processedTopic)) | ||
| .build(Queue.nonDurableExclusiveQueue()); | ||
| receiver.receiveAsync(inboundMessage -> received.add(inboundMessage.getPayloadAsString())); | ||
| receiver.start(); | ||
|
|
||
| // Produce messages | ||
| PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder() | ||
| .build() | ||
| .start(); | ||
| Topic tp = Topic.of(topic); | ||
| for (int i = 0; i < COUNT; i++) { | ||
| publisher.publish(String.valueOf(i + 1), tp); | ||
| } | ||
|
|
||
| await() | ||
| .atMost(Duration.ofSeconds(TIMEOUT_IN_SECONDS)) | ||
| .until(() -> received.size() == COUNT); | ||
| } | ||
|
|
||
| // | ||
| @Test | ||
| public void endToEndPerformanceTesttWithBackPressureElasticAndNoWaitForPublishReceipt() { | ||
| String processedTopic = topic + "/processed"; | ||
| MapBasedConfig config = new MapBasedConfig() | ||
| .with("mp.messaging.incoming.in.connector", "quarkus-solace") | ||
| .with("mp.messaging.incoming.in.consumer.queue.name", queue) | ||
| .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true") | ||
| .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start") | ||
| .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic) | ||
| .with("mp.messaging.outgoing.out.connector", "quarkus-solace") | ||
| .with("mp.messaging.outgoing.out.producer.topic", processedTopic) | ||
| .with("mp.messaging.outgoing.out.producer.back-pressure.strategy", "elastic") | ||
| .with("mp.messaging.outgoing.out.producer.waitForPublishReceipt", false); | ||
|
|
||
| // Run app that consumes messages | ||
| runApplication(config, MyProcessor.class); | ||
|
|
||
| List<String> received = new CopyOnWriteArrayList<>(); | ||
|
|
||
| // Start listening processed messages | ||
| PersistentMessageReceiver receiver = messagingService.createPersistentMessageReceiverBuilder() | ||
| .withMessageAutoAcknowledgement() | ||
| .withSubscriptions(TopicSubscription.of(processedTopic)) | ||
| .build(Queue.nonDurableExclusiveQueue()); | ||
| receiver.receiveAsync(inboundMessage -> received.add(inboundMessage.getPayloadAsString())); | ||
| receiver.start(); | ||
|
|
||
| // Produce messages | ||
| PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder() | ||
| .build() | ||
| .start(); | ||
| Topic tp = Topic.of(topic); | ||
| for (int i = 0; i < COUNT; i++) { | ||
| publisher.publish(String.valueOf(i + 1), tp); | ||
| } | ||
|
|
||
| await() | ||
| .atMost(Duration.ofSeconds(TIMEOUT_IN_SECONDS)) | ||
| .until(() -> received.size() == COUNT); | ||
| } | ||
|
|
||
| @ApplicationScoped | ||
| static class MyProcessor { | ||
| @Incoming("in") | ||
| @Outgoing("out") | ||
| Multi<Message<String>> in(SolaceInboundMessage<byte[]> msg) { | ||
| // return messagingService.messageBuilder().build(payload); | ||
| return Multi.createFrom().items(msg.getMessage().getPayloadAsString()) | ||
| .map(p -> Message.of(p).withAck(() -> { | ||
| msg.ack(); | ||
| return CompletableFuture.completedFuture(null); | ||
| })); | ||
| } | ||
|
|
||
| } | ||
| } |
89 changes: 89 additions & 0 deletions
89
...lus-connector/src/test/java/io/quarkiverse/solace/perf/SolaceConsumerPerformanceTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| package io.quarkiverse.solace.perf; | ||
|
|
||
| import static org.awaitility.Awaitility.await; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletionStage; | ||
| import java.util.concurrent.CopyOnWriteArrayList; | ||
| import java.util.concurrent.atomic.LongAdder; | ||
|
|
||
| import jakarta.enterprise.context.ApplicationScoped; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Incoming; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import com.solace.messaging.publisher.PersistentMessagePublisher; | ||
| import com.solace.messaging.resources.Topic; | ||
|
|
||
| import io.quarkiverse.solace.base.WeldTestBase; | ||
| import io.quarkiverse.solace.incoming.SolaceInboundMessage; | ||
| import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig; | ||
|
|
||
| public class SolaceConsumerPerformanceTest extends WeldTestBase { | ||
| private static final int COUNT = 100000; | ||
| private static final int TIMEOUT_IN_SECONDS = 400; | ||
|
|
||
| @Test | ||
| public void solaceConsumerPerformanceTest() { | ||
| // Produce messages | ||
| PersistentMessagePublisher publisher = messagingService.createPersistentMessagePublisherBuilder() | ||
| .build() | ||
| .start(); | ||
|
|
||
| MapBasedConfig config = new MapBasedConfig() | ||
| .with("mp.messaging.incoming.in.connector", "quarkus-solace") | ||
| .with("mp.messaging.incoming.in.consumer.queue.name", queue) | ||
| .with("mp.messaging.incoming.in.consumer.queue.add-additional-subscriptions", "true") | ||
| .with("mp.messaging.incoming.in.consumer.queue.missing-resource-creation-strategy", "create-on-start") | ||
| .with("mp.messaging.incoming.in.consumer.queue.subscriptions", topic); | ||
| // .with("mp.messaging.incoming.in.client.graceful-shutdown", false); | ||
|
|
||
| // Run app that consumes messages | ||
| MyConsumer app = runApplication(config, MyConsumer.class); | ||
|
|
||
| await().until(() -> isStarted() && isReady()); | ||
|
|
||
| Topic tp = Topic.of(topic); | ||
| for (int i = 0; i < COUNT; i++) { | ||
| publisher.publish(String.valueOf(i + 1), tp); | ||
| } | ||
|
|
||
| await() | ||
| .atMost(Duration.ofSeconds(TIMEOUT_IN_SECONDS)) | ||
| .until(() -> app.getCount() == COUNT); | ||
| long start = app.getStart(); | ||
| long end = System.currentTimeMillis(); | ||
|
|
||
| System.out.println("Total time : " + (end - start) + " ms"); | ||
|
|
||
| } | ||
|
|
||
| @ApplicationScoped | ||
| static class MyConsumer { | ||
| private final List<String> received = new CopyOnWriteArrayList<>(); | ||
| LongAdder count = new LongAdder(); | ||
| long start; | ||
|
|
||
| @Incoming("in") | ||
| public CompletionStage<Void> in(SolaceInboundMessage<byte[]> msg) { | ||
| if (count.longValue() == 0L) { | ||
| start = System.currentTimeMillis(); | ||
| } | ||
| count.increment(); | ||
| return msg.ack(); | ||
| } | ||
|
|
||
| public List<String> getReceived() { | ||
| return received; | ||
| } | ||
|
|
||
| public long getStart() { | ||
| return start; | ||
| } | ||
|
|
||
| public long getCount() { | ||
| return count.longValue(); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.