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

Add test cases to PulsarTemplateTests

Browse Source
This commit is contained in:
Chris Bono
2022-07-24 00:58:49 -05:00
parent 7bbe3250d0
commit d9cdec3f1b
6 changed files with 121 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -147,6 +147,7 @@ subprojects { subproject ->
dependencies {
implementation "com.google.code.findbugs:jsr305:$googleJsr305Version"
testImplementation 'org.junit.jupiter:junit-jupiter-api'
testImplementation 'org.junit.jupiter:junit-jupiter-params'
testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
testRuntimeOnly 'org.junit.platform:junit-platform-launcher'

29
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java
View File

@@ -38,7 +38,7 @@ public interface PulsarOperations<T> {
* @throws PulsarClientException if an error occurs
*/
default MessageId send(T message) throws PulsarClientException {
return send(null, message);
return send(null, message, null);
}
/**
@@ -48,7 +48,30 @@ public interface PulsarOperations<T> {
* @return the id of the sent message
* @throws PulsarClientException if an error occurs
*/
MessageId send(String topic, T message) throws PulsarClientException;
default MessageId send(String topic, T message) throws PulsarClientException {
return send(topic, message, null);
}
/**
* Sends a message to the default topic in a blocking manner.
* @param message the message to send
* @param messageRouter the optional message router to use
* @return the id of the sent message
* @throws PulsarClientException if an error occurs
*/
default MessageId send(T message, MessageRouter messageRouter) throws PulsarClientException {
return send(null, message, messageRouter);
}
/**
* Sends a message to the specified topic in a blocking manner.
* @param topic the topic to send the message to or {@code null} to send to the default topic
* @param message the message to send
* @param messageRouter the optional message router to use
* @return the id of the sent message
* @throws PulsarClientException if an error occurs
*/
MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException;
/**
* Sends a message to the default topic in a blocking manner.
@@ -57,7 +80,7 @@ public interface PulsarOperations<T> {
* @throws PulsarClientException if an error occurs
*/
default CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
return sendAsync(null, message);
return sendAsync(null, message, null);
}
/**

4
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -50,9 +50,9 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
}
@Override
public MessageId send(String topic, T message) throws PulsarClientException {
public MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException {
try {
return this.sendAsync(topic, message).get();
return this.sendAsync(topic, message, messageRouter).get();
}
catch (Exception ex) {
throw PulsarClientException.unwrap(ex);

10
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java
View File

@@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -45,12 +46,19 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests {
protected PulsarClient pulsarClient;
@BeforeEach
void setup() throws PulsarClientException {
void createPulsarClient() throws PulsarClientException {
pulsarClient = PulsarClient.builder()
.serviceUrl(getPulsarBrokerUrl())
.build();
}
@AfterEach
void closePulsarClient() throws PulsarClientException {
if (pulsarClient != null && !pulsarClient.isClosed()) {
pulsarClient.close();
}
}
@Test
void createProducerWithSpecificTopic() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());

163
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
View File

@@ -17,110 +17,111 @@
package org.springframework.pulsar.core;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.params.provider.Arguments.arguments;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.HashMap;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.MessageRouter;
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.Test;
import org.apache.pulsar.client.api.TopicMetadata;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Named;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
/**
* Tests for {@code PulsarTemplate}.
*
* @author Soby Chacko
* @author Chris Bono
*/
class PulsarTemplateTests extends AbstractContainerBaseTests {
public static final String TEST_TOPIC = "test_topic";
@Test
void testUsage() throws Exception {
testPulsarFunctionality(getPulsarBrokerUrl());
}
@Test
void testSendAsync() throws Exception {
Map<String, Object> config = new HashMap<>();
config.put("topicName", "foo-bar-123");
Map<String, Object> clientConfig = new HashMap<>();
clientConfig.put("serviceUrl", getPulsarBrokerUrl());
try (
PulsarClient client = PulsarClient.builder()
.loadConf(clientConfig)
.build();
Consumer<String> consumer = client.newConsumer(Schema.STRING)
.topic("foo-bar-123")
.subscriptionName("xyz-test-subs-123")
.subscribe()
) {
final DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config);
final PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
final CompletableFuture<MessageId> future = pulsarTemplate.sendAsync("hello john doe");
future.thenAccept(m -> { });
try {
Thread.sleep(2000);
future.get();
@ParameterizedTest(name = "{0}")
@MethodSource("sendMessageTestProvider")
void sendMessageTest(String topic, Map<String, Object> producerConfig, SendHandler<Object> handler, MessageRouter router) throws Exception {
String subscription = topic + "-sub";
String msgPayload = topic + "-msg";
if (router != null) {
try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build()) {
admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1);
}
catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) {
try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription).subscribe()) {
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client, producerConfig);
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory);
Object sendResponse = handler.doSend(pulsarTemplate, topic, msgPayload, router);
if (sendResponse instanceof CompletableFuture) {
sendResponse = ((CompletableFuture<?>) sendResponse).get(3, TimeUnit.SECONDS);
}
assertThat(sendResponse).isNotNull();
CompletableFuture<Message<String>> receiveMsgFuture = consumer.receiveAsync();
Message<String> msg = receiveMsgFuture.get(3, TimeUnit.SECONDS);
assertThat(msg.getData()).asString().isEqualTo(msgPayload);
// Make sure the producer was closed by the template (albeit indirectly as client removes closed producers)
await().atMost(Duration.ofSeconds(3)).untilAsserted(() ->
assertThat(client).extracting("producers").asInstanceOf(InstanceOfAssertFactories.COLLECTION).isEmpty());
}
CompletableFuture<Message<String>> future0 = consumer.receiveAsync();
Message<String> message = future0.get(5, TimeUnit.SECONDS);
assertThat(new String(message.getData()))
.isEqualTo("hello john doe");
}
}
@Test
void testSendSync() throws Exception {
Map<String, Object> config = new HashMap<>();
config.put("topicName", "foo-bar-123");
Map<String, Object> clientConfig = new HashMap<>();
clientConfig.put("serviceUrl", getPulsarBrokerUrl());
try (
PulsarClient client = PulsarClient.builder()
.loadConf(clientConfig)
.build();
Consumer<String> consumer = client.newConsumer(Schema.STRING)
.topic("foo-bar-123")
.subscriptionName("xyz-test-subs-123")
.subscribe();
) {
final DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config);
final PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
final MessageId messageId = pulsarTemplate.send("hello john doe");
CompletableFuture<Message<String>> future0 = consumer.receiveAsync();
Message<String> message = future0.get(5, TimeUnit.SECONDS);
assertThat(new String(message.getData()))
.isEqualTo("hello john doe");
}
static Stream<Arguments> sendMessageTestProvider() {
return Stream.of(
arguments(Named.of("sendMessageToDefaultTopicNoRouter", "smt-topic-1"), Collections.singletonMap("topicName", "smt-topic-1"),
(SendHandler<MessageId>) (template, topic, msg, router) -> template.send(msg),
null),
arguments(Named.of("sendMessageToDefaultTopicWithRouter", "smt-topic-2"), Collections.singletonMap("topicName", "smt-topic-2"),
(SendHandler<MessageId>) (template, topic, msg, router) -> template.send(msg, router),
mockRouter()),
arguments(Named.of("sendMessageToSpecificTopicNoRouter", "smt-topic-3"), Collections.emptyMap(),
(SendHandler<MessageId>) (template, topic, msg, router) -> template.send(topic, msg),
null),
arguments(Named.of("sendMessageToSpecificTopicWithRouter", "smt-topic-4"), Collections.emptyMap(),
(SendHandler<MessageId>) PulsarTemplate::send,
mockRouter()),
arguments(Named.of("sendAsyncMessageToDefaultTopicNoRouter", "smt-topic-5"), Collections.singletonMap("topicName", "smt-topic-5"),
(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, router) -> template.sendAsync(msg),
null),
arguments(Named.of("sendAsyncMessageToDefaultTopicWithRouter", "smt-topic-6"), Collections.singletonMap("topicName", "smt-topic-6"),
(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, router) -> template.sendAsync(msg, router),
mockRouter()),
arguments(Named.of("sendAsyncMessageToSpecificTopicNoRouter", "smt-topic-7"), Collections.emptyMap(),
(SendHandler<CompletableFuture<MessageId>>) (template, topic, msg, router) -> template.sendAsync(topic, msg),
null),
arguments(Named.of("sendAsyncMessageToSpecificTopicWithRouter", "smt-topic-8"), Collections.emptyMap(),
(SendHandler<CompletableFuture<MessageId>>) PulsarTemplate::sendAsync,
mockRouter())
);
}
private void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception {
try (
PulsarClient client = PulsarClient.builder()
.serviceUrl(pulsarBrokerUrl)
.build();
Consumer<byte[]> consumer = client.newConsumer()
.topic(TEST_TOPIC)
.subscriptionName("test-subs")
.subscribe();
Producer<byte[]> producer = client.newProducer()
.topic(TEST_TOPIC)
.create()
) {
producer.send("test containers".getBytes());
CompletableFuture<Message<byte[]>> future = consumer.receiveAsync();
Message<byte[]> message = future.get(5, TimeUnit.SECONDS);
private static MessageRouter mockRouter() {
MessageRouter router = mock(MessageRouter.class);
when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0);
return router;
}
assertThat(new String(message.getData()))
.isEqualTo("test containers");
}
@FunctionalInterface
interface SendHandler<V> {
V doSend(PulsarTemplate<String> template, String topic, String msg, MessageRouter router) throws PulsarClientException;
}
}

1
src/checkstyle/checkstyle.xml
View File

@@ -79,6 +79,7 @@
value="org.assertj.core.api.Assertions.*,
org.awaitility.Awaitility.*,
org.junit.jupiter.api.Assertions.*,
org.junit.jupiter.params.provider.Arguments.*,
org.junit.Assert.*,
org.junit.Assume.*,
org.junit.internal.matchers.ThrowableMessageMatcher.*,