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

Add ITs for custom schema mappings

Browse Source 
See #269
This commit is contained in:
Chris Bono
2023-01-18 21:44:40 -06:00
committed by Soby Chacko
parent cc6076ec6b
commit e37049b9dd
4 changed files with 237 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -47,10 +47,12 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.core.CachingPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.PulsarAdministration;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -102,7 +104,9 @@ class PulsarAutoConfigurationTests {
.hasSingleBean(PulsarTemplate.class).hasSingleBean(PulsarConsumerFactory.class)
.hasSingleBean(ConcurrentPulsarListenerContainerFactory.class)
.hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class)
.hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class));
.hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class)
.hasSingleBean(SchemaResolver.class).getBean(SchemaResolver.class)
.isInstanceOf(DefaultSchemaResolver.class));
}
@Test
@@ -142,6 +146,20 @@ class PulsarAutoConfigurationTests {
.run((context) -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class).isSameAs(template));
}
@Test
@SuppressWarnings("rawtypes")
void beansAreInjectedInPulsarTemplate() {
PulsarProducerFactory<?> producerFactory = mock(PulsarProducerFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner.withBean("customPulsarProducerFactory", PulsarProducerFactory.class, () -> producerFactory)
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver).run((context -> {
AbstractObjectAssert<? extends AbstractObjectAssert<?, PulsarTemplate>, PulsarTemplate> template = assertThat(
context).hasNotFailed().getBean(PulsarTemplate.class);
template.extracting("producerFactory").isSameAs(producerFactory);
template.extracting("schemaResolver").isSameAs(schemaResolver);
}));
}
@Test
void customPulsarConsumerFactoryIsRespected() {
PulsarConsumerFactory<String> consumerFactory = mock(PulsarConsumerFactory.class);
@@ -168,6 +186,21 @@ class PulsarAutoConfigurationTests {
.isSameAs(listenerContainerFactory));
}
@Test
@SuppressWarnings("rawtypes")
void beansAreInjectedInPulsarListenerContainerFactory() {
PulsarConsumerFactory<?> consumerFactory = mock(PulsarConsumerFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner.withBean("customPulsarConsumerFactory", PulsarConsumerFactory.class, () -> consumerFactory)
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver).run((context -> {
AbstractObjectAssert<? extends AbstractObjectAssert<?, ConcurrentPulsarListenerContainerFactory>, ConcurrentPulsarListenerContainerFactory> containerFactory = assertThat(
context).hasNotFailed().getBean(ConcurrentPulsarListenerContainerFactory.class);
containerFactory.extracting("consumerFactory").isSameAs(consumerFactory);
containerFactory.extracting(ConcurrentPulsarListenerContainerFactory::getContainerProperties)
.extracting(PulsarContainerProperties::getSchemaResolver).isSameAs(schemaResolver);
}));
}
@Test
void customPulsarListenerAnnotationBeanPostProcessorIsRespected() {
PulsarListenerAnnotationBeanPostProcessor<String> listenerAnnotationBeanPostProcessor = mock(

98
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -18,19 +18,25 @@ package org.springframework.pulsar.autoconfigure;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.schema.SchemaType;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Bean;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.SchemaResolver;
/**
* Tests for {@link PulsarListener}.
@@ -42,7 +48,11 @@ class PulsarListenerTests implements PulsarTestContainerSupport {
private static final CountDownLatch LATCH_1 = new CountDownLatch(1);
private static final CountDownLatch LATCH_2 = new CountDownLatch(10);
private static final CountDownLatch LATCH_2 = new CountDownLatch(1);
private static final CountDownLatch LATCH_3 = new CountDownLatch(1);
private static final CountDownLatch LATCH_4 = new CountDownLatch(10);
@Test
void basicPulsarListener() throws Exception {
@@ -58,6 +68,35 @@ class PulsarListenerTests implements PulsarTestContainerSupport {
}
}
@Test
void basicPulsarListenerCustomType() throws Exception {
SpringApplication app = new SpringApplication(BasicListenerCustomTypeConfig.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app
.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
PulsarTemplate<Foo> pulsarTemplate = context.getBean(PulsarTemplate.class);
pulsarTemplate.setSchema(Schema.JSON(Foo.class));
pulsarTemplate.send("plt-custom-topic1", new Foo("John Doe"));
assertThat(LATCH_2.await(20, TimeUnit.SECONDS)).isTrue();
}
}
@Test
void basicPulsarListenerCustomTypeWithTypeMapping() throws Exception {
SpringApplication app = new SpringApplication(BasicListenerCustomTypeWithTypeMappingConfig.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app
.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
PulsarTemplate<Foo> pulsarTemplate = context.getBean(PulsarTemplate.class);
pulsarTemplate.send("plt-custom-topic2", new Foo("John Doe"));
assertThat(LATCH_3.await(20, TimeUnit.SECONDS)).isTrue();
}
}
@Test
void batchPulsarListener() throws Exception {
SpringApplication app = new SpringApplication(BatchListenerConfig.class);
@@ -70,30 +109,61 @@ class PulsarListenerTests implements PulsarTestContainerSupport {
for (int i = 0; i < 10; i++) {
pulsarTemplate.send("plt-topic2", "John Doe");
}
assertThat(LATCH_2.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(LATCH_4.await(10, TimeUnit.SECONDS)).isTrue();
}
}
@Configuration(proxyBeanMethods = false)
@Import(PulsarAutoConfiguration.class)
@EnableAutoConfiguration
@SpringBootConfiguration
static class BasicListenerConfig {
@PulsarListener(subscriptionName = "plt-subscription1", topics = "plt-topic1")
@PulsarListener(subscriptionName = "plt-sub1", topics = "plt-topic1")
public void listen(String foo) {
LATCH_1.countDown();
}
}
@Configuration(proxyBeanMethods = false)
@Import(PulsarAutoConfiguration.class)
static class BatchListenerConfig {
@EnableAutoConfiguration
@SpringBootConfiguration
static class BasicListenerCustomTypeConfig {
@PulsarListener(subscriptionName = "plt-subscription2", topics = "plt-topic2", batch = true)
public void listen(List<String> foo) {
foo.forEach(t -> LATCH_2.countDown());
@PulsarListener(subscriptionName = "plt-custom-sub1", topics = "plt-custom-topic1",
schemaType = SchemaType.JSON)
public void listen(Foo foo) {
LATCH_2.countDown();
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
static class BasicListenerCustomTypeWithTypeMappingConfig {
@Bean
SchemaResolver customSchemaResolver() {
return new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.JSON(Foo.class)));
}
@PulsarListener(subscriptionName = "plt-custom-sub2", topics = "plt-custom-topic2")
public void listen(Foo foo) {
LATCH_3.countDown();
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
static class BatchListenerConfig {
@PulsarListener(subscriptionName = "plt-batch-sub", topics = "plt-topic2", batch = true)
public void listen(List<String> foo) {
foo.forEach(t -> LATCH_4.countDown());
}
}
record Foo(String value) {
}
}

32
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -49,6 +49,7 @@ import org.springframework.boot.test.context.FilteredClassLoader;
import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.pulsar.config.PulsarClientFactoryBean;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory;
import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory;
import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry;
@@ -125,6 +126,23 @@ class PulsarReactiveAutoConfigurationTests {
.run((context) -> assertThat(context).hasNotFailed().getBean(beanClass).isSameAs(bean));
}
@SuppressWarnings("rawtypes")
@Test
void beansAreInjectedInReactivePulsarListenerContainerFactory() {
ReactivePulsarConsumerFactory<?> consumerFactory = mock(ReactivePulsarConsumerFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner
.withBean("customReactivePulsarConsumerFactory", ReactivePulsarConsumerFactory.class,
() -> consumerFactory)
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver).run((context -> {
AbstractObjectAssert<? extends AbstractObjectAssert<?, DefaultReactivePulsarListenerContainerFactory>, DefaultReactivePulsarListenerContainerFactory> containerFactory = assertThat(
context).hasNotFailed().getBean(DefaultReactivePulsarListenerContainerFactory.class);
containerFactory.extracting("consumerFactory").isSameAs(consumerFactory);
containerFactory.extracting(DefaultReactivePulsarListenerContainerFactory::getContainerProperties)
.extracting(ReactivePulsarContainerProperties::getSchemaResolver).isSameAs(schemaResolver);
}));
}
@Test
void customReactivePulsarListenerContainerFactoryIsRespected() {
ReactivePulsarListenerContainerFactory<String> listenerContainerFactory = mock(
@@ -150,14 +168,18 @@ class PulsarReactiveAutoConfigurationTests {
}
@Test
@SuppressWarnings("rawtypes")
void beansAreInjectedInReactivePulsarTemplate() {
ReactivePulsarSenderFactory<?> senderFactory = mock(ReactivePulsarSenderFactory.class);
SchemaResolver schemaResolver = mock(SchemaResolver.class);
this.contextRunner
.withBean("customReactivePulsarSenderFactory", ReactivePulsarSenderFactory.class, () -> senderFactory)
.run((context -> assertThat(context).hasNotFailed().getBean(ReactivePulsarTemplate.class)
.extracting("reactiveMessageSenderFactory")
.asInstanceOf(InstanceOfAssertFactories.type(ReactivePulsarSenderFactory.class))
.isSameAs(senderFactory)));
.withBean("schemaResolver", SchemaResolver.class, () -> schemaResolver).run((context -> {
AbstractObjectAssert<? extends AbstractObjectAssert<?, ReactivePulsarTemplate>, ReactivePulsarTemplate> template = assertThat(
context).hasNotFailed().getBean(ReactivePulsarTemplate.class);
template.extracting("reactiveMessageSenderFactory").isSameAs(senderFactory);
template.extracting("schemaResolver").isSameAs(schemaResolver);
}));
}
@Test

105
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-2023 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.
@@ -18,23 +18,31 @@ package org.springframework.pulsar.autoconfigure;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.reactive.client.api.MessageResult;
import org.junit.jupiter.api.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -49,7 +57,11 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
private static final CountDownLatch LATCH1 = new CountDownLatch(1);
private static final CountDownLatch LATCH2 = new CountDownLatch(10);
private static final CountDownLatch LATCH2 = new CountDownLatch(1);
private static final CountDownLatch LATCH3 = new CountDownLatch(1);
private static final CountDownLatch LATCH4 = new CountDownLatch(10);
@Test
void basicListener() throws Exception {
@@ -59,12 +71,41 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
try (ConfigurableApplicationContext context = app
.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
PulsarTemplate<String> pulsarTemplate = context.getBean(PulsarTemplate.class);
pulsarTemplate.send("rplt-topic1", "John Doe");
ReactivePulsarTemplate<String> pulsarTemplate = context.getBean(ReactivePulsarTemplate.class);
pulsarTemplate.send("rplt-topic1", "John Doe").block();
assertThat(LATCH1.await(20, TimeUnit.SECONDS)).isTrue();
}
}
@Test
void basicListenerCustomType() throws Exception {
SpringApplication app = new SpringApplication(BasicListenerCustomTypeConfig.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app
.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
ReactivePulsarTemplate<Foo> pulsarTemplate = context.getBean(ReactivePulsarTemplate.class);
pulsarTemplate.setSchema(Schema.JSON(Foo.class));
pulsarTemplate.send("rplt-custom-topic1", new Foo("John Doe")).block();
assertThat(LATCH2.await(20, TimeUnit.SECONDS)).isTrue();
}
}
@Test
void basicListenerCustomTypeWithTypeMapping() throws Exception {
SpringApplication app = new SpringApplication(BasicListenerCustomTypeWithTypeMappingConfig.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext context = app
.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl())) {
@SuppressWarnings("unchecked")
ReactivePulsarTemplate<Foo> pulsarTemplate = context.getBean(ReactivePulsarTemplate.class);
pulsarTemplate.send("rplt-custom-topic2", new Foo("John Doe")).block();
assertThat(LATCH3.await(20, TimeUnit.SECONDS)).isTrue();
}
}
@Test
void fluxListener() throws Exception {
SpringApplication app = new SpringApplication(FluxListenerConfig.class);
@@ -75,17 +116,18 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
@SuppressWarnings("unchecked")
PulsarTemplate<String> pulsarTemplate = context.getBean(PulsarTemplate.class);
for (int i = 0; i < 10; i++) {
pulsarTemplate.send("rplt-topic2", "John Doe");
pulsarTemplate.send("rplt-batch-topic", "John Doe");
}
assertThat(LATCH2.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(LATCH4.await(10, TimeUnit.SECONDS)).isTrue();
}
}
@Configuration(proxyBeanMethods = false)
@Import({ PulsarAutoConfiguration.class, PulsarReactiveAutoConfiguration.class, ConsumerCustomizerConfig.class })
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(ConsumerCustomizerConfig.class)
static class BasicListenerConfig {
@ReactivePulsarListener(subscriptionName = "rplt-subscription1", topics = "rplt-topic1",
@ReactivePulsarListener(subscriptionName = "rplt-sub1", topics = "rplt-topic1",
consumerCustomizer = "consumerCustomizer")
public Mono<Void> listen(String foo) {
LATCH1.countDown();
@@ -94,14 +136,48 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
}
@Configuration(proxyBeanMethods = false)
@Import({ PulsarAutoConfiguration.class, PulsarReactiveAutoConfiguration.class, ConsumerCustomizerConfig.class })
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(ConsumerCustomizerConfig.class)
static class BasicListenerCustomTypeConfig {
@ReactivePulsarListener(subscriptionName = "rplt-custom-sub1", topics = "rplt-custom-topic1",
schemaType = SchemaType.JSON, consumerCustomizer = "consumerCustomizer")
public Mono<Void> listen(Foo foo) {
LATCH2.countDown();
return Mono.empty();
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(ConsumerCustomizerConfig.class)
static class BasicListenerCustomTypeWithTypeMappingConfig {
@Bean
SchemaResolver customSchemaResolver() {
return new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.JSON(Foo.class)));
}
@ReactivePulsarListener(subscriptionName = "rplt-custom-sub2", topics = "rplt-custom-topic2",
consumerCustomizer = "consumerCustomizer")
public Mono<Void> listen(Foo foo) {
LATCH3.countDown();
return Mono.empty();
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(ConsumerCustomizerConfig.class)
static class FluxListenerConfig {
@ReactivePulsarListener(subscriptionName = "rplt-subscription2", topics = "rplt-topic2", stream = true,
@ReactivePulsarListener(subscriptionName = "rplt-batch-sub", topics = "rplt-batch-topic", stream = true,
consumerCustomizer = "consumerCustomizer")
public Flux<MessageResult<Void>> listen(Flux<Message<String>> messages) {
return messages.doOnNext(t -> LATCH2.countDown()).map(MessageResult::acknowledge);
return messages.doOnNext(t -> LATCH4.countDown()).map(MessageResult::acknowledge);
}
}
@@ -116,4 +192,7 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
}
record Foo(String value) {
}
}