Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polish "Add consumer test utility" (#592)

Browse Source 
- update javadocs a bit
- reorder some constructors based on specificity
- add null check to "withSchema"
- rename the test to end w/ "Tests" (plural) for consistency
- simplified doc examples a bit w/ String message type
- drop the exact match API
- rename the conditions factory methods
This commit is contained in:
Chris Bono
2024-02-26 13:17:37 -06:00
committed by GitHub
parent 5bf9ba0944
commit 1f2f6ed46d
12 changed files with 319 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/testing-applications.adoc
View File

@@ -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<Message<MyMessage>> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
List<Message<String>> 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<Message<MyMessage>> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
List<Message<String>> 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<Message<MyMessage>> messages = PulsarConsumerTestUtil.consumeMessages(pulsarConsumerFactory)
List<Message<String>> 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();
----

11
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionTimeoutException.java
View File

@@ -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);
}
}

21
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConditionsSpec.java
View File

@@ -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 <T> the type of the message payload
* @author Jonas Geiregat
@@ -32,27 +30,24 @@ import org.springframework.pulsar.PulsarException;
public interface ConditionsSpec<T> {
/**
* 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<T> 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<T> until(ConsumedMessagesCondition<T> consumedMessagesCondition);
ConditionsSpec<T> until(ConsumedMessagesCondition<T> 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<Message<T>> get();

8
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesCondition.java
View File

@@ -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 <T> the type of the message
* @author Jonas Geiregat
@@ -31,9 +31,9 @@ import org.apache.pulsar.client.api.Message;
public interface ConsumedMessagesCondition<T> {
/**
* 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<Message<T>> messages);

29
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/ConsumedMessagesConditions.java
View File

@@ -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 <T> the type of the message
* @return the condition
*/
static <T> ConsumedMessagesCondition<T> anyMessageMatchesExpected(T expectation) {
return messages -> messages.stream().anyMatch(message -> message.getValue().equals(expectation));
static <T> ConsumedMessagesCondition<T> 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 <T> the type of the message
* @return the condition
*/
@SafeVarargs
@SuppressWarnings("varargs")
static <T> ConsumedMessagesCondition<T> containsAllExpectedValues(T... expectation) {
static <T> ConsumedMessagesCondition<T> 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 <T> the type of the message
* @return the condition
*/
@SafeVarargs
@SuppressWarnings("varargs")
static <T> ConsumedMessagesCondition<T> containsExactlyExpectedValues(T... expectation) {
return ConsumedMessagesConditions.<T>desiredMessageCount(expectation.length)
.and(containsAllExpectedValues(expectation));
}
}

30
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtil.java
View File

@@ -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.
* <p>
* Exposes a Fluent builder-style API to construct the specifications for the message
* consumption.
*
* @param <T> the type of the message payload
* @author Jonas Geiregat
@@ -67,6 +69,13 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, C
return this;
}
@Override
public ConditionsSpec<T> withSchema(Schema<T> schema) {
Assert.notNull(schema, "Schema must not be null");
this.schema = schema;
return this;
}
@Override
public ConditionsSpec<T> awaitAtMost(Duration timeout) {
Assert.notNull(timeout, "Timeout must not be null");
@@ -80,17 +89,11 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, C
return this;
}
@Override
public ConditionsSpec<T> withSchema(Schema<T> schema) {
this.schema = schema;
return this;
}
@Override
public List<Message<T>> get() {
var messages = new ArrayList<Message<T>>();
try {
String subscriptionName = UUID.randomUUID() + "-test-consumer";
var subscriptionName = "test-consumer-%s".formatted(UUID.randomUUID());
try (Consumer<T> consumer = consumerFactory.createConsumer(this.schema, this.topics, subscriptionName,
c -> c.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest))) {
long remainingMillis = timeout.toMillis();
@@ -101,10 +104,8 @@ public class PulsarConsumerTestUtil<T> implements TopicSpec<T>, SchemaSpec<T>, 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<T> implements TopicSpec<T>, SchemaSpec<T>, 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;
}

14
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/PulsarTestException.java
View File

@@ -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);
}
}

2
spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTest.java → spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionTests.java
View File

