From 2c7e3b248a9ffacf482f5c63d7f018815ca15ef5 Mon Sep 17 00:00:00 2001 From: KartikShrivastava Date: Sun, 10 Mar 2024 21:19:53 +0530 Subject: [PATCH] Enhance PulsarConsumerTestUtil consumeMessages This commit allows users to pass in no args, pulsar client, or broker url to PulsarConsumerTestUtil consumeMessages(). Resolves #599 --- .../test/support/PulsarConsumerTestUtil.java | 31 ++++++ .../support/PulsarTestContainerSupport.java | 8 ++ .../support/PulsarConsumerTestUtilTests.java | 99 ++++++++++++++++++- 3 files changed, 136 insertions(+), 2 deletions(-) diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java index a1b2893c..2c0b76ac 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java @@ -24,11 +24,13 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.springframework.pulsar.PulsarException; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.util.Assert; @@ -53,6 +55,30 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C private List topics; + private boolean untilMethodAlreadyCalled = false; + + public static TopicSpec consumeMessages() { + if (PulsarTestContainerSupport.isContainerStarted()) { + return PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl()); + } + return PulsarConsumerTestUtil.consumeMessages("pulsar://localhost:6650"); + } + + public static TopicSpec consumeMessages(String url) { + Assert.notNull(url, "url must not be null"); + try { + return PulsarConsumerTestUtil.consumeMessages(PulsarClient.builder().serviceUrl(url).build()); + } + catch (PulsarClientException ex) { + throw new PulsarException(ex); + } + } + + public static TopicSpec consumeMessages(PulsarClient pulsarClient) { + Assert.notNull(pulsarClient, "pulsarClient must not be null"); + return PulsarConsumerTestUtil.consumeMessages(new DefaultPulsarConsumerFactory<>(pulsarClient, List.of())); + } + public static TopicSpec consumeMessages(PulsarConsumerFactory pulsarConsumerFactory) { return new PulsarConsumerTestUtil<>(pulsarConsumerFactory); } @@ -85,6 +111,11 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C @Override public ConditionsSpec until(ConsumedMessagesCondition condition) { + if (untilMethodAlreadyCalled) { + throw new IllegalStateException( + "Multiple calls to 'until' are not allowed. Use 'and' to combine conditions."); + } + this.untilMethodAlreadyCalled = true; this.condition = condition; return this; } diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java index 9900a1df..03e3b6b7 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestContainerSupport.java @@ -48,4 +48,12 @@ public interface PulsarTestContainerSupport { return PULSAR_CONTAINER.getHttpServiceUrl(); } + static boolean isContainerStarted() { + return PULSAR_CONTAINER.isRunning(); + } + + static void stopContainer() { + PULSAR_CONTAINER.stop(); + } + } diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java index 5401ae59..4da7d0cf 100644 --- a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java +++ b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java @@ -34,6 +34,7 @@ import org.junit.jupiter.api.Test; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarTemplate; /** @@ -47,13 +48,15 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { private DefaultPulsarConsumerFactory pulsarConsumerFactory; + private PulsarClient pulsarClient; + private static String testTopic(String suffix) { return "ptctut-topic-" + suffix; } @BeforeEach void prepareForTest() throws PulsarClientException { - var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); + this.pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); this.pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, List.of()); this.pulsarTemplate = new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient)); } @@ -97,9 +100,101 @@ class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { .withMessage("Condition was not met within 5 seconds"); } + @Test + void messagesAreConsumedWhenContainerIsRunningAndConsumeMessagesIsCalledWithoutArguments() { + // depends upon pulsarClient created in prepareForTest + var topic = testTopic("e1"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages() + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void messagesAreConsumedWhenContainerIsStoppedAndConsumeMessagesIsCalledWithoutArguments() { + PulsarTestContainerSupport.stopContainer(); + PulsarConsumerTestUtil.consumeMessages(); + // TODO: Complete this test + } + + @Test + void messagesAreConsumedWhenConsumeMessagesIsCalledWithBrokerUrl() { + var topic = testTopic("e2"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullBrokerUrl() { + String url = null; + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(url)) + .withMessage("url must not be null"); + } + + @Test + void messagesAreConsumedWhenConsumeMessagesIsCalledWithPulsarClient() { + var topic = testTopic("e3"); + IntStream.range(0, 2).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(this.pulsarClient) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(2)) + .get(); + assertThat(msgs).hasSize(2); + } + + @Test + void exceptionIsThrownWhenConsumeMessagesIsCalledWithNullPulsarClient() { + PulsarClient localPulsarClient = null; + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(localPulsarClient)) + .withMessage("pulsarClient must not be null"); + } + + @Test + void whenChainedConditionAreSpecifiedMessagesAreConsumedUntilTheyAreMet() { + var topic = testTopic("d"); + IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + ConsumedMessagesCondition condition1 = ConsumedMessagesConditions.desiredMessageCount(5); + ConsumedMessagesCondition condition2 = ConsumedMessagesConditions.atLeastOneMessageMatches("message-1"); + var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(condition1.and(condition2)) + .get(); + assertThat(msgs).hasSize(5); + } + + @Test + void exceptionIsThrownWhenUntilIsCalledMultipleTimes() { + var topic = testTopic("e"); + IntStream.range(0, 1).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(ConsumedMessagesConditions.desiredMessageCount(1)) + .until(ConsumedMessagesConditions.atLeastOneMessageMatches("message-0")) + .get()) + .withMessage("Multiple calls to 'until' are not allowed. Use 'and' to combine conditions."); + } + @Test void consumerFactoryCannotBeNull() { - assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(null)) + PulsarConsumerFactory consumerFactory = null; + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(consumerFactory)) .withMessage("PulsarConsumerFactory must not be null"); }