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

Split reactive into its own module

Browse Source 
Remove redundant dep
This commit is contained in:
Chris Bono
2022-11-17 18:21:44 -06:00
committed by Soby Chacko
parent 0a6808441b
commit a8d65ec168
64 changed files with 1335 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
settings.gradle
View File

@@ -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'

2
spring-pulsar-reactive-spring-boot-starter/build.gradle
View File

@@ -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'
}

34
spring-pulsar-reactive/build.gradle Normal file
View File

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

111
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/aot/ReactivePulsarRuntimeHints.java Normal file
View File

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

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/AbstractReactivePulsarListenerEndpoint.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/AbstractReactivePulsarListenerEndpoint.java
View File

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

8
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/DefaultReactivePulsarListenerContainerFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java
View File

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

14
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/MethodReactivePulsarListenerEndpoint.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java
View File

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

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerContainerFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerContainerFactory.java
View File

@@ -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.

10
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpoint.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpoint.java
View File

@@ -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 <T> Message payload type.
* @author Christophe Bornet
* @author Chris Bono
*/
public interface ReactivePulsarListenerEndpoint<T> extends ListenerEndpoint<ReactivePulsarMessageListenerContainer<T>> {

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointAdapter.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointAdapter.java
View File

@@ -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;
/**

4
spring-pulsar/src/main/java/org/springframework/pulsar/config/reactive/ReactivePulsarListenerEndpointRegistry.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/ReactivePulsarListenerEndpointRegistry.java
View File

@@ -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

39
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/EnableReactivePulsar.java Normal file
View File

@@ -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 {
}

10
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarBootstrapConfiguration.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarBootstrapConfiguration.java
View File

@@ -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}.
*
* <p>
* 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 {

14
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListener.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListener.java
View File

@@ -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
* <p>
* 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.
* </p>
*
* @author Christophe Bornet
@@ -58,7 +59,7 @@ public @interface ReactivePulsarListener {
* <p>
* If none is specified an auto-generated id is used.
* <p>
* 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 "";

22
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -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}.
*
* <p>
* This post-processor is automatically registered by the {@link EnablePulsar} annotation.
* This post-processor is automatically registered by the {@link EnableReactivePulsar}
* annotation.
*
* <p>
* 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 <V> the payload type.
* @author Christophe Bornet
* @see ReactivePulsarListener
* @see EnablePulsar
* @see EnableReactivePulsar
* @see PulsarListenerConfigurer
* @see PulsarListenerEndpointRegistrar
* @see ReactivePulsarListenerEndpointRegistry

37
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListenerConfigurationSelector.java Normal file
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/ReactivePulsarListeners.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/ReactivePulsarListeners.java
View File

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

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/annotation/package-info.java Normal file
View File

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

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/package-info.java Normal file
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarConsumerFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactory.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarReaderFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactory.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.List;

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/MessageSpecBuilderCustomizer.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageConsumerBuilderCustomizer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageConsumerBuilderCustomizer.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageReaderBuilderCustomizer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageReaderBuilderCustomizer.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderBuilderCustomizer.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactiveMessageSenderUtils.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.Optional;

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarConsumerFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarConsumerFactory.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.List;

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarOperations.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarReaderFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarReaderFactory.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.List;

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarSenderFactory.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.List;

2
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplate.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.pulsar.core.reactive;
package org.springframework.pulsar.reactive.core;
import java.util.Collections;

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/package-info.java Normal file
View File

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

6
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainer.java
View File

@@ -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;
/**

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarContainerProperties.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageHandler.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageHandler.java
View File

@@ -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}.

4
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarMessageListenerContainer.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarMessageListenerContainer.java
View File

@@ -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

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarOneByOneMessageHandler.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarOneByOneMessageHandler.java
View File

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

2
spring-pulsar/src/main/java/org/springframework/pulsar/listener/reactive/ReactivePulsarStreamingHandler.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarStreamingHandler.java
View File

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

8
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveOneByOneMessagingMessageListenerAdapter.java
View File

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

8
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java → spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarReactiveStreamingMessagingMessageListenerAdapter.java
View File

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

69
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java Normal file
View File

@@ -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 <V> payload type.
* @author Soby Chacko
*/
@SuppressWarnings("serial")
public class PulsarRecordMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements PulsarAcknowledgingMessageListener<V> {
public PulsarRecordMessagingMessageListenerAdapter(Object bean, Method method) {
super(bean, method);
}
@Override
public void received(Consumer<V> consumer, Message<V> 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;
}
}
}

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/adapter/package-info.java Normal file
View File

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

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/package-info.java Normal file
View File

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

