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

Support Pulsar Headers (#151)

Browse Source 
Convert Pulsar Message metadata as PulsarHeaders and make them
available on PulsarListener through Spring's @Header annotation.

See #150
This commit is contained in:
Soby Chacko
2022-09-28 13:06:41 -04:00
committed by GitHub
parent 030a84bd73
commit 4eac6b50f2
9 changed files with 610 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java
View File

@@ -41,6 +41,7 @@ import org.springframework.core.log.LogAccessor;
import org.springframework.expression.BeanResolver;
import org.springframework.lang.Nullable;
import org.springframework.messaging.converter.SmartMessageConverter;
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;
@@ -131,11 +132,13 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
MethodParameter messageParameter = null;
final Optional<MethodParameter> parameter = Arrays.stream(methodParameters)
.filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class)
|| !methodParameter1.getParameterType().equals(Acknowledgement.class))
|| !methodParameter1.getParameterType().equals(Acknowledgement.class)
|| !methodParameter1.hasParameterAnnotation(Header.class))
.findFirst();
final long count = Arrays.stream(methodParameters)
.filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class)
&& !methodParameter1.getParameterType().equals(Acknowledgement.class))
&& !methodParameter1.getParameterType().equals(Acknowledgement.class)
&& !methodParameter1.hasParameterAnnotation(Header.class))
.count();
Assert.isTrue(count == 1, "More than 1 expected payload types found");
if (parameter.isPresent()) {
@@ -222,7 +225,8 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
if (rawClass != null && isContainerType(rawClass)) {
resolvableType = resolvableType.getGeneric(0);
}
if (Message.class.isAssignableFrom(resolvableType.getRawClass())) {
if (Message.class.isAssignableFrom(resolvableType.getRawClass())
|| org.springframework.messaging.Message.class.isAssignableFrom(resolvableType.getRawClass())) {
resolvableType = resolvableType.getGeneric(0);
}
return resolvableType;
@@ -230,7 +234,8 @@ public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpo
private boolean isContainerType(Class<?> rawClass) {
return rawClass.isAssignableFrom(List.class) || rawClass.isAssignableFrom(Message.class)
|| rawClass.isAssignableFrom(Messages.class);
|| rawClass.isAssignableFrom(Messages.class)
|| rawClass.isAssignableFrom(org.springframework.messaging.Message.class);
}
protected HandlerAdapter configureListenerAdapter(PulsarMessagingMessageListenerAdapter<V> messageListener) {

111
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java
View File

@@ -18,8 +18,10 @@ package org.springframework.pulsar.listener.adapter;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Messages;
@@ -68,58 +70,83 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagi
@Override
public void received(Consumer<V> consumer, List<org.apache.pulsar.client.api.Message<V>> msg,
@Nullable Acknowledgement acknowledgement) {
Message<?> message;
if (!isConsumerRecordList()) {
if (isMessageList()) {
List<Message<?>> messages = new ArrayList<>(msg.size());
for (org.apache.pulsar.client.api.Message<V> record : msg) {
messages.add(toMessagingMessage(record, consumer));
Message<?> message = null;
Object theRecord = null;
if (isPulsarMessageList() && !isHeaderFound()) { // List<PulsarMessage>
theRecord = msg; // the incoming list as is.
}
else if (isPulsarMessageList() && isHeaderFound()) { // List<PulsarMessage>,
// @Header
List<Message<?>> messages = toSpringMessages(consumer, msg);
final Map<String, List<Object>> aggregatedHeaders = withAggregatedHeaders(messages);
List<Object> list1 = new ArrayList<>(msg);
message = MessageBuilder.withPayload(list1).copyHeaders(aggregatedHeaders).build();
}
else if (isMessageList() && !isHeaderFound()) { // List<SpringMessage>
List<Message<?>> messages = toSpringMessages(consumer, msg);
message = MessageBuilder.withPayload(messages).build();
}
else if (isMessageList() && isHeaderFound()) { // List<SpringMessage>, @Header
List<Message<?>> messages = toSpringMessages(consumer, msg);
final Map<String, List<Object>> aggregatedHeaders = withAggregatedHeaders(messages);
message = MessageBuilder.withPayload(messages).copyHeaders(aggregatedHeaders).build();
}
else if (this.isSimpleExtraction()) { // List<Object>
List<V> list = new ArrayList<>(msg.size());
msg.stream().iterator().forEachRemaining(vMessage -> list.add(vMessage.getValue()));
theRecord = list;
}
else if (isHeaderFound()) { // List<Object>, @Header
List<Message<?>> messages = toSpringMessages(consumer, msg);
final Map<String, List<Object>> aggregatedHeaders = withAggregatedHeaders(messages);
List<V> list = new ArrayList<>(msg.size());
msg.stream().iterator().forEachRemaining(vMessage -> list.add(vMessage.getValue()));
message = MessageBuilder.withPayload(list).copyHeaders(aggregatedHeaders).build();
}
if (isConsumerRecords()) { // Messages<String>
theRecord = new Messages<V>() {
@Override
public Iterator<org.apache.pulsar.client.api.Message<V>> iterator() {
return msg.iterator();
}
message = MessageBuilder.withPayload(messages).build();
}
else {
message = toMessagingMessage(msg, consumer);
}
@Override
public int size() {
return msg.size();
}
};
}
else {
message = MessageBuilder.withPayload(msg).build();
}
logger.debug(() -> "Processing [" + message + "]");
// In order to avoid clash with target List payload type.
final Messages<V> messages = new Messages<>() {
@Override
public Iterator<org.apache.pulsar.client.api.Message<V>> iterator() {
return msg.iterator();
}
@Override
public int size() {
return msg.size();
}
};
invoke(messages, consumer, message, acknowledgement);
invoke(theRecord, consumer, message, acknowledgement);
}
protected void invoke(Object records, Consumer<V> consumer, final Message<?> messageArg,
Acknowledgement acknowledgement) {
private Map<String, List<Object>> withAggregatedHeaders(List<Message<?>> messages) {
final Map<String, List<Object>> aggregatedHeaders = new HashMap<>();
for (Message<?> message : messages) {
message.getHeaders().forEach((s, o) -> {
List<Object> objects = aggregatedHeaders.computeIfAbsent(s, k -> new ArrayList<>());
objects.add(o);
});
}
return aggregatedHeaders;
}
Message<?> message = messageArg;
private List<Message<?>> toSpringMessages(Consumer<V> consumer, List<org.apache.pulsar.client.api.Message<V>> msg) {
List<Message<?>> messages = new ArrayList<>(msg.size());
msg.stream().iterator().forEachRemaining(record -> messages.add(toMessagingMessage(record, consumer)));
return messages;
}
protected void invoke(Object records, Consumer<V> consumer, final Message<?> message,
Acknowledgement acknowledgement) {
try {
Object result = invokeHandler(records, message, consumer, acknowledgement);
// if (result != null) {
// handleResult(result, records, message);
// }
invokeHandler(records, message, consumer, acknowledgement);
}
catch (Exception e) {
throw e;
}
}
protected Message<?> toMessagingMessage(List<org.apache.pulsar.client.api.Message<V>> msg, Consumer<V> consumer) {
return getBatchMessageConverter().toMessage(msg, consumer, getType());
}
}

158
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java
View File

@@ -20,7 +20,6 @@ import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.WildcardType;
import java.util.Collection;
import java.util.List;
import org.apache.commons.logging.LogFactory;
@@ -37,8 +36,10 @@ import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.expression.spel.support.StandardTypeConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.pulsar.listener.Acknowledgement;
import org.springframework.pulsar.support.DefaultPulsarMessageHeaderMapper;
import org.springframework.pulsar.support.converter.PulsarMessagingMessageConverter;
import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter;
import org.springframework.util.Assert;
@@ -64,19 +65,22 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
private HandlerAdapter handlerMethod;
private boolean conversionNeeded = true;
private boolean headerFound = false;
private boolean messageReturnType;
private boolean simpleExtraction = false;
private boolean isConsumerRecordList;
private boolean isPulsarMessageList;
private boolean isMessageList;
private boolean isSpringMessageList;
private boolean isSpringMessage;
private boolean isConsumerRecords;
private boolean converterSet;
private PulsarRecordMessageConverter<V> messageConverter = new PulsarMessagingMessageConverter<V>();
private PulsarRecordMessageConverter<V> messageConverter = new PulsarMessagingMessageConverter<V>(
new DefaultPulsarMessageHeaderMapper());
private Type fallbackType = Object.class;
@@ -112,16 +116,8 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
this.handlerMethod = handlerMethod;
}
protected boolean isConsumerRecordList() {
return this.isConsumerRecordList;
}
public boolean isConsumerRecords() {
return this.isConsumerRecords;
}
public boolean isConversionNeeded() {
return this.conversionNeeded;
protected boolean isPulsarMessageList() {
return this.isPulsarMessageList;
}
public void setBeanResolver(BeanResolver beanResolver) {
@@ -131,7 +127,7 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
}
protected boolean isMessageList() {
return this.isMessageList;
return this.isSpringMessageList;
}
protected org.springframework.messaging.Message<?> toMessagingMessage(Message<V> record, Consumer<V> consumer) {
@@ -143,12 +139,6 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
try {
return this.handlerMethod.invoke(message, data, consumer, acknowledgement);
// if (data instanceof List && !this.isConsumerRecordList) {
// return this.handlerMethod.invoke(message, consumer);
// }
// else {
// return this.handlerMethod.invoke(message, data, consumer);
// }
}
catch (Exception ex) {
throw new MessageConversionException("Cannot handle message", ex);
@@ -159,10 +149,32 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
if (method == null) {
return null;
}
Type genericParameterType = null;
int allowedBatchParameters = 1;
int notConvertibleParameters = 0;
boolean pulsarMessageFound = false;
boolean collectionFound = false;
for (int i = 0; i < method.getParameterCount(); i++) {
MethodParameter methodParameter = new MethodParameter(method, i);
Type parameterType = methodParameter.getGenericParameterType();
if (methodParameter.hasParameterAnnotation(Header.class)) {
this.headerFound = true;
}
else if (parameterIsType(parameterType, org.springframework.messaging.Message.class)) {
this.isSpringMessage = true;
}
else if (parameterIsType(parameterType, Message.class)) {
pulsarMessageFound = true;
}
else if (parameterIsType(parameterType, List.class) || parameterIsType(parameterType, Messages.class)) {
collectionFound = true;
}
}
if (!this.headerFound && !this.isSpringMessage && !pulsarMessageFound && !collectionFound) {
this.simpleExtraction = true;
}
for (int i = 0; i < method.getParameterCount(); i++) {
MethodParameter methodParameter = new MethodParameter(method, i);
@@ -173,10 +185,7 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
*/
Type parameterType = methodParameter.getGenericParameterType();
boolean isNotConvertible = parameterIsType(parameterType, Message.class);
boolean isConsumer = parameterIsType(parameterType, Consumer.class);
if (isNotConvertible) {
notConvertibleParameters++;
}
if (!isNotConvertible && !isMessageWithNoTypeInfo(parameterType)
&& (methodParameter.getParameterAnnotations().length == 0
|| methodParameter.hasParameterAnnotation(Payload.class))) {
@@ -189,37 +198,13 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
break;
}
}
else {
if (isConsumer) {
allowedBatchParameters++;
}
else {
if (parameterType instanceof ParameterizedType
&& ((ParameterizedType) parameterType).getRawType().equals(Consumer.class)) {
allowedBatchParameters++;
}
}
}
}
if (notConvertibleParameters == method.getParameterCount() && method.getReturnType().equals(void.class)) {
this.conversionNeeded = false;
}
boolean validParametersForBatch = method.getGenericParameterTypes().length <= allowedBatchParameters;
if (!validParametersForBatch) {
String stateMessage = "A parameter of type '%s' must be the only parameter "
+ "(except for an optional 'Acknowledgment' and/or 'Consumer' "
+ "and/or '@Header(KafkaHeaders.GROUP_ID) String groupId'";
}
this.messageReturnType = returnTypeMessageOrCollectionOf(method);
return genericParameterType;
}
private Type extractGenericParameterTypFromMethodParameter(MethodParameter methodParameter) {
Type genericParameterType = methodParameter.getGenericParameterType();
if (genericParameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) genericParameterType;
if (genericParameterType instanceof ParameterizedType parameterizedType) {
if (parameterizedType.getRawType().equals(org.springframework.messaging.Message.class)) {
genericParameterType = ((ParameterizedType) genericParameterType).getActualTypeArguments()[0];
}
@@ -227,14 +212,19 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
&& parameterizedType.getActualTypeArguments().length == 1) {
Type paramType = parameterizedType.getActualTypeArguments()[0];
this.isConsumerRecordList = paramType instanceof ParameterizedType
this.isPulsarMessageList = paramType instanceof ParameterizedType
&& ((ParameterizedType) paramType).getRawType().equals(Message.class);
boolean messageHasGeneric = paramType instanceof ParameterizedType && ((ParameterizedType) paramType)
.getRawType().equals(org.springframework.messaging.Message.class);
this.isMessageList = paramType.equals(org.springframework.messaging.Message.class) || messageHasGeneric;
this.isSpringMessageList = paramType.equals(org.springframework.messaging.Message.class)
|| messageHasGeneric;
if (messageHasGeneric) {
genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0];
}
if (!this.isSpringMessageList && !this.isPulsarMessageList && !isHeaderFound()) {
this.simpleExtraction = true;
}
}
else {
this.isConsumerRecords = parameterizedType.getRawType().equals(Messages.class);
@@ -243,33 +233,8 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
return genericParameterType;
}
public static boolean returnTypeMessageOrCollectionOf(Method method) {
Type returnType = method.getGenericReturnType();
if (returnType.equals(org.springframework.messaging.Message.class)) {
return true;
}
if (returnType instanceof ParameterizedType) {
ParameterizedType prt = (ParameterizedType) returnType;
Type rawType = prt.getRawType();
if (rawType.equals(org.springframework.messaging.Message.class)) {
return true;
}
if (rawType.equals(Collection.class)) {
Type collectionType = prt.getActualTypeArguments()[0];
if (collectionType.equals(org.springframework.messaging.Message.class)) {
return true;
}
return collectionType instanceof ParameterizedType && ((ParameterizedType) collectionType).getRawType()
.equals(org.springframework.messaging.Message.class);
}
}
return false;
}
private boolean parameterIsType(Type parameterType, Type type) {
if (parameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) parameterType;
if (parameterType instanceof ParameterizedType parameterizedType) {
Type rawType = parameterizedType.getRawType();
if (rawType.equals(type)) {
return true;
@@ -279,20 +244,29 @@ public abstract class PulsarMessagingMessageListenerAdapter<V> {
}
private boolean isMessageWithNoTypeInfo(Type parameterType) {
if (parameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) parameterType;
if (parameterType instanceof ParameterizedType parameterizedType) {
Type rawType = parameterizedType.getRawType();
if (rawType.equals(org.springframework.messaging.Message.class)) {
return parameterizedType.getActualTypeArguments()[0] instanceof WildcardType;
}
}
return parameterType.equals(org.springframework.messaging.Message.class); // could
// be
// Message
// without
// a
// generic
// type
return parameterType.equals(org.springframework.messaging.Message.class);
}
public boolean isSimpleExtraction() {
return this.simpleExtraction;
}
public boolean isConsumerRecords() {
return this.isConsumerRecords;
}
public boolean isHeaderFound() {
return this.headerFound;
}
public boolean isSpringMessage() {
return this.isSpringMessage;
}
}

15
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java
View File

@@ -45,22 +45,21 @@ public class PulsarRecordMessagingMessageListenerAdapter<V> extends PulsarMessag
@Override
public void received(Consumer<V> consumer, Message<V> record, @Nullable Acknowledgement acknowledgement) {
org.springframework.messaging.Message<?> message = null;
if (isConversionNeeded()) {
Object theRecord = record;
if (isHeaderFound() || isSpringMessage()) {
message = toMessagingMessage(record, consumer);
}
else {
// message = NULL_MESSAGE;
else if (isSimpleExtraction()) {
theRecord = record.getValue();
}
if (logger.isDebugEnabled()) {
this.logger.debug("Processing [" + message + "]");
}
try {
Object result = invokeHandler(record, message, consumer, acknowledgement);
if (result != null) {
// handleResult(result, record, message);
}
invokeHandler(theRecord, message, consumer, acknowledgement);
}
catch (Exception e) { // NOSONAR ex flow control
catch (Exception e) {
throw e;
}
}

57
spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* 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.support;
import java.util.Map;
import org.apache.pulsar.client.api.Message;
/**
* Implementation of {@link PulsarMessageHeaderMapper}.
*
* @author Soby Chacko
*/
public class DefaultPulsarMessageHeaderMapper implements PulsarMessageHeaderMapper {
@Override
public void toHeaders(Message<?> source, Map<String, Object> target) {
target.putAll(source.getProperties());
if (source.hasKey()) {
target.put(PulsarHeaders.KEY, source.getKey());
target.put(PulsarHeaders.KEY_BYTES, source.getKeyBytes());
}
if (source.hasOrderingKey()) {
target.put(PulsarHeaders.ORDERING_KEY, source.getOrderingKey());
}
if (source.hasIndex()) {
target.put(PulsarHeaders.INDEX, source.getIndex());
}
target.put(PulsarHeaders.MESSAGE_ID, source.getMessageId());
target.put(PulsarHeaders.BROKER_PUBLISH_TIME, source.getBrokerPublishTime());
target.put(PulsarHeaders.EVENT_TIME, source.getEventTime());
target.put(PulsarHeaders.MESSAGE_SIZE, source.size());
target.put(PulsarHeaders.PRODUCER_NAME, source.getProducerName());
target.put(PulsarHeaders.RAW_DATA, source.getData());
target.put(PulsarHeaders.PUBLISH_TIME, source.getPublishTime());
target.put(PulsarHeaders.REDELIVERY_COUNT, source.getRedeliveryCount());
target.put(PulsarHeaders.REPLICATED_FROM, source.getReplicatedFrom());
target.put(PulsarHeaders.SCHEMA_VERSION, source.getSchemaVersion());
target.put(PulsarHeaders.SEQUENCE_ID, source.getSequenceId());
target.put(PulsarHeaders.TOPIC_NAME, source.getTopicName());
}
}

116
spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java Normal file
View File

@@ -0,0 +1,116 @@
/*
* 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.support;
/**
* Pulsar specific message headers.
*
* @author Soby Chacko
*/
public abstract class PulsarHeaders {
/**
* The prefix for Pulsar headers.
*/
public static final String PREFIX = "pulsar_";
/**
* The prefix for the message.
*/
public static final String PULSAR_MESSAGE = PREFIX + "message_";
/**
* Prefix for the unique message id.
*/
public static final String MESSAGE_ID = PULSAR_MESSAGE + "id";
/**
* Prefix for the raw message data.
*/
public static final String RAW_DATA = PULSAR_MESSAGE + "raw_data";
/**
* Prefix for message size.
*/
public static final String MESSAGE_SIZE = PULSAR_MESSAGE + "size";
/**
* Prefix for message publish time.
*/
public static final String PUBLISH_TIME = PULSAR_MESSAGE + "publish_time";
/**
* Prefix for event time.
*/
public static final String EVENT_TIME = PULSAR_MESSAGE + "event_time";
/**
* Prefix for message sequence id.
*/
public static final String SEQUENCE_ID = PULSAR_MESSAGE + "sequence_id";
/**
* prefix for the producer name.
*/
public static final String PRODUCER_NAME = PULSAR_MESSAGE + "producer_name";
/**
* Prefix for the message key.
*/
public static final String KEY = PULSAR_MESSAGE + "key";
/**
* Prefix for the message key as bytes.
*/
public static final String KEY_BYTES = PULSAR_MESSAGE + "key_bytes";
/**
* Prefix for the order key.
*/
public static final String ORDERING_KEY = PULSAR_MESSAGE + "ordering_key";
/**
* Prefix for the topic name.
*/
public static final String TOPIC_NAME = PULSAR_MESSAGE + "topic_name";
/**
* Prefix for redelivery count.
*/
public static final String REDELIVERY_COUNT = PULSAR_MESSAGE + "redelivery_count";
/**
* Prefix for schema version.
*/
public static final String SCHEMA_VERSION = PULSAR_MESSAGE + "schema_version";
/**
* Prefix for the cluster replicated from.
*/
public static final String REPLICATED_FROM = PULSAR_MESSAGE + "replicated_from";
/**
* Prefix for broker publish time.
*/
public static final String BROKER_PUBLISH_TIME = PULSAR_MESSAGE + "broker_publish_time";
/**
* Prefix for index.
*/
public static final String INDEX = PULSAR_MESSAGE + "index";
}

38
spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java Normal file
View File

@@ -0,0 +1,38 @@
/*
* 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.support;
import java.util.Map;
import org.apache.pulsar.client.api.Message;
/**
* API for Pulsar message header mapper.
*
* @author Soby Chacko
*/
public interface PulsarMessageHeaderMapper {
/**
* Map from the given message metadata to a map of headers for the eventual
* {@link org.springframework.messaging.MessageHeaders}.
* @param source Pulsar message.
* @param target the target headers.
*/
void toHeaders(Message<?> source, Map<String, Object> target);
}

29
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java
View File

@@ -16,17 +16,17 @@
package org.springframework.pulsar.support.converter;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.pulsar.client.api.Consumer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.pulsar.support.PulsarMessageHeaderMapper;
/**
*
@@ -38,21 +38,22 @@ import org.springframework.messaging.support.MessageBuilder;
*/
public class PulsarMessagingMessageConverter<V> implements PulsarRecordMessageConverter<V> {
private final PulsarMessageHeaderMapper pulsarMessageHeaderMapper;
private SmartMessageConverter messagingConverter;
public PulsarMessagingMessageConverter(PulsarMessageHeaderMapper pulsarMessageHeaderMapper) {
this.pulsarMessageHeaderMapper = pulsarMessageHeaderMapper;
}
@Override
public Message<?> toMessage(org.apache.pulsar.client.api.Message<V> record, Consumer<V> consumer, Type type) {
Message<?> message = MessageBuilder.createMessage(extractAndConvertValue(record, type),
new MessageHeaders(Collections.emptyMap()));
if (this.messagingConverter != null) {
Class<?> clazz = type instanceof Class ? (Class<?>) type : type instanceof ParameterizedType
? (Class<?>) ((ParameterizedType) type).getRawType() : Object.class;
Object payload = this.messagingConverter.fromMessage(message, clazz, type);
if (payload != null) {
message = new GenericMessage<>(payload, message.getHeaders());
}
}
final Map<String, Object> messageHeaders = new HashMap<>();
this.pulsarMessageHeaderMapper.toHeaders(record, messageHeaders);
Message<?> message = MessageBuilder.createMessage(extractAndConvertValue(record),
new MessageHeaders(messageHeaders));
return message;
}
@@ -77,7 +78,7 @@ public class PulsarMessagingMessageConverter<V> implements PulsarRecordMessageCo
this.messagingConverter = messagingConverter;
}
protected Object extractAndConvertValue(org.apache.pulsar.client.api.Message<V> record, Type type) {
protected Object extractAndConvertValue(org.apache.pulsar.client.api.Message<V> record) {
return record.getValue();
}

233
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -19,6 +19,8 @@ package org.springframework.pulsar.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -27,9 +29,12 @@ import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Messages;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RedeliveryBackoff;
@@ -49,6 +54,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
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.PulsarListener;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
@@ -64,6 +70,7 @@ 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.support.PulsarHeaders;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@@ -610,4 +617,230 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
}
@Nested
@ContextConfiguration(classes = PulsarListenerTests.PulsarHeadersTest.PulsarListerWithHeadersConfig.class)
class PulsarHeadersTest {
static CountDownLatch simpleListenerLatch = new CountDownLatch(1);
static CountDownLatch pulsarMessageListenerLatch = new CountDownLatch(1);
static CountDownLatch springMessagingMessageListenerLatch = new CountDownLatch(1);
static volatile String capturedData;
static volatile MessageId messageId;
static volatile String topicName;
static volatile String fooValue;
static volatile byte[] rawData;
static CountDownLatch simpleBatchListenerLatch = new CountDownLatch(1);
static CountDownLatch pulsarMessageBatchListenerLatch = new CountDownLatch(1);
static CountDownLatch springMessagingMessageBatchListenerLatch = new CountDownLatch(1);
static CountDownLatch pulsarMessagesBatchListenerLatch = new CountDownLatch(1);
static volatile List<String> capturedBatchData;
static volatile List<MessageId> batchMessageIds;
static volatile List<String> batchTopicNames;
static volatile List<String> batchFooValues;
@Test
void simpleListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-simple-listener")
.withMessageCustomizer(
messageBuilder -> messageBuilder.property("foo", "simpleListenerWithHeaders"))
.withTopic("simpleListenerWithHeaders").send();
assertThat(simpleListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedData).isEqualTo("hello-simple-listener");
assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId);
assertThat(topicName).isEqualTo("persistent://public/default/simpleListenerWithHeaders");
assertThat(fooValue).isEqualTo("simpleListenerWithHeaders");
assertThat(rawData).isEqualTo("hello-simple-listener".getBytes(StandardCharsets.UTF_8));
}
@Test
void pulsarMessageListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-message-listener")
.withMessageCustomizer(
messageBuilder -> messageBuilder.property("foo", "pulsarMessageListenerWithHeaders"))
.withTopic("pulsarMessageListenerWithHeaders").send();
assertThat(pulsarMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedData).isEqualTo("hello-pulsar-message-listener");
assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId);
assertThat(topicName).isEqualTo("persistent://public/default/pulsarMessageListenerWithHeaders");
assertThat(fooValue).isEqualTo("pulsarMessageListenerWithHeaders");
assertThat(rawData).isEqualTo("hello-pulsar-message-listener".getBytes(StandardCharsets.UTF_8));
}
@Test
void springMessagingMessageListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-spring-messaging-message-listener")
.withMessageCustomizer(messageBuilder -> messageBuilder.property("foo",
"springMessagingMessageListenerWithHeaders"))
.withTopic("springMessagingMessageListenerWithHeaders").send();
assertThat(springMessagingMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedData).isEqualTo("hello-spring-messaging-message-listener");
assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId);
assertThat(topicName).isEqualTo("persistent://public/default/springMessagingMessageListenerWithHeaders");
assertThat(fooValue).isEqualTo("springMessagingMessageListenerWithHeaders");
assertThat(rawData).isEqualTo("hello-spring-messaging-message-listener".getBytes(StandardCharsets.UTF_8));
}
@Test
void simpleBatchListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-simple-batch-listener")
.withMessageCustomizer(
messageBuilder -> messageBuilder.property("foo", "simpleBatchListenerWithHeaders"))
.withTopic("simpleBatchListenerWithHeaders").send();
assertThat(simpleBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedBatchData).containsExactly("hello-simple-batch-listener");
assertThat(batchMessageIds).containsExactly(messageId);
assertThat(batchTopicNames).containsExactly("persistent://public/default/simpleBatchListenerWithHeaders");
assertThat(batchFooValues).containsExactly("simpleBatchListenerWithHeaders");
}
@Test
void pulsarMessageBatchListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-message-batch-listener")
.withMessageCustomizer(
messageBuilder -> messageBuilder.property("foo", "pulsarMessageBatchListenerWithHeaders"))
.withTopic("pulsarMessageBatchListenerWithHeaders").send();
assertThat(pulsarMessageBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedBatchData).containsExactly("hello-pulsar-message-batch-listener");
assertThat(batchTopicNames)
.containsExactly("persistent://public/default/pulsarMessageBatchListenerWithHeaders");
assertThat(batchFooValues).containsExactly("pulsarMessageBatchListenerWithHeaders");
assertThat(batchMessageIds).containsExactly(messageId);
}
@Test
void springMessagingMessageBatchListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-spring-messaging-message-batch-listener")
.withMessageCustomizer(messageBuilder -> messageBuilder.property("foo",
"springMessagingMessageBatchListenerWithHeaders"))
.withTopic("springMessagingMessageBatchListenerWithHeaders").send();
assertThat(springMessagingMessageBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedBatchData).containsExactly("hello-spring-messaging-message-batch-listener");
assertThat(batchTopicNames)
.containsExactly("persistent://public/default/springMessagingMessageBatchListenerWithHeaders");
assertThat(batchFooValues).containsExactly("springMessagingMessageBatchListenerWithHeaders");
assertThat(batchMessageIds).containsExactly(messageId);
}
@Test
void pulsarMessagesBatchListenerWithHeaders() throws Exception {
final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-messages-batch-listener")
.withMessageCustomizer(
messageBuilder -> messageBuilder.property("foo", "pulsarMessagesBatchListenerWithHeaders"))
.withTopic("pulsarMessagesBatchListenerWithHeaders").send();
assertThat(pulsarMessagesBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(capturedBatchData).containsExactly("hello-pulsar-messages-batch-listener");
assertThat(batchTopicNames)
.containsExactly("persistent://public/default/pulsarMessagesBatchListenerWithHeaders");
assertThat(batchFooValues).containsExactly("pulsarMessagesBatchListenerWithHeaders");
assertThat(batchMessageIds).containsExactly(messageId);
}
@EnablePulsar
@Configuration
static class PulsarListerWithHeadersConfig {
@PulsarListener(subscriptionName = "simple-listener-with-headers-sub", topics = "simpleListenerWithHeaders")
void simpleListenerWithHeaders(String data, @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
@Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData,
@Header("foo") String foo) {
capturedData = data;
PulsarHeadersTest.messageId = messageId;
PulsarHeadersTest.topicName = topicName;
fooValue = foo;
PulsarHeadersTest.rawData = rawData;
simpleListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub",
topics = "pulsarMessageListenerWithHeaders")
void pulsarMessageListenerWithHeaders(Message<String> data,
@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
@Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData,
@Header("foo") String foo) {
capturedData = data.getValue();
PulsarHeadersTest.messageId = messageId;
PulsarHeadersTest.topicName = topicName;
fooValue = foo;
PulsarHeadersTest.rawData = rawData;
pulsarMessageListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub",
topics = "springMessagingMessageListenerWithHeaders")
void springMessagingMessageListenerWithHeaders(org.springframework.messaging.Message<String> data,
@Header(PulsarHeaders.MESSAGE_ID) MessageId messageId,
@Header(PulsarHeaders.RAW_DATA) byte[] rawData, @Header(PulsarHeaders.TOPIC_NAME) String topicName,
@Header("foo") String foo) {
capturedData = data.getPayload();
PulsarHeadersTest.messageId = messageId;
PulsarHeadersTest.topicName = topicName;
fooValue = foo;
PulsarHeadersTest.rawData = rawData;
springMessagingMessageListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "simple-batch-listener-with-headers-sub",
topics = "simpleBatchListenerWithHeaders", batch = true)
void simpleBatchListenerWithHeaders(List<String> data,
@Header(PulsarHeaders.MESSAGE_ID) List<MessageId> messageIds,
@Header(PulsarHeaders.TOPIC_NAME) List<String> topicNames, @Header("foo") List<String> fooValues) {
capturedBatchData = data;
batchMessageIds = messageIds;
batchTopicNames = topicNames;
batchFooValues = fooValues;
simpleBatchListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "pulsarMessage-batch-listener-with-headers-sub",
topics = "pulsarMessageBatchListenerWithHeaders", batch = true)
void pulsarMessageBatchListenerWithHeaders(List<Message<String>> data,
@Header(PulsarHeaders.MESSAGE_ID) List<MessageId> messageIds,
@Header(PulsarHeaders.TOPIC_NAME) List<String> topicNames, @Header("foo") List<String> fooValues) {
capturedBatchData = data.stream().map(Message::getValue).collect(Collectors.toList());
batchMessageIds = messageIds;
batchTopicNames = topicNames;
batchFooValues = fooValues;
pulsarMessageBatchListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "spring-messaging-message-batch-listener-with-headers-sub",
topics = "springMessagingMessageBatchListenerWithHeaders", batch = true)
void springMessagingMessageBatchListenerWithHeaders(
List<org.springframework.messaging.Message<String>> data,
@Header(PulsarHeaders.MESSAGE_ID) List<MessageId> messageIds,
@Header(PulsarHeaders.TOPIC_NAME) List<String> topicNames, @Header("foo") List<String> fooValues) {
capturedBatchData = data.stream().map(org.springframework.messaging.Message::getPayload)
.collect(Collectors.toList());
batchMessageIds = messageIds;
batchTopicNames = topicNames;
batchFooValues = fooValues;
springMessagingMessageBatchListenerLatch.countDown();
}
@PulsarListener(subscriptionName = "pulsarMessages-batch-listener-with-headers-sub",
topics = "pulsarMessagesBatchListenerWithHeaders", batch = true)
void pulsarMessagesBatchListenerWithHeaders(Messages<String> data,
@Header(PulsarHeaders.MESSAGE_ID) List<MessageId> messageIds,
@Header(PulsarHeaders.TOPIC_NAME) List<String> topicNames, @Header("foo") List<String> fooValues) {
List<String> list = new ArrayList<>();
data.iterator().forEachRemaining(m -> list.add(m.getValue()));
capturedBatchData = list;
batchMessageIds = messageIds;
batchTopicNames = topicNames;
batchFooValues = fooValues;
pulsarMessagesBatchListenerLatch.countDown();
}
}
}
}