From e37049b9dd65141b4d63d68b2f8523a1031f5665 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Wed, 18 Jan 2023 21:44:40 -0600 Subject: [PATCH] Add ITs for custom schema mappings See #269 --- .../PulsarAutoConfigurationTests.java | 35 +++++- .../autoconfigure/PulsarListenerTests.java | 98 +++++++++++++--- .../PulsarReactiveAutoConfigurationTests.java | 32 +++++- .../ReactivePulsarListenerTests.java | 105 +++++++++++++++--- 4 files changed, 237 insertions(+), 33 deletions(-) diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index bb0d90a1..3e14ce60 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -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, PulsarTemplate> template = assertThat( + context).hasNotFailed().getBean(PulsarTemplate.class); + template.extracting("producerFactory").isSameAs(producerFactory); + template.extracting("schemaResolver").isSameAs(schemaResolver); + })); + } + @Test void customPulsarConsumerFactoryIsRespected() { PulsarConsumerFactory 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, 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 listenerAnnotationBeanPostProcessor = mock( diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java index 309157b8..eb02963f 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java @@ -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 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 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 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 foo) { + foo.forEach(t -> LATCH_4.countDown()); + } + + } + + record Foo(String value) { + } + } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java index 920d46f0..2cdde9c6 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfigurationTests.java @@ -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, 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 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, ReactivePulsarTemplate> template = assertThat( + context).hasNotFailed().getBean(ReactivePulsarTemplate.class); + template.extracting("reactiveMessageSenderFactory").isSameAs(senderFactory); + template.extracting("schemaResolver").isSameAs(schemaResolver); + })); } @Test diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java index 4969e204..f864c4b7 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java @@ -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 pulsarTemplate = context.getBean(PulsarTemplate.class); - pulsarTemplate.send("rplt-topic1", "John Doe"); + ReactivePulsarTemplate 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 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 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 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 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 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 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> listen(Flux> 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) { + } + }