diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/testing-applications.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/testing-applications.adoc index 7fd2eae8..f4544525 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/testing-applications.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/testing-applications.adoc @@ -1,42 +1,46 @@ [[testing-applications]] = Testing Applications + include::../attributes/attributes.adoc[] The `spring-pulsar-test` dependency includes some useful utilities when testing your applications. == PulsarConsumerTestUtil -`org.springframework.pulsar.test.support.PulsarConsumerTestUtil` provides a type-safe fluent API for consuming messages from a Pulsar topic within a test. +The `org.springframework.pulsar.test.support.PulsarConsumerTestUtil` provides a type-safe fluent API for consuming messages from a Pulsar topic within a test. -[source,java] +The following example shows how to consume messages from a topic for 5 seconds: +[source,java,indent=0,subs="verbatim"] ---- -List> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) +List> messages = PulsarConsumerTestUtil.consumeMessages(consumerFactory) .fromTopic("my-topic") - .withSchema(Schema.JSON(MyMessage.class)) + .withSchema(Schema.STRING) .awaitAtMost(Duration.ofSeconds(5)) .get(); ---- -A `until` method is also available to allow you to specify a condition that must be met before the messages are returned. +An `until` method is also available to allow you to specify a condition that must be met before the messages are returned. +The following example uses a condition to consume 5 messages from a topic. -[source,java] +[source,java,indent=0,subs="verbatim"] ---- -List> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) +List> messages = PulsarConsumerTestUtil.consumeMessages(consumerFactory) .fromTopic("my-topic") - .withSchema(Schema.JSON(MyMessage.class)) - .until(messages -> messages.size() == 5) + .withSchema(Schema.STRING) .awaitAtMost(Duration.ofSeconds(5)) + .until(messages -> messages.size() == 5) .get(); ---- A set of commonly used conditions are available in `org.springframework.pulsar.test.support.ConsumedMessagesConditions`. +The following example uses the factory-provided `atLeastOneMessageMatches` condition to return the consumed messages once one of them has a value of `"boom"`. -[source,java] +[source,java,indent=0,subs="verbatim"] ---- -List> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) +List> messages = PulsarConsumerTestUtil.consumeMessages(consumerFactory) .fromTopic("my-topic") - .withSchema(Schema.JSON(MyMessage.class)) + .withSchema(Schema.STRING) .awaitAtMost(Duration.ofSeconds(5)) - .until(containsExactlyExpectedValues(new MyMessage("foo"), new MyMessage("bar"))) + .until(ConsumedMessagesConditions.atLeastOneMessageMatches("boom")) .get(); ---- diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionTimeoutException.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionTimeoutException.java index af4aefc8..cd841f6f 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionTimeoutException.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionTimeoutException.java @@ -17,18 +17,19 @@ package org.springframework.pulsar.test.support; /** - * Exception thrown when a test times out. + * Exception thrown when a condition was not fulfilled within the specified timeout. * * @author Jonas Geiregat */ public class ConditionTimeoutException extends PulsarTestException { - public ConditionTimeoutException(String message, Throwable exception) { - super(message, exception); - } - public ConditionTimeoutException(String message) { super(message); } + @SuppressWarnings("unused") + public ConditionTimeoutException(String message, Throwable exception) { + super(message, exception); + } + } diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionsSpec.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionsSpec.java index dfce3a8f..44a7cdb7 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionsSpec.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionsSpec.java @@ -21,10 +21,8 @@ import java.util.List; import org.apache.pulsar.client.api.Message; -import org.springframework.pulsar.PulsarException; - /** - * Assertions related step in the fluent API for building a Pulsar test consumer. + * Conditions related step in the fluent API for building a Pulsar test consumer. * * @param the type of the message payload * @author Jonas Geiregat @@ -32,27 +30,24 @@ import org.springframework.pulsar.PulsarException; public interface ConditionsSpec { /** - * The maximum timeout duration to wait for the desired number of messages to be - * reached. - * @param timeout the maximum timeout duration to wait + * The maximum amount of time to consume messages and wait for the condition to be + * satisfied. + * @param timeout the maximum amount of time for the condition to be met * @return the next step in the fluent API */ ConditionsSpec awaitAtMost(Duration timeout); /** - * Start consuming until the given condition is met. - * @param consumedMessagesCondition the condition to be met + * Consume messages until the condition is satisfied. + * @param condition the condition to be met * @return the next step in the fluent API */ - ConditionsSpec until(ConsumedMessagesCondition consumedMessagesCondition); + ConditionsSpec until(ConsumedMessagesCondition condition); /** - * - * Terminal operation that will get the consumed messages within the timeout verifying - * the given condition if any. + * Terminal operation that begins the message consumption using the configured specs. * @return the consumed messages * @throws ConditionTimeoutException if the condition is not met within the timeout - * @throws PulsarException if the condition is not met within the timeout */ List> get(); diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesCondition.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesCondition.java index b85aeabd..143016aa 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesCondition.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesCondition.java @@ -21,8 +21,8 @@ import java.util.List; import org.apache.pulsar.client.api.Message; /** - * A condition to be used in {@link PulsarConsumerTestUtil} to verify if it meets the - * consumed messages. + * A condition to be used in {@link PulsarConsumerTestUtil} to verify if the consumed + * messages satisfy the given criteria. * * @param the type of the message * @author Jonas Geiregat @@ -31,9 +31,9 @@ import org.apache.pulsar.client.api.Message; public interface ConsumedMessagesCondition { /** - * Verifies that the consumed messages meet the condition. + * Determines if the consumed messages meets the condition. * @param messages the consumed messages - * @return {@code true} if the condition is met + * @return whether the consumed messages meet the condition */ boolean meets(List> messages); diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesConditions.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesConditions.java index af30c8e1..e8da8618 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesConditions.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesConditions.java @@ -23,7 +23,8 @@ import org.apache.pulsar.client.api.Message; import org.springframework.util.Assert; /** - * Exposes a set of commonly used conditions to be used in {@link PulsarConsumerTestUtil}. + * A factory for creating commonly used {@link ConsumedMessagesCondition conditions} that + * can be used with {@link PulsarConsumerTestUtil}. * * @author Jonas Geiregat */ @@ -41,42 +42,30 @@ public interface ConsumedMessagesConditions { } /** - * Verifies that any of the consumed messages has a payload that equals the specified - * value. + * Verifies that the expected value equals the message payload value of at least one + * consumed message. * @param expectation the expected value * @param the type of the message * @return the condition */ - static ConsumedMessagesCondition anyMessageMatchesExpected(T expectation) { - return messages -> messages.stream().anyMatch(message -> message.getValue().equals(expectation)); + static ConsumedMessagesCondition atLeastOneMessageMatches(T expectation) { + return messages -> messages.stream().map(Message::getValue).anyMatch(expectation::equals); } /** - * Verifies that the consumed messages value contains at all expected values. + * Verifies that each expected value equals the message payload value of at least one + * consumed message. * @param expectation the expected values * @param the type of the message * @return the condition */ @SafeVarargs @SuppressWarnings("varargs") - static ConsumedMessagesCondition containsAllExpectedValues(T... expectation) { + static ConsumedMessagesCondition atLeastOneMessageMatchesEachOf(T... expectation) { return messages -> { var values = messages.stream().map(Message::getValue).toList(); return Stream.of(expectation).allMatch(values::contains); }; } - /** - * Verifies that the consumed messages value contains exactly the expected values. - * @param expectation the expected values - * @param the type of the message - * @return the condition - */ - @SafeVarargs - @SuppressWarnings("varargs") - static ConsumedMessagesCondition containsExactlyExpectedValues(T... expectation) { - return ConsumedMessagesConditions.desiredMessageCount(expectation.length) - .and(containsAllExpectedValues(expectation)); - } - } 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 97011155..a1b2893c 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 @@ -33,8 +33,10 @@ import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.util.Assert; /** - * Fluent API, to be used in tests, for consuming messages from Pulsar topics until a - * certain {@code Condition} has been met. + * Utility for consuming messages from Pulsar topics. + *

+ * Exposes a Fluent builder-style API to construct the specifications for the message + * consumption. * * @param the type of the message payload * @author Jonas Geiregat @@ -67,6 +69,13 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C return this; } + @Override + public ConditionsSpec withSchema(Schema schema) { + Assert.notNull(schema, "Schema must not be null"); + this.schema = schema; + return this; + } + @Override public ConditionsSpec awaitAtMost(Duration timeout) { Assert.notNull(timeout, "Timeout must not be null"); @@ -80,17 +89,11 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C return this; } - @Override - public ConditionsSpec withSchema(Schema schema) { - this.schema = schema; - return this; - } - @Override public List> get() { var messages = new ArrayList>(); try { - String subscriptionName = UUID.randomUUID() + "-test-consumer"; + var subscriptionName = "test-consumer-%s".formatted(UUID.randomUUID()); try (Consumer consumer = consumerFactory.createConsumer(this.schema, this.topics, subscriptionName, c -> c.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest))) { long remainingMillis = timeout.toMillis(); @@ -101,10 +104,8 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C messages.add(message); consumer.acknowledge(message); } - if (this.condition != null) { - if (this.condition.meets(messages)) { - return messages; - } + if (this.condition != null && this.condition.meets(messages)) { + return messages; } remainingMillis -= System.currentTimeMillis() - loopStartTime; } @@ -115,7 +116,8 @@ public class PulsarConsumerTestUtil implements TopicSpec, SchemaSpec, C throw new PulsarException(ex); } if (this.condition != null && !this.condition.meets(messages)) { - throw new ConditionTimeoutException("Condition was not met within " + timeout.toSeconds() + " seconds"); + throw new ConditionTimeoutException( + "Condition was not met within %d seconds".formatted(timeout.toSeconds())); } return messages; } diff --git a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestException.java b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestException.java index fd037507..36328705 100644 --- a/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestException.java +++ b/spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestException.java @@ -16,20 +16,22 @@ package org.springframework.pulsar.test.support; +import org.springframework.core.NestedRuntimeException; + /** - * Exception thrown when a test fails. + * Generic exception thrown when something related to testing fails. * * @author Jonas Geiregat */ -public class PulsarTestException extends RuntimeException { - - public PulsarTestException(String message, Throwable exception) { - super(message, exception); - } +public class PulsarTestException extends NestedRuntimeException { @SuppressWarnings("unused") public PulsarTestException(String message) { super(message); } + public PulsarTestException(String message, Throwable exception) { + super(message, exception); + } + } diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTest.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTests.java similarity index 97% rename from spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTest.java rename to spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTests.java index d47d6014..000dcdae 100644 --- a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTest.java +++ b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTests.java @@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test; * * @author Jonas Geiregat */ -class ConsumedMessagesConditionTest { +class ConsumedMessagesConditionTests { @Test void bothConditionShouldBeMetInOrderForAChainedAndConditionToBeMet() { diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTest.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTest.java deleted file mode 100644 index a891fd48..00000000 --- a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTest.java +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Copyright 2024 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.pulsar.test.support; - -import static org.springframework.pulsar.test.support.ConsumedMessagesConditions.desiredMessageCount; - -import java.nio.ByteBuffer; -import java.util.List; -import java.util.stream.IntStream; - -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.MessageImpl; -import org.apache.pulsar.common.api.proto.MessageMetadata; -import org.assertj.core.api.Assertions; -import org.junit.jupiter.api.Nested; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; - -/** - * Tests for {@link ConsumedMessagesConditions}. - * - * @author Jonas Geiregat - */ -class ConsumedMessagesConditionsTest { - - private List> createStringMessages(int count) { - return IntStream.range(0, count) - .>mapToObj(i -> MessageImpl.create(new MessageMetadata(), - ByteBuffer.wrap(("message-" + i).getBytes()), Schema.STRING, "topic")) - .toList(); - } - - @Nested - class DesiredMessageCountTests { - - @Test - void receivedMessageCountMeetCondition() { - ConsumedMessagesCondition consumedMessagesCondition = desiredMessageCount(2); - - boolean result = consumedMessagesCondition.meets(createStringMessages(2)); - - Assertions.assertThat(result).isTrue(); - } - - @Test - void receivedMessageCountDoesNotMeetCondition() { - ConsumedMessagesCondition consumedMessagesCondition = desiredMessageCount(3); - - boolean result = consumedMessagesCondition.meets(createStringMessages(2)); - - Assertions.assertThat(result).isFalse(); - } - - @ParameterizedTest - @ValueSource(ints = { 0, -1 }) - void throwExceptionWhenDesiredMessageCountEqualOrLessThanZero(int messageCount) { - Assertions.assertThatThrownBy(() -> desiredMessageCount(messageCount)) - .isInstanceOf(IllegalStateException.class) - .hasMessage("Desired message count must be greater than 0"); - } - - } - - @Nested - class AnyMessageMatchesExpectedConditionTests { - - @Test - void messageValuesContainsExpectation() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .anyMessageMatchesExpected("message-1"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(3)); - - Assertions.assertThat(result).isTrue(); - } - - @Test - void messageValuesDoesNotContainExpectation() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .anyMessageMatchesExpected("message-3"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(3)); - - Assertions.assertThat(result).isFalse(); - } - - } - - @Nested - class ContainsAllExpectedValuesConditionTests { - - @Test - void messageValuesContainsExpectation() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .containsAllExpectedValues("message-1", "message-2"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(3)); - - Assertions.assertThat(result).isTrue(); - } - - @Test - void messageValuesDoesNotContainExpectation() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .containsAllExpectedValues("message-3", "message-4"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(3)); - - Assertions.assertThat(result).isFalse(); - } - - } - - @Nested - class ContainsExactlyExpectedValuesConditionTests { - - @Test - void messageValuesContainsExpectations() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .containsExactlyExpectedValues("message-0", "message-1"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(2)); - - Assertions.assertThat(result).isTrue(); - } - - @Test - void messageValuesDoesNotContainExpectation() { - ConsumedMessagesCondition consumedMessagesCondition = ConsumedMessagesConditions - .containsExactlyExpectedValues("message-0", "message-1"); - - boolean result = consumedMessagesCondition.meets(createStringMessages(3)); - - Assertions.assertThat(result).isFalse(); - } - - } - -} diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTests.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTests.java new file mode 100644 index 00000000..e3d05dce --- /dev/null +++ b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTests.java @@ -0,0 +1,118 @@ +/* + * Copyright 2024 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.pulsar.test.support; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.springframework.pulsar.test.support.ConsumedMessagesConditions.desiredMessageCount; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.stream.IntStream; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.common.api.proto.MessageMetadata; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Tests for {@link ConsumedMessagesConditions}. + * + * @author Jonas Geiregat + */ +class ConsumedMessagesConditionsTests { + + private List> createStringMessages(int count) { + return IntStream.range(0, count) + .>mapToObj(i -> MessageImpl.create(new MessageMetadata(), + ByteBuffer.wrap(("message-" + i).getBytes()), Schema.STRING, "topic")) + .toList(); + } + + @Nested + class DesiredMessageCount { + + @Test + void receivedMessageCountMeetCondition() { + ConsumedMessagesCondition condition = desiredMessageCount(2); + var messages = createStringMessages(2); + assertThat(condition.meets(messages)).isTrue(); + } + + @Test + void receivedMessageCountDoesNotMeetCondition() { + ConsumedMessagesCondition condition = desiredMessageCount(3); + var messages = createStringMessages(2); + assertThat(condition.meets(messages)).isFalse(); + } + + @ParameterizedTest + @ValueSource(ints = { 0, -1 }) + void throwExceptionWhenDesiredMessageCountEqualOrLessThanZero(int messageCount) { + assertThatIllegalStateException().isThrownBy(() -> desiredMessageCount(messageCount)) + .withMessage("Desired message count must be greater than 0"); + } + + } + + @Nested + class AtLeaseOneMessageMatches { + + @Test + void messageValuesContainsExpectation() { + ConsumedMessagesCondition condition = ConsumedMessagesConditions + .atLeastOneMessageMatches("message-1"); + var messages = createStringMessages(3); + assertThat(condition.meets(messages)).isTrue(); + } + + @Test + void messageValuesDoesNotContainExpectation() { + ConsumedMessagesCondition condition = ConsumedMessagesConditions + .atLeastOneMessageMatches("message-5"); + var messages = createStringMessages(3); + assertThat(condition.meets(messages)).isFalse(); + } + + } + + @Nested + class AtLeaseOneMessageMatchesEachOf { + + @Test + void messageValuesContainsExpectation() { + ConsumedMessagesCondition condition = ConsumedMessagesConditions + .atLeastOneMessageMatchesEachOf("message-1", "message-2"); + var messages = createStringMessages(3); + assertThat(condition.meets(messages)).isTrue(); + } + + @Test + void messageValuesDoesNotContainExpectation() { + ConsumedMessagesCondition condition = ConsumedMessagesConditions + .atLeastOneMessageMatchesEachOf("message-3", "message-4"); + var messages = createStringMessages(3); + assertThat(condition.meets(messages)).isFalse(); + } + + } + +} 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 new file mode 100644 index 00000000..5401ae59 --- /dev/null +++ b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java @@ -0,0 +1,132 @@ +/* + * Copyright 2024 the original author or authors. + * + * Licensed 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 + * + * https://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.springframework.pulsar.test.support; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.springframework.pulsar.test.support.ConsumedMessagesConditions.desiredMessageCount; + +import java.time.Duration; +import java.util.List; +import java.util.stream.IntStream; + +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.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarTemplate; + +/** + * Tests for {@link PulsarConsumerTestUtil}. + * + * @author Jonas Geiregat + */ +class PulsarConsumerTestUtilTests implements PulsarTestContainerSupport { + + private PulsarTemplate pulsarTemplate; + + private DefaultPulsarConsumerFactory pulsarConsumerFactory; + + private static String testTopic(String suffix) { + return "ptctut-topic-" + suffix; + } + + @BeforeEach + void prepareForTest() throws PulsarClientException { + var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); + this.pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, List.of()); + this.pulsarTemplate = new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient)); + } + + @Test + void whenConditionIsSpecifiedMessagesAreConsumedUntilConditionIsMet() { + var topic = testTopic("a"); + IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(desiredMessageCount(3)) + .get(); + assertThat(msgs).hasSize(3); + } + + @Test + void whenConditionIsNotSpecifiedMessagesAreConsumedUntilAwaitDuration() { + var topic = testTopic("b"); + IntStream.range(0, 5).forEach(i -> pulsarTemplate.send(topic, "message-" + i)); + var msgs = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(topic) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(null) + .get(); + assertThat(msgs).extracting(Message::getValue) + .containsExactlyInAnyOrderElementsOf(IntStream.range(0, 5).mapToObj(i -> "message-" + i).toList()); + } + + @Test + void exceptionIsThrownWhenConditionNotMetWithinAwaitDuration() { + assertThatExceptionOfType(ConditionTimeoutException.class) + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic(testTopic("c")) + .withSchema(Schema.STRING) + .awaitAtMost(Duration.ofSeconds(5)) + .until(ConsumedMessagesConditions.desiredMessageCount(3)) + .get()) + .withMessage("Condition was not met within 5 seconds"); + } + + @Test + void consumerFactoryCannotBeNull() { + assertThatIllegalArgumentException().isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(null)) + .withMessage("PulsarConsumerFactory must not be null"); + } + + @Test + void topicCannotBeNull() { + assertThatIllegalArgumentException() + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory).fromTopic(null)) + .withMessage("Topic must not be null"); + } + + @Test + void schemaCannotBeNull() { + assertThatIllegalArgumentException() + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic("topic-foo") + .withSchema(null)) + .withMessage("Schema must not be null"); + } + + @Test + void awaitAtMostCannotBeNull() { + assertThatIllegalArgumentException() + .isThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) + .fromTopic("topic-foo") + .withSchema(Schema.STRING) + .awaitAtMost(null)) + .withMessage("Timeout must not be null"); + } + +} diff --git a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarTestConsumerTestUtilTest.java b/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarTestConsumerTestUtilTest.java deleted file mode 100644 index 238d07ba..00000000 --- a/spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarTestConsumerTestUtilTest.java +++ /dev/null @@ -1,133 +0,0 @@ -/* - * Copyright 2024 the original author or authors. - * - * Licensed 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 - * - * https://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.springframework.pulsar.test.support; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.springframework.pulsar.test.support.ConsumedMessagesConditions.desiredMessageCount; - -import java.time.Duration; -import java.util.List; -import java.util.stream.IntStream; - -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.assertj.core.api.ThrowableAssert.ThrowingCallable; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; - -import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.PulsarTemplate; - -/** - * Tests for {@link PulsarConsumerTestUtil}. - * - * @author Jonas Geiregat - */ -class PulsarTestConsumerTestUtilTest implements PulsarTestContainerSupport { - - private PulsarTemplate pulsarTemplate; - - private DefaultPulsarConsumerFactory pulsarConsumerFactory; - - @BeforeEach - void setup() throws PulsarClientException { - var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); - this.pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, List.of()); - this.pulsarTemplate = new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient)); - } - - @Test - void consumerFactoryCannotBeNull() { - assertThatThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(null)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("PulsarConsumerFactory must not be null"); - } - - @Test - void topicCannotBeNull() { - assertThatThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory).fromTopic(null)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Topic must not be null"); - } - - @Test - void awaitAtMostTimeoutCannotBeNull() { - assertThatThrownBy(() -> PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic("topic-a") - .withSchema(Schema.STRING) - .awaitAtMost(null)).isInstanceOf(IllegalArgumentException.class).hasMessage("Timeout must not be null"); - } - - @Test - void untilConditionCanBeNull() { - var testConsumer = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic("topic-b") - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(2)) - .until(null); - - pulsarTemplate.send("topic-b", "message"); - - assertThat(testConsumer.get()).hasSize(1).map(Message::getValue).containsExactly("message"); - } - - @Test - void consumerReturnsWhenConditionIsMet() { - var testConsumer = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic("topic-c") - .withSchema(Schema.STRING); - - IntStream.range(0, 10).forEach(i -> pulsarTemplate.send("topic-c", "message-" + i)); - - List> messages = testConsumer.until(desiredMessageCount(2)).get(); - - assertThat(messages).hasSize(2).map(Message::getValue).containsExactly("message-0", "message-1"); - } - - @Test - void consumerReturnsAllMessagesWhenNoConditionIsPresent() { - var testConsumer = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic("topic-d") - .withSchema(Schema.STRING); - - IntStream.range(0, 10).forEach(i -> pulsarTemplate.send("topic-d", "message-" + i)); - - List> messages = testConsumer.get(); - - assertThat(messages).hasSize(10) - .map(Message::getValue) - .containsExactlyElementsOf(IntStream.range(0, 10).mapToObj(i -> "message-" + i).toList()); - } - - @Test - void throwExceptionWhenConditionIsNotMet() { - var testConsumer = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory) - .fromTopic("topic-e") - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)); - - ThrowingCallable consume = () -> testConsumer.until(desiredMessageCount(20)).get(); - - assertThatThrownBy(consume).isInstanceOf(ConditionTimeoutException.class) - .hasMessage("Condition was not met within 5 seconds"); - } - -}