1
spring-pulsar-reactive/src/main/resources/META-INF/spring/aot.factories Normal file
View File

@@ -0,0 +1 @@
org.springframework.aot.hint.RuntimeHintsRegistrar=org.springframework.pulsar.reactive.aot.ReactivePulsarRuntimeHints

64
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/core/PulsarTestContainerSupport.java Normal file
View File

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

9
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageConsumerFactoryTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageConsumerFactoryTests.java
View File

@@ -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<String> consumerFactory = new DefaultReactivePulsarConsumerFactory<>(
private org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory<String> 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<String> consumerFactory;
private org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory<String> consumerFactory;
@BeforeEach
void createConsumerFactory() {

9
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageReaderFactoryTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageReaderFactoryTests.java
View File

@@ -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<String> readerFactory = new DefaultReactivePulsarReaderFactory<>(
org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory<String> readerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory<>(
AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
ReactiveMessageReader<String> reader = readerFactory.createReader(schema);
@@ -55,7 +56,7 @@ class DefaultReactiveMessageReaderFactoryTests {
void createReaderWithCustomizer() {
MutableReactiveMessageReaderSpec spec = new MutableReactiveMessageReaderSpec();
spec.setReaderName("test-reader");
DefaultReactivePulsarReaderFactory<String> readerFactory = new DefaultReactivePulsarReaderFactory<>(
org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory<String> readerFactory = new DefaultReactivePulsarReaderFactory<>(
AdaptedReactivePulsarClientFactory.create((PulsarClient) null), spec);
ReactiveMessageReader<String> reader = readerFactory.createReader(schema,

13
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageSenderFactoryTests.java
View File

@@ -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<String> customizer1 = builder -> builder.topic("topic1");
org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer<String> customizer1 = builder -> builder
.topic("topic1");
ReactiveMessageSenderCache cache = AdaptedReactivePulsarClientFactory.createCache();
ReactiveMessageSenderBuilderCustomizer<String> customizer2 = builder -> builder.cache(cache);
org.springframework.pulsar.reactive.core.ReactiveMessageSenderBuilderCustomizer<String> customizer2 = builder -> builder
.cache(cache);
ReactiveMessageSender<String> sender = testCreateSender(null, null, "topic0",
Arrays.asList(customizer1, customizer2), "topic1");
@@ -73,7 +76,7 @@ class DefaultReactiveMessageSenderFactoryTests {
@Test
void createSenderWithNoTopic() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(
org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory<String> 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");

14
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java
View File

@@ -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<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec, null);
ReactivePulsarTemplate<Foo> pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory);
org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory<Foo> producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>(
client, senderSpec, null);
org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<Foo> pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>(
producerFactory);
pulsarTemplate.setSchema(Schema.JSON(Foo.class));
List<Foo> foos = new ArrayList<>();
@@ -104,7 +105,8 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
}
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec, null);
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<String> pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>(
senderFactory);
Mono<MessageId> sendResponse;
if (testArgs.useTemplateSchema) {
pulsarTemplate.setSchema(Schema.STRING);

8
spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/DefaultReactivePulsarMessageListenerContainerTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java
View File

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

754
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/Proto.java Normal file
View File

@@ -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 {
/**
* <code>optional int32 id = 1;</code>
* @return Whether the id field is set.
*/
boolean hasId();
/**
* <code>optional int32 id = 1;</code>
* @return The id.
*/
int getId();
/**
* <code>optional string name = 2;</code>
* @return Whether the name field is set.
*/
boolean hasName();
/**
* <code>optional string name = 2;</code>
* @return The name.
*/
String getName();
/**
* <code>optional string name = 2;</code>
* @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_;
/**
* <code>optional int32 id = 1;</code>
* @return Whether the id field is set.
*/
@Override
public boolean hasId() {
return ((bitField0_ & 0x00000001) != 0);
}
/**
* <code>optional int32 id = 1;</code>
* @return The id.
*/
@Override
public int getId() {
return id_;
}
public static final int NAME_FIELD_NUMBER = 2;
private volatile Object name_;
/**
* <code>optional string name = 2;</code>
* @return Whether the name field is set.
*/
@Override
public boolean hasName() {
return ((bitField0_ & 0x00000002) != 0);
}
/**
* <code>optional string name = 2;</code>
* @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;
}
}
/**
* <code>optional string name = 2;</code>
* @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<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_;
/**
* <code>optional int32 id = 1;</code>
* @return Whether the id field is set.
*/
@Override
public boolean hasId() {
return ((bitField0_ & 0x00000001) != 0);
}
/**
* <code>optional int32 id = 1;</code>
* @return The id.
*/
@Override
public int getId() {
return id_;
}
/**
* <code>optional int32 id = 1;</code>
* @param value The id to set.
* @return This builder for chaining.
*/
public Builder setId(int value) {
bitField0_ |= 0x00000001;
id_ = value;
onChanged();
return this;
}
/**
* <code>optional int32 id = 1;</code>
* @return This builder for chaining.
*/
public Builder clearId() {
bitField0_ = (bitField0_ & ~0x00000001);
id_ = 0;
onChanged();
return this;
}
private Object name_ = "";
/**
* <code>optional string name = 2;</code>
* @return Whether the name field is set.
*/
public boolean hasName() {
return ((bitField0_ & 0x00000002) != 0);
}
/**
* <code>optional string name = 2;</code>
* @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;
}
}
/**
* <code>optional string name = 2;</code>
* @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;
}
}
/**
* <code>optional string name = 2;</code>
* @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;
}
/**
* <code>optional string name = 2;</code>
* @return This builder for chaining.
*/
public Builder clearName() {
bitField0_ = (bitField0_ & ~0x00000002);
name_ = getDefaultInstance().getName();
onChanged();
return this;
}
/**
* <code>optional string name = 2;</code>
* @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<Person> PARSER = new com.google.protobuf.AbstractParser<Person>() {
@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<Person> parser() {
return PARSER;
}
@Override
public com.google.protobuf.Parser<Person> 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)
}

33
spring-pulsar/src/test/java/org/springframework/pulsar/listener/reactive/ReactivePulsarListenerTests.java → spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java
View File

@@ -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 {

12
spring-pulsar-reactive/src/test/resources/logback-test.xml Normal file
View File

@@ -0,0 +1,12 @@
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
</encoder>
</appender>
<root level="WARN">
<appender-ref ref="STDOUT"/>
</root>
<logger name="org.testcontainers" level="ERROR"/>
<logger name="com.github.dockerjava" level="ERROR"/>
</configuration>

6
spring-pulsar-sample-apps/sample-reactive/src/main/java/org.springframework.pulsar.example/ReactiveSpringPulsarBootApp.java
View File

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

2
spring-pulsar-spring-boot-autoconfigure/build.gradle
View File

@@ -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'

12
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java
View File

@@ -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 {

14
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java
View File

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

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

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

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

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

2
spring-pulsar/build.gradle
View File

@@ -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'

7
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerConfigurationSelector.java
View File

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

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/ListenerContainerFactory.java
View File

@@ -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<C extends MessageListenerContainer, E
/**
* Create and configure a container without a listener; used to create containers that
* are not used for {@link PulsarListener} and {@link ReactivePulsarListener}
* annotations. Containers created using this method are not added to the listener
* endpoint registry.
* are not used for {@link PulsarListener} annotations. Containers created using this
* method are not added to the listener endpoint registry.
* @param topics the topics.
* @return the container.
*/

2
spring-pulsar/src/main/resources/META-INF/spring/aot.factories
View File

@@ -1 +1 @@
org.springframework.aot.hint.RuntimeHintsRegistrar=org.springframework.pulsar.aot.PulsarRuntimeHints
org.springframework.aot.hint.RuntimeHintsRegistrar=org.springframework.pulsar.aot.PulsarRuntimeHints