@@ -27,7 +27,7 @@ import org.junit.jupiter.api.Test;
*
* @author Jonas Geiregat
*/
class ConsumedMessagesConditionTest {
class ConsumedMessagesConditionTests {
@Test
void bothConditionShouldBeMetInOrderForAChainedAndConditionToBeMet() {

155
spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTest.java
View File

@@ -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<Message<String>> createStringMessages(int count) {
return IntStream.range(0, count)
.<Message<String>>mapToObj(i -> MessageImpl.create(new MessageMetadata(),
ByteBuffer.wrap(("message-" + i).getBytes()), Schema.STRING, "topic"))
.toList();
}
@Nested
class DesiredMessageCountTests {
@Test
void receivedMessageCountMeetCondition() {
ConsumedMessagesCondition<String> consumedMessagesCondition = desiredMessageCount(2);
boolean result = consumedMessagesCondition.meets(createStringMessages(2));
Assertions.assertThat(result).isTrue();
}
@Test
void receivedMessageCountDoesNotMeetCondition() {
ConsumedMessagesCondition<String> 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<String> consumedMessagesCondition = ConsumedMessagesConditions
.anyMessageMatchesExpected("message-1");
boolean result = consumedMessagesCondition.meets(createStringMessages(3));
Assertions.assertThat(result).isTrue();
}
@Test
void messageValuesDoesNotContainExpectation() {
ConsumedMessagesCondition<String> consumedMessagesCondition = ConsumedMessagesConditions
.anyMessageMatchesExpected("message-3");
boolean result = consumedMessagesCondition.meets(createStringMessages(3));
Assertions.assertThat(result).isFalse();
}
}
@Nested
class ContainsAllExpectedValuesConditionTests {
@Test
void messageValuesContainsExpectation() {
ConsumedMessagesCondition<String> consumedMessagesCondition = ConsumedMessagesConditions
.containsAllExpectedValues("message-1", "message-2");
boolean result = consumedMessagesCondition.meets(createStringMessages(3));
Assertions.assertThat(result).isTrue();
}
@Test
void messageValuesDoesNotContainExpectation() {
ConsumedMessagesCondition<String> consumedMessagesCondition = ConsumedMessagesConditions
.containsAllExpectedValues("message-3", "message-4");
boolean result = consumedMessagesCondition.meets(createStringMessages(3));
Assertions.assertThat(result).isFalse();
}
}
@Nested
class ContainsExactlyExpectedValuesConditionTests {
@Test
void messageValuesContainsExpectations() {
ConsumedMessagesCondition<String> consumedMessagesCondition = ConsumedMessagesConditions
.containsExactlyExpectedValues("message-0", "message-1");
boolean result = consumedMessagesCondition.meets(createStringMessages(2));
Assertions.assertThat(result).isTrue();
}
@Test
void messageValuesDoesNotContainExpectation() {
ConsumedMessagesCondition<String> consumedMessagesCondition = ConsumedMessagesConditions
.containsExactlyExpectedValues("message-0", "message-1");
boolean result = consumedMessagesCondition.meets(createStringMessages(3));
Assertions.assertThat(result).isFalse();
}
}
}

118
spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/ConsumedMessagesConditionsTests.java Normal file
View File

@@ -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<Message<String>> createStringMessages(int count) {
return IntStream.range(0, count)
.<Message<String>>mapToObj(i -> MessageImpl.create(new MessageMetadata(),
ByteBuffer.wrap(("message-" + i).getBytes()), Schema.STRING, "topic"))
.toList();
}
@Nested
class DesiredMessageCount {
@Test
void receivedMessageCountMeetCondition() {
ConsumedMessagesCondition<String> condition = desiredMessageCount(2);
var messages = createStringMessages(2);
assertThat(condition.meets(messages)).isTrue();
}
@Test
void receivedMessageCountDoesNotMeetCondition() {
ConsumedMessagesCondition<String> 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<String> condition = ConsumedMessagesConditions
.atLeastOneMessageMatches("message-1");
var messages = createStringMessages(3);
assertThat(condition.meets(messages)).isTrue();
}
@Test
void messageValuesDoesNotContainExpectation() {
ConsumedMessagesCondition<String> condition = ConsumedMessagesConditions
.atLeastOneMessageMatches("message-5");
var messages = createStringMessages(3);
assertThat(condition.meets(messages)).isFalse();
}
}
@Nested
class AtLeaseOneMessageMatchesEachOf {
@Test
void messageValuesContainsExpectation() {
ConsumedMessagesCondition<String> condition = ConsumedMessagesConditions
.atLeastOneMessageMatchesEachOf("message-1", "message-2");
var messages = createStringMessages(3);
assertThat(condition.meets(messages)).isTrue();
}
@Test
void messageValuesDoesNotContainExpectation() {
ConsumedMessagesCondition<String> condition = ConsumedMessagesConditions
.atLeastOneMessageMatchesEachOf("message-3", "message-4");
var messages = createStringMessages(3);
assertThat(condition.meets(messages)).isFalse();
}
}
}

132
spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarConsumerTestUtilTests.java Normal file
View File

@@ -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<Object> pulsarTemplate;
private DefaultPulsarConsumerFactory<String> 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");
}
}

133
spring-pulsar-test/src/test/java/org/springframework/pulsar/test/support/PulsarTestConsumerTestUtilTest.java
View File

@@ -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<Object> pulsarTemplate;
private DefaultPulsarConsumerFactory<String> 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<Message<String>> 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<Message<String>> 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");
}
}