diff --git a/settings.gradle b/settings.gradle index e6f61552..d1c1165e 100644 --- a/settings.gradle +++ b/settings.gradle @@ -24,6 +24,7 @@ settings.gradle.projectsLoaded { rootProject.name = 'spring-pulsar-dist' include 'spring-pulsar' +include 'spring-pulsar-reactive' include 'spring-pulsar-dependencies' include 'spring-pulsar-spring-boot-autoconfigure' include 'spring-pulsar-spring-boot-starter' diff --git a/spring-pulsar-reactive-spring-boot-starter/build.gradle b/spring-pulsar-reactive-spring-boot-starter/build.gradle index 28ee6334..859c7c75 100644 --- a/spring-pulsar-reactive-spring-boot-starter/build.gradle +++ b/spring-pulsar-reactive-spring-boot-starter/build.gradle @@ -6,8 +6,8 @@ description = 'Spring Pulsar Reactive Spring Boot Starter' dependencies { api project (':spring-pulsar') + api project (':spring-pulsar-reactive') api project (':spring-pulsar-spring-boot-autoconfigure') - api 'org.apache.pulsar:pulsar-client-reactive-adapter' api 'org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine' api 'org.springframework.boot:spring-boot-starter' } diff --git a/spring-pulsar-reactive/build.gradle b/spring-pulsar-reactive/build.gradle new file mode 100644 index 00000000..d0408e1a --- /dev/null +++ b/spring-pulsar-reactive/build.gradle @@ -0,0 +1,34 @@ +plugins { + id 'org.springframework.pulsar.spring-module' +} + +description = 'Spring Pulsar Reactive Support' + +dependencies { + api project (':spring-pulsar') + api 'org.apache.pulsar:pulsar-client-reactive-adapter' + implementation 'com.fasterxml.jackson.core:jackson-core' + implementation 'com.fasterxml.jackson.core:jackson-databind' + implementation 'com.google.code.findbugs:jsr305' + optional 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8' + optional 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310' + optional 'com.fasterxml.jackson.datatype:jackson-datatype-joda' + optional 'com.jayway.jsonpath:json-path' + optional 'io.projectreactor:reactor-core' + + testImplementation 'org.junit.jupiter:junit-jupiter' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' + testRuntimeOnly 'ch.qos.logback:logback-classic' + testImplementation 'io.projectreactor:reactor-test' + testImplementation 'org.assertj:assertj-core' + testImplementation 'org.awaitility:awaitility' + testImplementation 'org.hamcrest:hamcrest' + testImplementation 'org.mockito:mockito-junit-jupiter' + testImplementation 'org.springframework:spring-test' + testImplementation 'org.testcontainers:junit-jupiter' + testImplementation 'org.testcontainers:pulsar' +} + +test { + testLogging.showStandardStreams = true +} diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/aot/ReactivePulsarRuntimeHints.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/aot/ReactivePulsarRuntimeHints.java new file mode 100644 index 00000000..d8ab9b83 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/aot/ReactivePulsarRuntimeHints.java @@ -0,0 +1,111 @@ +/* + * Copyright 2022 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.reactive.aot; + +import java.util.HashSet; +import java.util.TreeMap; +import java.util.stream.Stream; + +import org.apache.pulsar.client.admin.internal.OffloadProcessStatusImpl; +import org.apache.pulsar.client.admin.internal.PulsarAdminBuilderImpl; +import org.apache.pulsar.client.api.Authentication; +import org.apache.pulsar.client.api.AuthenticationDataProvider; +import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; +import org.apache.pulsar.client.util.SecretsSerializer; +import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.shade.io.netty.buffer.AbstractByteBufAllocator; +import org.apache.pulsar.shade.io.netty.channel.socket.nio.NioDatagramChannel; +import org.apache.pulsar.shade.io.netty.channel.socket.nio.NioSocketChannel; +import org.apache.pulsar.shade.io.netty.util.ReferenceCountUtil; + +import org.springframework.aot.hint.MemberCategory; +import org.springframework.aot.hint.ReflectionHints; +import org.springframework.aot.hint.RuntimeHints; +import org.springframework.aot.hint.RuntimeHintsRegistrar; +import org.springframework.aot.hint.TypeReference; +import org.springframework.lang.Nullable; + +/** + * {@link RuntimeHintsRegistrar} for Spring for Apache Pulsar. + * + * @author Soby Chacko + */ +public class ReactivePulsarRuntimeHints implements RuntimeHintsRegistrar { + + @Override + public void registerHints(RuntimeHints hints, @Nullable ClassLoader classLoader) { + ReflectionHints reflectionHints = hints.reflection(); + // The following components need access to declared constructors, invoke declared + // methods + // and introspect all public methods. The components are a mix of JDK classes, + // core Pulsar classes, + // some other shaded components available through Pulsar client. + Stream.of(HashSet.class, TreeMap.class, Authentication.class, AuthenticationDataProvider.class, + SecretsSerializer.class, NioSocketChannel.class, AbstractByteBufAllocator.class, + NioDatagramChannel.class, PulsarAdminBuilderImpl.class, OffloadProcessStatusImpl.class, Commands.class, + ReferenceCountUtil.class).forEach( + type -> reflectionHints.registerType(type, + builder -> builder.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS, + MemberCategory.INVOKE_DECLARED_METHODS, + MemberCategory.INTROSPECT_PUBLIC_METHODS))); + + // In addition to the above member category levels, these components need field + // and declared class level access. + Stream.of(ClientConfigurationData.class, ConsumerConfigurationData.class, ProducerConfigurationData.class) + .forEach(type -> reflectionHints.registerType(type, + builder -> builder.withMembers(MemberCategory.INVOKE_DECLARED_CONSTRUCTORS, + MemberCategory.INVOKE_DECLARED_METHODS, MemberCategory.INTROSPECT_PUBLIC_METHODS, + MemberCategory.DECLARED_CLASSES, MemberCategory.DECLARED_FIELDS))); + + // These are inaccessible interfaces/classes in a normal scenario, thus using the + // String version, + // and we need field level access in them. + Stream.of( + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields", + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields", + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields", + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField", + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField", + "org.apache.pulsar.shade.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField") + .forEach(typeName -> reflectionHints.registerTypeIfPresent(classLoader, typeName, + MemberCategory.DECLARED_FIELDS)); + + Stream.of("reactor.core.publisher.Flux", "com.github.benmanes.caffeine.cache.SSMSA", + "com.github.benmanes.caffeine.cache.PSAMS") + .forEach(typeName -> reflectionHints.registerTypeIfPresent(classLoader, typeName, + MemberCategory.INVOKE_DECLARED_CONSTRUCTORS, MemberCategory.INVOKE_DECLARED_METHODS, + MemberCategory.INTROSPECT_PUBLIC_METHODS)); + + // Registering JDK dynamic proxies for these interfaces. Since the Connection + // interface is protected, + // wee need to use the string version of proxy registration. Although the other + // interfaces are public, + // due to ConnectionHandler$Connection being protected forces all of them to be + // registered using the + // string version of the API because all of them need to be registered through a + // single call. + hints.proxies().registerJdkProxy(TypeReference.of("org.apache.pulsar.shade.io.netty.util.TimerTask"), + TypeReference.of("org.apache.pulsar.client.impl.ConnectionHandler$Connection"), + TypeReference.of("org.apache.pulsar.client.api.Producer"), + TypeReference.of("org.springframework.aop.SpringProxy"), + TypeReference.of("org.springframework.aop.framework.Advised"), + TypeReference.of("org.springframework.core.DecoratingProxy")); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/AbstractReactivePulsarListenerEndpoint.java similarity index 97% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/AbstractReactivePulsarListenerEndpoint.java index bd6384fd..a56783a4 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/AbstractReactivePulsarListenerEndpoint.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import java.util.ArrayList; import java.util.Arrays; @@ -35,8 +35,8 @@ import org.springframework.context.expression.BeanFactoryResolver; import org.springframework.expression.BeanResolver; import org.springframework.lang.Nullable; import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageHandler; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; import org.springframework.pulsar.support.MessageConverter; import org.springframework.util.Assert; import org.springframework.util.StringUtils; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java index 2972e6a4..226705b6 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import java.util.Arrays; import java.util.List; @@ -22,9 +22,9 @@ import java.util.List; import org.apache.pulsar.client.api.Schema; import org.springframework.core.log.LogAccessor; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.listener.reactive.DefaultReactivePulsarMessageListenerContainer; -import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.listener.DefaultReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties; import org.springframework.pulsar.support.JavaUtils; import org.springframework.pulsar.support.MessageConverter; import org.springframework.util.CollectionUtils; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java index ae62fe2a..927d19b7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import java.lang.reflect.Method; import java.util.Arrays; @@ -44,15 +44,15 @@ import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.pulsar.core.SchemaUtils; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.adapter.HandlerAdapter; import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; -import org.springframework.pulsar.listener.adapter.PulsarReactiveOneByOneMessagingMessageListenerAdapter; -import org.springframework.pulsar.listener.adapter.PulsarReactiveStreamingMessagingMessageListenerAdapter; -import org.springframework.pulsar.listener.reactive.DefaultReactivePulsarMessageListenerContainer; -import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.listener.DefaultReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.adapter.PulsarReactiveOneByOneMessagingMessageListenerAdapter; +import org.springframework.pulsar.reactive.listener.adapter.PulsarReactiveStreamingMessagingMessageListenerAdapter; import org.springframework.pulsar.support.MessageConverter; import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; import org.springframework.util.Assert; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerContainerFactory.java similarity index 89% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerContainerFactory.java index cfec5a99..415bb867 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerContainerFactory.java @@ -14,10 +14,10 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import org.springframework.pulsar.config.ListenerContainerFactory; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; /** * Factory for Pulsar reactive message listener containers. diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpoint.java similarity index 76% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpoint.java index d0b751c6..5c1b9f80 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpoint.java @@ -14,19 +14,21 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import org.springframework.lang.Nullable; import org.springframework.pulsar.config.ListenerEndpoint; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerConfigurationSelector; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; /** * Model for a Pulsar reactive listener endpoint. Can be used against a - * {@link org.springframework.pulsar.annotation.PulsarListenerConfigurer} to register - * endpoints programmatically. + * {@link ReactivePulsarListenerConfigurationSelector} to register endpoints + * programmatically. * * @param Message payload type. * @author Christophe Bornet + * @author Chris Bono */ public interface ReactivePulsarListenerEndpoint extends ListenerEndpoint> { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointAdapter.java similarity index 94% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointAdapter.java index 7b5837b9..f8f8bca8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointAdapter.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import java.util.Collections; import java.util.List; @@ -23,7 +23,7 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.lang.Nullable; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; import org.springframework.pulsar.support.MessageConverter; /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointRegistry.java similarity index 94% rename from spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointRegistry.java index 70a7f0ee..2db09a14 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointRegistry.java @@ -14,10 +14,10 @@ * limitations under the License. */ -package org.springframework.pulsar.config.reactive; +package org.springframework.pulsar.reactive.config; import org.springframework.pulsar.config.ListenerEndpointRegistry; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageListenerContainer; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListenerContainer; /** * Creates the necessary {@link ReactivePulsarMessageListenerContainer} instances for the diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/EnableReactivePulsar.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/EnableReactivePulsar.java new file mode 100644 index 00000000..2bfce186 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/EnableReactivePulsar.java @@ -0,0 +1,39 @@ +/* + * Copyright 2022 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.reactive.config.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.springframework.context.annotation.Import; + +/** + * Enables detection of {@link ReactivePulsarListener} annotations on any Spring-managed + * bean in the container. + * + * @author Chris Bono + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@Documented +@Import(ReactivePulsarListenerConfigurationSelector.class) +public @interface EnableReactivePulsar { + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarBootstrapConfiguration.java similarity index 92% rename from spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarBootstrapConfiguration.java index f109f7d8..c06cc414 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarBootstrapConfiguration.java @@ -14,14 +14,14 @@ * limitations under the License. */ -package org.springframework.pulsar.annotation; +package org.springframework.pulsar.reactive.config.annotation; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; import org.springframework.core.type.AnnotationMetadata; import org.springframework.pulsar.config.PulsarListenerBeanNames; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; /** * An {@link ImportBeanDefinitionRegistrar} class that registers a @@ -30,13 +30,13 @@ import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpoint * {@link ReactivePulsarListenerEndpointRegistry}. * *

- * This configuration class is automatically imported when using the @{@link EnablePulsar} - * annotation. + * This configuration class is automatically imported when using + * the @{@link EnableReactivePulsar} annotation. * * @author Christophe Bornet * @see ReactivePulsarListenerAnnotationBeanPostProcessor * @see ReactivePulsarListenerEndpointRegistry - * @see EnablePulsar + * @see EnableReactivePulsar */ public class ReactivePulsarBootstrapConfiguration implements ImportBeanDefinitionRegistrar { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListener.java similarity index 91% rename from spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListener.java index 06bfa52f..27eeb3ba 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListener.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.annotation; +package org.springframework.pulsar.reactive.config.annotation; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; @@ -26,8 +26,9 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.messaging.handler.annotation.MessageMapping; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; /** * Annotation that marks a method to be the target of a Pulsar message listener on the @@ -42,7 +43,7 @@ import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpoint *

* Processing of {@code @ReactivePulsarListener} annotations is performed by registering a * {@link ReactivePulsarListenerAnnotationBeanPostProcessor}. This can be done manually - * or, more conveniently, through {@link EnablePulsar} annotation. + * or, more conveniently, through {@link EnableReactivePulsar} annotation. *

* * @author Christophe Bornet @@ -58,7 +59,7 @@ public @interface ReactivePulsarListener { *

* If none is specified an auto-generated id is used. *

- * SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * SpEL {@code #{...}} and property placeholders {@code ${...}} are supported. * @return the {@code id} for the container managing for this endpoint. * @see ReactivePulsarListenerEndpointRegistry#getListenerContainer(String) */ @@ -164,8 +165,7 @@ public @interface ReactivePulsarListener { /** * The bean name or a 'SpEL' expression that resolves to a - * {@link org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer} - * to use to configure the consumer. + * {@link ReactiveMessageConsumerBuilderCustomizer} to use to configure the consumer. * @return the bean name or empty string to not configure the consumer. */ String consumerCustomizer() default ""; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java similarity index 98% rename from spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java index 4f28a903..2adcf728 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.annotation; +package org.springframework.pulsar.reactive.config.annotation; import java.io.IOException; import java.io.StringReader; @@ -76,13 +76,14 @@ import org.springframework.messaging.handler.annotation.support.DefaultMessageHa import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.pulsar.annotation.PulsarListenerConfigurer; import org.springframework.pulsar.config.PulsarListenerBeanNames; import org.springframework.pulsar.config.PulsarListenerEndpointRegistrar; -import org.springframework.pulsar.config.reactive.MethodReactivePulsarListenerEndpoint; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpoint; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.config.MethodReactivePulsarListenerEndpoint; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpoint; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; @@ -99,18 +100,19 @@ import org.springframework.validation.Validator; * {@link ReactivePulsarListener}. * *

- * This post-processor is automatically registered by the {@link EnablePulsar} annotation. + * This post-processor is automatically registered by the {@link EnableReactivePulsar} + * annotation. * *

* Auto-detect any {@link PulsarListenerConfigurer} instances in the container, allowing * for customization of the registry to be used, the default container factory or for - * fine-grained control over endpoints registration. See {@link EnablePulsar} Javadoc for - * complete usage details. + * fine-grained control over endpoints registration. See {@link EnableReactivePulsar} + * Javadoc for complete usage details. * * @param the payload type. * @author Christophe Bornet * @see ReactivePulsarListener - * @see EnablePulsar + * @see EnableReactivePulsar * @see PulsarListenerConfigurer * @see PulsarListenerEndpointRegistrar * @see ReactivePulsarListenerEndpointRegistry diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerConfigurationSelector.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerConfigurationSelector.java new file mode 100644 index 00000000..93200b3c --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerConfigurationSelector.java @@ -0,0 +1,37 @@ +/* + * Copyright 2022 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.reactive.config.annotation; + +import org.springframework.context.annotation.DeferredImportSelector; +import org.springframework.core.annotation.Order; +import org.springframework.core.type.AnnotationMetadata; + +/** + * A {@link DeferredImportSelector} implementation with the lowest order to import + * {@link ReactivePulsarBootstrapConfiguration} as late as possible. + * + * @author Chris Bono + */ +@Order +public class ReactivePulsarListenerConfigurationSelector implements DeferredImportSelector { + + @Override + public String[] selectImports(AnnotationMetadata importingClassMetadata) { + return new String[] { ReactivePulsarBootstrapConfiguration.class.getName() }; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListeners.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListeners.java index 0afdef60..c5c6a755 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListeners.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.annotation; +package org.springframework.pulsar.reactive.config.annotation; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/package-info.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/package-info.java new file mode 100644 index 00000000..06c28176 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing annotations used by the framework. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.reactive.config.annotation; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/package-info.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/package-info.java new file mode 100644 index 00000000..66a4c52a --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing Spring configuration classes for the framework. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.reactive.config; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java similarity index 98% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java index fe57d466..cd1e85c3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.Collections; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarReaderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java similarity index 97% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarReaderFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java index 5575d6d1..12d1b43d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarReaderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.Collections; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java similarity index 98% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java index 9004221f..afc6e069 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/MessageSpecBuilderCustomizer.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/MessageSpecBuilderCustomizer.java index f62394c5..36aa7ba8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/MessageSpecBuilderCustomizer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import org.apache.pulsar.reactive.client.api.MessageSpecBuilder; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java index c83f0b22..5fcb6f16 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import org.apache.pulsar.reactive.client.api.ReactiveMessageConsumerBuilder; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageReaderBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageReaderBuilderCustomizer.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageReaderBuilderCustomizer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageReaderBuilderCustomizer.java index 292b05ce..10da508a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageReaderBuilderCustomizer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageReaderBuilderCustomizer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import org.apache.pulsar.reactive.client.api.ReactiveMessageReaderBuilder; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderBuilderCustomizer.java similarity index 95% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderBuilderCustomizer.java index b95e9749..f562e89c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderBuilderCustomizer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderUtils.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderUtils.java index be9857b7..49af1609 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderUtils.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.Optional; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarConsumerFactory.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarConsumerFactory.java index fc7870af..c31e48ce 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarConsumerFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarOperations.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java similarity index 98% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarOperations.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java index 145a60d9..8e1b0687 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarOperations.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import org.apache.pulsar.client.api.MessageId; import org.reactivestreams.Publisher; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarReaderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarReaderFactory.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarReaderFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarReaderFactory.java index 0e2b362c..983dd86a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarReaderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarReaderFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarSenderFactory.java similarity index 97% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarSenderFactory.java index c11ce923..8f41f294 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarSenderFactory.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.List; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplate.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java similarity index 99% rename from spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplate.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java index da622040..f46c6514 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplate.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import java.util.Collections; diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/package-info.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/package-info.java new file mode 100644 index 00000000..8104a519 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing the core reactive components of the framework. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.reactive.core; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainer.java similarity index 97% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainer.java index 4dc2d272..61466b76 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainer.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import java.util.ArrayList; import java.util.List; @@ -28,8 +28,8 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessagePipelineBuilder.Conc import org.apache.pulsar.reactive.client.internal.api.ApiImplementationFactory; import org.springframework.core.log.LogAccessor; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; import org.springframework.util.CollectionUtils; /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java similarity index 98% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java index e815a0c6..1bd7f599 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import java.time.Duration; import java.util.Collection; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageHandler.java similarity index 94% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageHandler.java index 94c46db7..3efeb387 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageHandler.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; /** * Reactive message handler used by {@link DefaultReactivePulsarMessageListenerContainer}. diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageListenerContainer.java similarity index 92% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageListenerContainer.java index 48bbc469..cb3ff1a9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageListenerContainer.java @@ -14,10 +14,10 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.listener.MessageListenerContainer; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; /** * Internal abstraction used by the framework representing a reactive message listener diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarOneByOneMessageHandler.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarOneByOneMessageHandler.java index 92f50f72..d9834271 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarOneByOneMessageHandler.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.reactive.client.api.ReactiveMessagePipelineBuilder; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarStreamingHandler.java similarity index 96% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarStreamingHandler.java index b070cecb..8a55c1ad 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarStreamingHandler.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.reactive.client.api.MessageResult; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java similarity index 85% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java index ceddcc54..789c9d97 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java @@ -14,15 +14,17 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.adapter; +package org.springframework.pulsar.reactive.listener.adapter; import java.lang.reflect.Method; import org.apache.pulsar.client.api.Message; import org.reactivestreams.Publisher; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; -import org.springframework.pulsar.listener.reactive.ReactivePulsarOneByOneMessageHandler; +import org.springframework.pulsar.listener.adapter.HandlerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageHandler; +import org.springframework.pulsar.reactive.listener.ReactivePulsarOneByOneMessageHandler; import reactor.core.publisher.Mono; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java similarity index 83% rename from spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java rename to spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java index 7cc99acd..6c6ece43 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java @@ -14,15 +14,17 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.adapter; +package org.springframework.pulsar.reactive.listener.adapter; import java.lang.reflect.Method; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.reactive.client.api.MessageResult; -import org.springframework.pulsar.listener.reactive.ReactivePulsarMessageHandler; -import org.springframework.pulsar.listener.reactive.ReactivePulsarStreamingHandler; +import org.springframework.pulsar.listener.adapter.HandlerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; +import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageHandler; +import org.springframework.pulsar.reactive.listener.ReactivePulsarStreamingHandler; import reactor.core.publisher.Flux; diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java new file mode 100644 index 00000000..fb23b2da --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java @@ -0,0 +1,69 @@ +/* + * Copyright 2022 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.reactive.listener.adapter; + +import java.lang.reflect.Method; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageListener; + +import org.springframework.lang.Nullable; +import org.springframework.pulsar.listener.Acknowledgement; +import org.springframework.pulsar.listener.PulsarAcknowledgingMessageListener; +import org.springframework.pulsar.listener.adapter.HandlerAdapter; +import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; + +/** + * A {@link MessageListener MessageListener} adapter that invokes a configurable + * {@link HandlerAdapter}; used when the factory is configured for the listener to receive + * individual messages. + * + * @param payload type. + * @author Soby Chacko + */ +@SuppressWarnings("serial") +public class PulsarRecordMessagingMessageListenerAdapter extends PulsarMessagingMessageListenerAdapter + implements PulsarAcknowledgingMessageListener { + + public PulsarRecordMessagingMessageListenerAdapter(Object bean, Method method) { + super(bean, method); + } + + @Override + public void received(Consumer consumer, Message record, @Nullable Acknowledgement acknowledgement) { + org.springframework.messaging.Message message = null; + Object theRecord = record; + if (isHeaderFound() || isSpringMessage()) { + message = toMessagingMessage(record, consumer); + } + else if (isSimpleExtraction()) { + theRecord = record.getValue(); + } + + if (logger.isDebugEnabled()) { + this.logger.debug("Processing [" + message + "]"); + } + try { + invokeHandler(theRecord, message, consumer, acknowledgement); + } + catch (Exception e) { + throw e; + } + } + +} diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/package-info.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/package-info.java new file mode 100644 index 00000000..e3818503 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing listener components for receiving Pulsar messages. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.reactive.listener.adapter; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/package-info.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/package-info.java new file mode 100644 index 00000000..31206317 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/package-info.java @@ -0,0 +1,9 @@ +/** + * Package containing listener components for receiving Pulsar messages. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.reactive.listener; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar-reactive/src/main/resources/META-INF/spring/aot.factories b/spring-pulsar-reactive/src/main/resources/META-INF/spring/aot.factories new file mode 100644 index 00000000..a67465ea --- /dev/null +++ b/spring-pulsar-reactive/src/main/resources/META-INF/spring/aot.factories @@ -0,0 +1 @@ +org.springframework.aot.hint.RuntimeHintsRegistrar=org.springframework.pulsar.reactive.aot.ReactivePulsarRuntimeHints diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/core/PulsarTestContainerSupport.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/core/PulsarTestContainerSupport.java new file mode 100644 index 00000000..1e77e70f --- /dev/null +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/core/PulsarTestContainerSupport.java @@ -0,0 +1,64 @@ +/* + * Copyright 2022 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.core; + +import java.util.Locale; + +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.PulsarContainer; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +/** + * Provides a static {@link PulsarContainer} that can be shared across test classes. + * + * @author Chris Bono + */ +@Testcontainers(disabledWithoutDocker = true) +public interface PulsarTestContainerSupport { + + PulsarContainer PULSAR_CONTAINER = new PulsarContainer( + isRunningOnMacM1() ? getMacM1PulsarImage() : getStandardPulsarImage()); + + @BeforeAll + static void startContainer() { + PULSAR_CONTAINER.start(); + } + + static String getPulsarBrokerUrl() { + return PULSAR_CONTAINER.getPulsarBrokerUrl(); + } + + static String getHttpServiceUrl() { + return PULSAR_CONTAINER.getHttpServiceUrl(); + } + + private static boolean isRunningOnMacM1() { + String osName = System.getProperty("os.name").toLowerCase(Locale.ENGLISH); + String osArchitecture = System.getProperty("os.arch").toLowerCase(Locale.ENGLISH); + return osName.contains("mac") && osArchitecture.equals("aarch64"); + } + + private static DockerImageName getStandardPulsarImage() { + return DockerImageName.parse("apachepulsar/pulsar:2.10.1"); + } + + private static DockerImageName getMacM1PulsarImage() { + return DockerImageName.parse("kezhenxu94/pulsar").asCompatibleSubstituteFor("apachepulsar/pulsar"); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageConsumerFactoryTests.java similarity index 88% rename from spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageConsumerFactoryTests.java index 89ae3d87..2602002e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageConsumerFactoryTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; @@ -32,7 +32,8 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; /** - * Tests for {@link DefaultReactivePulsarConsumerFactory} + * Tests for + * {@link org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory} * * @author Christophe Bornet * @author Chris Bono @@ -44,7 +45,7 @@ class DefaultReactiveMessageConsumerFactoryTests { @Nested class FactoryCreatedWithoutSpec { - private DefaultReactivePulsarConsumerFactory consumerFactory = new DefaultReactivePulsarConsumerFactory<>( + private org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory consumerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory<>( AdaptedReactivePulsarClientFactory.create((PulsarClient) null), null); @Test @@ -71,7 +72,7 @@ class DefaultReactiveMessageConsumerFactoryTests { @Nested class FactoryCreatedWithSpec { - private DefaultReactivePulsarConsumerFactory consumerFactory; + private org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory consumerFactory; @BeforeEach void createConsumerFactory() { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageReaderFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageReaderFactoryTests.java similarity index 83% rename from spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageReaderFactoryTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageReaderFactoryTests.java index fb1443cf..be6a2462 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageReaderFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageReaderFactoryTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; @@ -30,7 +30,8 @@ import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; /** - * Tests for {@link DefaultReactivePulsarReaderFactory} + * Tests for + * {@link org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory} * * @author Christophe Bornet */ @@ -42,7 +43,7 @@ class DefaultReactiveMessageReaderFactoryTests { void createReader() { MutableReactiveMessageReaderSpec spec = new MutableReactiveMessageReaderSpec(); spec.setReaderName("test-reader"); - DefaultReactivePulsarReaderFactory readerFactory = new DefaultReactivePulsarReaderFactory<>( + org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory readerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory<>( AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec); ReactiveMessageReader reader = readerFactory.createReader(schema); @@ -55,7 +56,7 @@ class DefaultReactiveMessageReaderFactoryTests { void createReaderWithCustomizer() { MutableReactiveMessageReaderSpec spec = new MutableReactiveMessageReaderSpec(); spec.setReaderName("test-reader"); - DefaultReactivePulsarReaderFactory readerFactory = new DefaultReactivePulsarReaderFactory<>( + org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory readerFactory = new DefaultReactivePulsarReaderFactory<>( AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec); ReactiveMessageReader reader = readerFactory.createReader(schema, diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageSenderFactoryTests.java similarity index 85% rename from spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageSenderFactoryTests.java index 670acdfe..2f5e2e4a 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageSenderFactoryTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; @@ -34,7 +34,8 @@ import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; /** - * Tests for {@link DefaultReactivePulsarSenderFactory} + * Tests for + * {@link org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory} * * @author Christophe Bornet */ @@ -62,9 +63,11 @@ class DefaultReactiveMessageSenderFactoryTests { @Test void createSenderWithMultipleSenderCustomizer() { - ReactiveMessageSenderBuilderCustomizer customizer1 = builder -> builder.topic("topic1"); + org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer customizer1 = builder -> builder + .topic("topic1"); ReactiveMessageSenderCache cache = AdaptedReactivePulsarClientFactory.createCache(); - ReactiveMessageSenderBuilderCustomizer customizer2 = builder -> builder.cache(cache); + org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer customizer2 = builder -> builder + .cache(cache); ReactiveMessageSender sender = testCreateSender(null, null, "topic0", Arrays.asList(customizer1, customizer2), "topic1"); @@ -73,7 +76,7 @@ class DefaultReactiveMessageSenderFactoryTests { @Test void createSenderWithNoTopic() { - ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>( + org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory senderFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>( (PulsarClient) null, null, null); assertThatIllegalArgumentException().isThrownBy(() -> senderFactory.createSender(null, schema)) .withMessageContaining("Topic must be specified when no default topic is configured"); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java similarity index 91% rename from spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java index 2b553877..680d6aea 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.core.reactive; +package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; @@ -45,7 +45,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; /** - * Tests for {@link ReactivePulsarTemplate}. + * Tests for {@link org.springframework.pulsar.reactive.core.ReactivePulsarTemplate}. * * @author Christophe Bornet */ @@ -60,9 +60,10 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { .subscriptionName("test-specific-schema-subscription").subscribe()) { MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); senderSpec.setTopicName(topic); - ReactivePulsarSenderFactory producerFactory = new DefaultReactivePulsarSenderFactory<>(client, - senderSpec, null); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory); + org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>( + client, senderSpec, null); + org.springframework.pulsar.reactive.core.ReactivePulsarTemplate pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>( + producerFactory); pulsarTemplate.setSchema(Schema.JSON(Foo.class)); List foos = new ArrayList<>(); @@ -104,7 +105,8 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { } ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, senderSpec, null); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory); + org.springframework.pulsar.reactive.core.ReactivePulsarTemplate pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>( + senderFactory); Mono sendResponse; if (testArgs.useTemplateSchema) { pulsarTemplate.setSchema(Schema.STRING); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java similarity index 98% rename from spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java index 4524f987..be480902 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import static org.assertj.core.api.Assertions.assertThat; @@ -40,9 +40,9 @@ import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; import org.springframework.pulsar.core.PulsarTestContainerSupport; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarTemplate; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java new file mode 100644 index 00000000..c2f7fc6f --- /dev/null +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java @@ -0,0 +1,754 @@ +// Generated by the protocol buffer compiler. DO NOT EDIT! +// source: person.proto + +package org.springframework.pulsar.reactive.listener; + +public final class Proto { + + private Proto() { + } + + public static void registerAllExtensions(com.google.protobuf.ExtensionRegistryLite registry) { + } + + public static void registerAllExtensions(com.google.protobuf.ExtensionRegistry registry) { + registerAllExtensions((com.google.protobuf.ExtensionRegistryLite) registry); + } + + public interface PersonOrBuilder extends + // @@protoc_insertion_point(interface_extends:proto.Person) + com.google.protobuf.MessageOrBuilder { + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + boolean hasId(); + + /** + * optional int32 id = 1; + * @return The id. + */ + int getId(); + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + boolean hasName(); + + /** + * optional string name = 2; + * @return The name. + */ + String getName(); + + /** + * optional string name = 2; + * @return The bytes for name. + */ + com.google.protobuf.ByteString getNameBytes(); + + } + + /** + * Protobuf type {@code proto.Person} + */ + public static final class Person extends com.google.protobuf.GeneratedMessageV3 implements + // @@protoc_insertion_point(message_implements:proto.Person) + PersonOrBuilder { + + private static final long serialVersionUID = 0L; + + // Use Person.newBuilder() to construct. + private Person(com.google.protobuf.GeneratedMessageV3.Builder builder) { + super(builder); + } + + private Person() { + name_ = ""; + } + + @Override + @SuppressWarnings({ "unused" }) + protected Object newInstance(UnusedPrivateParameter unused) { + return new Person(); + } + + @Override + public final com.google.protobuf.UnknownFieldSet getUnknownFields() { + return this.unknownFields; + } + + private Person(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + this(); + if (extensionRegistry == null) { + throw new NullPointerException(); + } + int mutable_bitField0_ = 0; + com.google.protobuf.UnknownFieldSet.Builder unknownFields = com.google.protobuf.UnknownFieldSet + .newBuilder(); + try { + boolean done = false; + while (!done) { + int tag = input.readTag(); + switch (tag) { + case 0: + done = true; + break; + case 8: { + bitField0_ |= 0x00000001; + id_ = input.readInt32(); + break; + } + case 18: { + String s = input.readStringRequireUtf8(); + bitField0_ |= 0x00000002; + name_ = s; + break; + } + default: { + if (!parseUnknownField(input, unknownFields, extensionRegistry, tag)) { + done = true; + } + break; + } + } + } + } + catch (com.google.protobuf.InvalidProtocolBufferException e) { + throw e.setUnfinishedMessage(this); + } + catch (com.google.protobuf.UninitializedMessageException e) { + throw e.asInvalidProtocolBufferException().setUnfinishedMessage(this); + } + catch (java.io.IOException e) { + throw new com.google.protobuf.InvalidProtocolBufferException(e).setUnfinishedMessage(this); + } + finally { + this.unknownFields = unknownFields.build(); + makeExtensionsImmutable(); + } + } + + public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { + return Proto.internal_static_proto_Person_descriptor; + } + + @Override + protected FieldAccessorTable internalGetFieldAccessorTable() { + return Proto.internal_static_proto_Person_fieldAccessorTable.ensureFieldAccessorsInitialized(Person.class, + Builder.class); + } + + private int bitField0_; + + public static final int ID_FIELD_NUMBER = 1; + + private int id_; + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + @Override + public boolean hasId() { + return ((bitField0_ & 0x00000001) != 0); + } + + /** + * optional int32 id = 1; + * @return The id. + */ + @Override + public int getId() { + return id_; + } + + public static final int NAME_FIELD_NUMBER = 2; + + private volatile Object name_; + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + @Override + public boolean hasName() { + return ((bitField0_ & 0x00000002) != 0); + } + + /** + * optional string name = 2; + * @return The name. + */ + @Override + public String getName() { + Object ref = name_; + if (ref instanceof String) { + return (String) ref; + } + else { + com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; + String s = bs.toStringUtf8(); + name_ = s; + return s; + } + } + + /** + * optional string name = 2; + * @return The bytes for name. + */ + @Override + public com.google.protobuf.ByteString getNameBytes() { + Object ref = name_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((String) ref); + name_ = b; + return b; + } + else { + return (com.google.protobuf.ByteString) ref; + } + } + + private byte memoizedIsInitialized = -1; + + @Override + public final boolean isInitialized() { + byte isInitialized = memoizedIsInitialized; + if (isInitialized == 1) + return true; + if (isInitialized == 0) + return false; + + memoizedIsInitialized = 1; + return true; + } + + @Override + public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException { + if (((bitField0_ & 0x00000001) != 0)) { + output.writeInt32(1, id_); + } + if (((bitField0_ & 0x00000002) != 0)) { + com.google.protobuf.GeneratedMessageV3.writeString(output, 2, name_); + } + unknownFields.writeTo(output); + } + + @Override + public int getSerializedSize() { + int size = memoizedSize; + if (size != -1) + return size; + + size = 0; + if (((bitField0_ & 0x00000001) != 0)) { + size += com.google.protobuf.CodedOutputStream.computeInt32Size(1, id_); + } + if (((bitField0_ & 0x00000002) != 0)) { + size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, name_); + } + size += unknownFields.getSerializedSize(); + memoizedSize = size; + return size; + } + + @Override + public boolean equals(final Object obj) { + if (obj == this) { + return true; + } + if (!(obj instanceof Person)) { + return super.equals(obj); + } + Person other = (Person) obj; + + if (hasId() != other.hasId()) + return false; + if (hasId()) { + if (getId() != other.getId()) + return false; + } + if (hasName() != other.hasName()) + return false; + if (hasName()) { + if (!getName().equals(other.getName())) + return false; + } + if (!unknownFields.equals(other.unknownFields)) + return false; + return true; + } + + @Override + public int hashCode() { + if (memoizedHashCode != 0) { + return memoizedHashCode; + } + int hash = 41; + hash = (19 * hash) + getDescriptor().hashCode(); + if (hasId()) { + hash = (37 * hash) + ID_FIELD_NUMBER; + hash = (53 * hash) + getId(); + } + if (hasName()) { + hash = (37 * hash) + NAME_FIELD_NUMBER; + hash = (53 * hash) + getName().hashCode(); + } + hash = (29 * hash) + unknownFields.hashCode(); + memoizedHashCode = hash; + return hash; + } + + public static Person parseFrom(java.nio.ByteBuffer data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static Person parseFrom(java.nio.ByteBuffer data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static Person parseFrom(com.google.protobuf.ByteString data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static Person parseFrom(com.google.protobuf.ByteString data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static Person parseFrom(byte[] data) throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static Person parseFrom(byte[] data, com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static Person parseFrom(java.io.InputStream input) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input); + } + + public static Person parseFrom(java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry); + } + + public static Person parseDelimitedFrom(java.io.InputStream input) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input); + } + + public static Person parseDelimitedFrom(java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input, + extensionRegistry); + } + + public static Person parseFrom(com.google.protobuf.CodedInputStream input) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input); + } + + public static Person parseFrom(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry); + } + + @Override + public Builder newBuilderForType() { + return newBuilder(); + } + + public static Builder newBuilder() { + return DEFAULT_INSTANCE.toBuilder(); + } + + public static Builder newBuilder(Person prototype) { + return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype); + } + + @Override + public Builder toBuilder() { + return this == DEFAULT_INSTANCE ? new Builder() : new Builder().mergeFrom(this); + } + + @Override + protected Builder newBuilderForType(BuilderParent parent) { + Builder builder = new Builder(parent); + return builder; + } + + /** + * Protobuf type {@code proto.Person} + */ + public static final class Builder extends com.google.protobuf.GeneratedMessageV3.Builder implements + // @@protoc_insertion_point(builder_implements:proto.Person) + PersonOrBuilder { + + public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { + return Proto.internal_static_proto_Person_descriptor; + } + + @Override + protected FieldAccessorTable internalGetFieldAccessorTable() { + return Proto.internal_static_proto_Person_fieldAccessorTable + .ensureFieldAccessorsInitialized(Person.class, Builder.class); + } + + // Construct using + // org.springframework.pulsar.listener.Proto.Person.newBuilder() + private Builder() { + maybeForceBuilderInitialization(); + } + + private Builder(BuilderParent parent) { + super(parent); + maybeForceBuilderInitialization(); + } + + private void maybeForceBuilderInitialization() { + if (com.google.protobuf.GeneratedMessageV3.alwaysUseFieldBuilders) { + } + } + + @Override + public Builder clear() { + super.clear(); + id_ = 0; + bitField0_ = (bitField0_ & ~0x00000001); + name_ = ""; + bitField0_ = (bitField0_ & ~0x00000002); + return this; + } + + @Override + public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() { + return Proto.internal_static_proto_Person_descriptor; + } + + @Override + public Person getDefaultInstanceForType() { + return Person.getDefaultInstance(); + } + + @Override + public Person build() { + Person result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + @Override + public Person buildPartial() { + Person result = new Person(this); + int from_bitField0_ = bitField0_; + int to_bitField0_ = 0; + if (((from_bitField0_ & 0x00000001) != 0)) { + result.id_ = id_; + to_bitField0_ |= 0x00000001; + } + if (((from_bitField0_ & 0x00000002) != 0)) { + to_bitField0_ |= 0x00000002; + } + result.name_ = name_; + result.bitField0_ = to_bitField0_; + onBuilt(); + return result; + } + + @Override + public Builder clone() { + return super.clone(); + } + + @Override + public Builder setField(com.google.protobuf.Descriptors.FieldDescriptor field, Object value) { + return super.setField(field, value); + } + + @Override + public Builder clearField(com.google.protobuf.Descriptors.FieldDescriptor field) { + return super.clearField(field); + } + + @Override + public Builder clearOneof(com.google.protobuf.Descriptors.OneofDescriptor oneof) { + return super.clearOneof(oneof); + } + + @Override + public Builder setRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, int index, + Object value) { + return super.setRepeatedField(field, index, value); + } + + @Override + public Builder addRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, Object value) { + return super.addRepeatedField(field, value); + } + + @Override + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof Person) { + return mergeFrom((Person) other); + } + else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(Person other) { + if (other == Person.getDefaultInstance()) + return this; + if (other.hasId()) { + setId(other.getId()); + } + if (other.hasName()) { + bitField0_ |= 0x00000002; + name_ = other.name_; + onChanged(); + } + this.mergeUnknownFields(other.unknownFields); + onChanged(); + return this; + } + + @Override + public final boolean isInitialized() { + return true; + } + + @Override + public Builder mergeFrom(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + Person parsedMessage = null; + try { + parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry); + } + catch (com.google.protobuf.InvalidProtocolBufferException e) { + parsedMessage = (Person) e.getUnfinishedMessage(); + throw e.unwrapIOException(); + } + finally { + if (parsedMessage != null) { + mergeFrom(parsedMessage); + } + } + return this; + } + + private int bitField0_; + + private int id_; + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + @Override + public boolean hasId() { + return ((bitField0_ & 0x00000001) != 0); + } + + /** + * optional int32 id = 1; + * @return The id. + */ + @Override + public int getId() { + return id_; + } + + /** + * optional int32 id = 1; + * @param value The id to set. + * @return This builder for chaining. + */ + public Builder setId(int value) { + bitField0_ |= 0x00000001; + id_ = value; + onChanged(); + return this; + } + + /** + * optional int32 id = 1; + * @return This builder for chaining. + */ + public Builder clearId() { + bitField0_ = (bitField0_ & ~0x00000001); + id_ = 0; + onChanged(); + return this; + } + + private Object name_ = ""; + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + public boolean hasName() { + return ((bitField0_ & 0x00000002) != 0); + } + + /** + * optional string name = 2; + * @return The name. + */ + public String getName() { + Object ref = name_; + if (!(ref instanceof String)) { + com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; + String s = bs.toStringUtf8(); + name_ = s; + return s; + } + else { + return (String) ref; + } + } + + /** + * optional string name = 2; + * @return The bytes for name. + */ + public com.google.protobuf.ByteString getNameBytes() { + Object ref = name_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((String) ref); + name_ = b; + return b; + } + else { + return (com.google.protobuf.ByteString) ref; + } + } + + /** + * optional string name = 2; + * @param value The name to set. + * @return This builder for chaining. + */ + public Builder setName(String value) { + if (value == null) { + throw new NullPointerException(); + } + bitField0_ |= 0x00000002; + name_ = value; + onChanged(); + return this; + } + + /** + * optional string name = 2; + * @return This builder for chaining. + */ + public Builder clearName() { + bitField0_ = (bitField0_ & ~0x00000002); + name_ = getDefaultInstance().getName(); + onChanged(); + return this; + } + + /** + * optional string name = 2; + * @param value The bytes for name to set. + * @return This builder for chaining. + */ + public Builder setNameBytes(com.google.protobuf.ByteString value) { + if (value == null) { + throw new NullPointerException(); + } + checkByteStringIsUtf8(value); + bitField0_ |= 0x00000002; + name_ = value; + onChanged(); + return this; + } + + @Override + public final Builder setUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.setUnknownFields(unknownFields); + } + + @Override + public final Builder mergeUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.mergeUnknownFields(unknownFields); + } + + // @@protoc_insertion_point(builder_scope:proto.Person) + + } + + // @@protoc_insertion_point(class_scope:proto.Person) + private static final Person DEFAULT_INSTANCE; + static { + DEFAULT_INSTANCE = new Person(); + } + + public static Person getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final com.google.protobuf.Parser PARSER = new com.google.protobuf.AbstractParser() { + @Override + public Person parsePartialFrom(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return new Person(input, extensionRegistry); + } + }; + + public static com.google.protobuf.Parser parser() { + return PARSER; + } + + @Override + public com.google.protobuf.Parser getParserForType() { + return PARSER; + } + + @Override + public Person getDefaultInstanceForType() { + return DEFAULT_INSTANCE; + } + + } + + private static final com.google.protobuf.Descriptors.Descriptor internal_static_proto_Person_descriptor; + + private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_proto_Person_fieldAccessorTable; + + public static com.google.protobuf.Descriptors.FileDescriptor getDescriptor() { + return descriptor; + } + + private static com.google.protobuf.Descriptors.FileDescriptor descriptor; + static { + String[] descriptorData = { "\n\014person.proto\022\005proto\"<\n\006Person\022\017\n\002id\030\001 " + + "\001(\005H\000\210\001\001\022\021\n\004name\030\002 \001(\tH\001\210\001\001B\005\n\003_idB\007\n\005_n" + + "ameB,\n#org.springframework.pulsar.listen" + "erB\005Protob\006proto3" }; + descriptor = com.google.protobuf.Descriptors.FileDescriptor.internalBuildGeneratedFileFrom(descriptorData, + new com.google.protobuf.Descriptors.FileDescriptor[] {}); + internal_static_proto_Person_descriptor = getDescriptor().getMessageTypes().get(0); + internal_static_proto_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable( + internal_static_proto_Person_descriptor, new String[] { "Id", "Name", "Id", "Name", }); + } + + // @@protoc_insertion_point(outer_class_scope) + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java similarity index 96% rename from spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java rename to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index 37024110..944da9c2 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.listener.reactive; +package org.springframework.pulsar.reactive.listener; import static org.assertj.core.api.Assertions.assertThat; @@ -56,23 +56,22 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.handler.annotation.Header; -import org.springframework.pulsar.annotation.EnablePulsar; -import org.springframework.pulsar.annotation.ReactivePulsarListener; import org.springframework.pulsar.config.PulsarClientConfiguration; import org.springframework.pulsar.config.PulsarClientFactoryBean; -import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTestContainerSupport; import org.springframework.pulsar.core.PulsarTopic; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.listener.Proto; +import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -97,7 +96,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { private PulsarClient pulsarClient; @Configuration(proxyBeanMethods = false) - @EnablePulsar + @EnableReactivePulsar public static class TopLevelConfig { @Bean @@ -191,7 +190,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class TestPulsarListenersForBasicScenario { @@ -254,7 +253,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class TestPulsarListenersForStreaming { @@ -300,7 +299,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class DeadLetterPolicyConfig { @@ -397,7 +396,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue(); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class SchemaTestConfig { @@ -551,7 +550,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { .isEqualTo("hello-spring-messaging-message-listener".getBytes(StandardCharsets.UTF_8)); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class PulsarListenerWithHeadersConfig { @@ -636,7 +635,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { assertThat(queue.poll(5, TimeUnit.SECONDS)).isEqualTo("second"); } - @EnablePulsar + @EnableReactivePulsar @Configuration static class TestPulsarListenersForConcurrency { diff --git a/spring-pulsar-reactive/src/test/resources/logback-test.xml b/spring-pulsar-reactive/src/test/resources/logback-test.xml new file mode 100644 index 00000000..554cd1bd --- /dev/null +++ b/spring-pulsar-reactive/src/test/resources/logback-test.xml @@ -0,0 +1,12 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + diff --git a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java index 30cc19cd..1ed503e3 100644 --- a/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java +++ b/spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java @@ -36,9 +36,9 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.PulsarListener; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarTemplate; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; import reactor.core.publisher.Flux; diff --git a/spring-pulsar-spring-boot-autoconfigure/build.gradle b/spring-pulsar-spring-boot-autoconfigure/build.gradle index 9aba3321..547062c4 100644 --- a/spring-pulsar-spring-boot-autoconfigure/build.gradle +++ b/spring-pulsar-spring-boot-autoconfigure/build.gradle @@ -10,7 +10,7 @@ dependencies { annotationProcessor 'org.springframework.boot:spring-boot-configuration-processor' optional project (':spring-pulsar') - optional 'org.apache.pulsar:pulsar-client-reactive-adapter' + optional project (':spring-pulsar-reactive') optional 'org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine' implementation 'org.springframework.boot:spring-boot-starter' implementation 'com.google.code.findbugs:jsr305' diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java index e14fed6a..a4021116 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java @@ -22,11 +22,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.config.PulsarListenerBeanNames; -import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerProperties; +import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.listener.ReactivePulsarContainerProperties; /** * Configuration for Reactive Pulsar annotation-driven support. @@ -34,7 +34,7 @@ import org.springframework.pulsar.listener.reactive.ReactivePulsarContainerPrope * @author Christophe Bornet */ @Configuration(proxyBeanMethods = false) -@ConditionalOnClass(EnablePulsar.class) +@ConditionalOnClass(EnableReactivePulsar.class) public class PulsarReactiveAnnotationDrivenConfiguration { private final PulsarReactiveProperties properties; @@ -62,7 +62,7 @@ public class PulsarReactiveAnnotationDrivenConfiguration { } @Configuration(proxyBeanMethods = false) - @EnablePulsar + @EnableReactivePulsar @ConditionalOnMissingBean(name = PulsarListenerBeanNames.REACTIVE_PULSAR_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME) static class EnableReactivePulsarConfiguration { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java index 91d58ade..4eec43ff 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java @@ -32,13 +32,13 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarReaderFactory; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarReaderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarSenderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarTemplate; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; import com.github.benmanes.caffeine.cache.Caffeine; 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 c5ece4f9..d5651709 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 @@ -45,20 +45,20 @@ import org.springframework.boot.autoconfigure.AutoConfigurations; 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.annotation.EnablePulsar; -import org.springframework.pulsar.annotation.ReactivePulsarBootstrapConfiguration; -import org.springframework.pulsar.annotation.ReactivePulsarListenerAnnotationBeanPostProcessor; import org.springframework.pulsar.config.PulsarClientFactoryBean; -import org.springframework.pulsar.config.reactive.DefaultReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerContainerFactory; -import org.springframework.pulsar.config.reactive.ReactivePulsarListenerEndpointRegistry; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarReaderFactory; -import org.springframework.pulsar.core.reactive.DefaultReactivePulsarSenderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarConsumerFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarReaderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarSenderFactory; -import org.springframework.pulsar.core.reactive.ReactivePulsarTemplate; +import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; +import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; +import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarBootstrapConfiguration; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerAnnotationBeanPostProcessor; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory; +import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarReaderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory; +import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate; /** * Autoconfiguration tests for {@link PulsarReactiveAutoConfiguration}. @@ -86,7 +86,7 @@ class PulsarReactiveAutoConfigurationTests { @Test void annotationDrivenConfigurationSkippedWhenEnablePulsarAnnotationNotOnClasspath() { - this.contextRunner.withClassLoader(new FilteredClassLoader(EnablePulsar.class)) + this.contextRunner.withClassLoader(new FilteredClassLoader(EnableReactivePulsar.class)) .run((context) -> assertThat(context).hasNotFailed() .doesNotHaveBean(PulsarReactiveAnnotationDrivenConfiguration.class)); } 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 15728b2d..9f304f28 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 @@ -32,9 +32,9 @@ 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.annotation.ReactivePulsarListener; import org.springframework.pulsar.core.PulsarTemplate; -import org.springframework.pulsar.core.reactive.ReactiveMessageConsumerBuilderCustomizer; +import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener; +import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index 8e69e022..177c0938 100644 --- a/spring-pulsar/build.gradle +++ b/spring-pulsar/build.gradle @@ -18,7 +18,6 @@ dependencies { implementation 'com.fasterxml.jackson.core:jackson-core' implementation 'com.fasterxml.jackson.core:jackson-databind' implementation 'com.google.code.findbugs:jsr305' - optional 'org.apache.pulsar:pulsar-client-reactive-adapter' optional 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8' optional 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310' optional 'com.fasterxml.jackson.datatype:jackson-datatype-joda' @@ -31,7 +30,6 @@ dependencies { testImplementation 'io.micrometer:micrometer-tracing-bridge-brave' testImplementation 'io.micrometer:micrometer-tracing-test' testImplementation 'io.micrometer:micrometer-tracing-integration-test' - testImplementation 'io.projectreactor:reactor-test' testImplementation 'org.assertj:assertj-core' testImplementation 'org.awaitility:awaitility' testImplementation 'org.hamcrest:hamcrest' diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java index 2eeac1de..5b662a40 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java @@ -22,19 +22,16 @@ import org.springframework.core.type.AnnotationMetadata; /** * A {@link DeferredImportSelector} implementation with the lowest order to import - * {@link PulsarBootstrapConfiguration} and {@link ReactivePulsarBootstrapConfiguration} - * as late as possible. + * {@link PulsarBootstrapConfiguration} as late as possible. * * @author Soby Chacko - * */ @Order public class PulsarListenerConfigurationSelector implements DeferredImportSelector { @Override public String[] selectImports(AnnotationMetadata importingClassMetadata) { - return new String[] { PulsarBootstrapConfiguration.class.getName(), - ReactivePulsarBootstrapConfiguration.class.getName() }; + return new String[] { PulsarBootstrapConfiguration.class.getName() }; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java index deb8249d..54155f9e 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java @@ -17,7 +17,6 @@ package org.springframework.pulsar.config; import org.springframework.pulsar.annotation.PulsarListener; -import org.springframework.pulsar.annotation.ReactivePulsarListener; import org.springframework.pulsar.listener.MessageListenerContainer; /** @@ -40,9 +39,8 @@ public interface ListenerContainerFactory