Deprecate (Reactive)PulsarListenerEndpointAdapter (#481)

This commit deprecates PulsarListenerEndpointAdapter and 
ReactivePulsarListenerEndpointAdapter in favor of default methods on 
ListenerEndpoint and its sub-interfaces, which makes it a bit easier to
provide custom ListenerEndpoint implementations.
This commit is contained in:
Vedran Pavic
2023-12-16 04:39:39 +01:00
committed by GitHub
parent aebb7112d1
commit cf782afb78
7 changed files with 51 additions and 16 deletions

View File

@@ -156,7 +156,7 @@ public class DefaultReactivePulsarListenerContainerFactory<T> implements Reactiv
@Override
public DefaultReactivePulsarMessageListenerContainer<T> createContainer(String... topics) {
ReactivePulsarListenerEndpoint<T> endpoint = new ReactivePulsarListenerEndpointAdapter<>() {
ReactivePulsarListenerEndpoint<T> endpoint = new ReactivePulsarListenerEndpoint<>() {
@Override
public List<String> getTopics() {

View File

@@ -29,12 +29,17 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarMessageListene
* @param <T> Message payload type.
* @author Christophe Bornet
* @author Chris Bono
* @author Vedran Pavic
*/
public interface ReactivePulsarListenerEndpoint<T> extends ListenerEndpoint<ReactivePulsarMessageListenerContainer<T>> {
boolean isFluxListener();
default boolean isFluxListener() {
return false;
}
@Nullable
Boolean getUseKeyOrderedProcessing();
default Boolean getUseKeyOrderedProcessing() {
return null;
}
}

View File

@@ -31,7 +31,9 @@ import org.springframework.pulsar.support.MessageConverter;
*
* @param <T> Message payload type.
* @author Christophe Bornet
* @deprecated for removal in favor of {@link ReactivePulsarListenerEndpoint}
*/
@Deprecated(forRemoval = true)
public class ReactivePulsarListenerEndpointAdapter<T> implements ReactivePulsarListenerEndpoint<T> {
@Override

View File

@@ -54,7 +54,7 @@ public class ConcurrentPulsarListenerContainerFactory<T>
@Override
public ConcurrentPulsarMessageListenerContainer<T> createContainer(String... topics) {
PulsarListenerEndpoint endpoint = new PulsarListenerEndpointAdapter() {
PulsarListenerEndpoint endpoint = new PulsarListenerEndpoint() {
@Override
public Collection<String> getTopics() {

View File

@@ -17,6 +17,7 @@
package org.springframework.pulsar.config;
import java.util.Collection;
import java.util.Collections;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.schema.SchemaType;
@@ -32,6 +33,7 @@ import org.springframework.pulsar.support.MessageConverter;
*
* @param <C> Message listener container type.
* @author Christophe Bornet
* @author Vedran Pavic
*/
public interface ListenerEndpoint<C extends MessageListenerContainer> {
@@ -42,53 +44,69 @@ public interface ListenerEndpoint<C extends MessageListenerContainer> {
* @see ListenerContainerFactory#createListenerContainer
*/
@Nullable
String getId();
default String getId() {
return null;
}
/**
* Return the subscription name for this endpoint's container.
* @return the subscription name.
*/
@Nullable
String getSubscriptionName();
default String getSubscriptionName() {
return null;
}
/**
* Return the subscription type for this endpoint's container.
* @return the subscription type.
*/
@Nullable
SubscriptionType getSubscriptionType();
default SubscriptionType getSubscriptionType() {
return SubscriptionType.Exclusive;
}
/**
* Return the topics for this endpoint's container.
* @return the topics.
*/
Collection<String> getTopics();
default Collection<String> getTopics() {
return Collections.emptyList();
}
/**
* Return the topic pattern for this endpoint's container.
* @return the topic pattern.
*/
String getTopicPattern();
default String getTopicPattern() {
return null;
}
/**
* Return the autoStartup for this endpoint's container.
* @return the autoStartup.
*/
@Nullable
Boolean getAutoStartup();
default Boolean getAutoStartup() {
return null;
}
/**
* Return the schema type for this endpoint's container.
* @return the schema type.
*/
SchemaType getSchemaType();
default SchemaType getSchemaType() {
return null;
}
/**
* Return the concurrency for this endpoint's container.
* @return the concurrency.
*/
@Nullable
Integer getConcurrency();
default Integer getConcurrency() {
return null;
}
/**
* Setup the specified message listener container with the model defined by this
@@ -101,6 +119,7 @@ public interface ListenerEndpoint<C extends MessageListenerContainer> {
* @param listenerContainer the listener container to configure
* @param messageConverter the message converter - can be null
*/
void setupListenerContainer(C listenerContainer, @Nullable MessageConverter messageConverter);
default void setupListenerContainer(C listenerContainer, @Nullable MessageConverter messageConverter) {
}
}

View File

@@ -28,13 +28,20 @@ import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
*
* @author Soby Chacko
* @author Alexander Preuß
* @author Vedran Pavic
*/
public interface PulsarListenerEndpoint extends ListenerEndpoint<PulsarMessageListenerContainer> {
boolean isBatchListener();
default boolean isBatchListener() {
return false;
}
Properties getConsumerProperties();
default Properties getConsumerProperties() {
return null;
}
AckMode getAckMode();
default AckMode getAckMode() {
return null;
}
}

View File

@@ -33,7 +33,9 @@ import org.springframework.pulsar.support.MessageConverter;
*
* @author Soby Chacko
* @author Alexander Preuß
* @deprecated for removal in favor of {@link PulsarListenerEndpoint}
*/
@Deprecated(forRemoval = true)
public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint {
@Override