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

Binder preserves headers on send/receive (#359)

Browse Source 
- Refactor the existing PulsarMessageHeaderMapper to include to/from support.
- Use the new API in the binder send/receive
- Add tests for header mapper

See #358
This commit is contained in:
Chris Bono
2023-02-24 09:31:36 -06:00
committed by GitHub
parent e989f4e03b
commit 0458f01b3e
14 changed files with 481 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
View File

@@ -40,6 +40,7 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties;
import org.springframework.pulsar.autoconfigure.ProducerConfigProperties;
@@ -48,6 +49,7 @@ import org.springframework.pulsar.core.ProducerBuilderCustomizer;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.core.TypedMessageBuilderCustomizer;
import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -57,6 +59,7 @@ import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarCo
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarExtendedBindingProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarProducerProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
import org.springframework.pulsar.support.PulsarHeaderMapper;
/**
* {@link Binder} implementation for Apache Pulsar.
@@ -72,20 +75,24 @@ public class PulsarMessageChannelBinder extends
private final PulsarConsumerFactory<?> pulsarConsumerFactory;
private final PulsarBinderConfigurationProperties binderConfigProps;
private final SchemaResolver schemaResolver;
private PulsarBinderConfigurationProperties binderConfigProps;
private final PulsarHeaderMapper headerMapper;
private PulsarExtendedBindingProperties extendedBindingProperties = new PulsarExtendedBindingProperties();
public PulsarMessageChannelBinder(PulsarTopicProvisioner provisioningProvider,
PulsarTemplate<Object> pulsarTemplate, PulsarConsumerFactory<?> pulsarConsumerFactory,
PulsarBinderConfigurationProperties binderConfigProps, SchemaResolver schemaResolver) {
PulsarBinderConfigurationProperties binderConfigProps, SchemaResolver schemaResolver,
PulsarHeaderMapper headerMapper) {
super(null, provisioningProvider);
this.pulsarTemplate = pulsarTemplate;
this.pulsarConsumerFactory = pulsarConsumerFactory;
this.binderConfigProps = binderConfigProps;
this.schemaResolver = schemaResolver;
this.headerMapper = headerMapper;
}
@Override
@@ -110,7 +117,8 @@ public class PulsarMessageChannelBinder extends
binderProducerProps, bindingProducerProps);
var handler = new PulsarProducerConfigurationMessageHandler(this.pulsarTemplate, schema, destination.getName(),
(builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, mergedProducerProps));
(builder) -> ProducerBuilderConfigurationUtil.loadConf(builder, mergedProducerProps),
this.headerMapper);
handler.setApplicationContext(getApplicationContext());
handler.setBeanFactory(getBeanFactory());
@@ -125,9 +133,10 @@ public class PulsarMessageChannelBinder extends
var messageDrivenChannelAdapter = new PulsarMessageDrivenChannelAdapter();
containerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
Message<Object> message = MessageBuilder.withPayload(msg.getValue()).build();
var message = MessageBuilder.createMessage(msg.getValue(), this.headerMapper.toSpringHeaders(msg));
messageDrivenChannelAdapter.send(message);
});
if (properties.isUseNativeDecoding()) {
var schema = resolveSchema(properties.getExtension().getSchemaType(),
properties.getExtension().getMessageType(), properties.getExtension().getMessageKeyType(),
@@ -243,22 +252,26 @@ public class PulsarMessageChannelBinder extends
static class PulsarProducerConfigurationMessageHandler extends AbstractMessageProducingHandler
implements ManageableLifecycle {
private final PulsarTemplate<Object> pulsarTemplate;
private final Schema<Object> schema;
private final String destination;
private final ProducerBuilderCustomizer<Object> layeredProducerPropsCustomizer;
private final PulsarHeaderMapper headerMapper;
private boolean running = true;
final PulsarTemplate<Object> pulsarTemplate;
final Schema<Object> schema;
final String destination;
final ProducerBuilderCustomizer<Object> layeredProducerPropsCustomizer;
PulsarProducerConfigurationMessageHandler(PulsarTemplate<Object> pulsarTemplate, Schema<Object> schema,
String destination, ProducerBuilderCustomizer<Object> layeredProducerPropsCustomizer) {
String destination, ProducerBuilderCustomizer<Object> layeredProducerPropsCustomizer,
PulsarHeaderMapper headerMapper) {
this.pulsarTemplate = pulsarTemplate;
this.schema = schema;
this.destination = destination;
this.layeredProducerPropsCustomizer = layeredProducerPropsCustomizer;
this.headerMapper = headerMapper;
}
@Override
@@ -286,14 +299,24 @@ public class PulsarMessageChannelBinder extends
@Override
protected void handleMessageInternal(Message<?> message) {
try {
this.pulsarTemplate.newMessage(message.getPayload()).withTopic(this.destination).withSchema(this.schema)
.withProducerCustomizer(this.layeredProducerPropsCustomizer).sendAsync();
// @formatter:off
this.pulsarTemplate.newMessage(message.getPayload())
.withTopic(this.destination)
.withSchema(this.schema)
.withProducerCustomizer(this.layeredProducerPropsCustomizer)
.withMessageCustomizer(this.applySpringHeadersAsPulsarProperties(message.getHeaders()))
.sendAsync();
// @formatter:on
}
catch (PulsarClientException ex) {
logger.trace(ex, "Failed to send message to destination: " + this.destination);
}
}
private TypedMessageBuilderCustomizer<Object> applySpringHeadersAsPulsarProperties(MessageHeaders headers) {
return (mb) -> this.headerMapper.fromSpringHeaders(headers).forEach(mb::property);
}
}
}

11
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/config/PulsarBinderConfiguration.java
View File

@@ -30,6 +30,8 @@ import org.springframework.pulsar.spring.cloud.stream.binder.PulsarMessageChanne
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBinderConfigurationProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarExtendedBindingProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
import org.springframework.pulsar.support.DefaultPulsarHeaderMapper;
import org.springframework.pulsar.support.PulsarHeaderMapper;
/**
* Pulsar binder {@link Configuration}.
@@ -48,13 +50,18 @@ public class PulsarBinderConfiguration {
return new PulsarTopicProvisioner(pulsarAdministration, pulsarBinderConfigurationProperties);
}
@Bean
public PulsarHeaderMapper pulsarHeaderMapper() {
return new DefaultPulsarHeaderMapper();
}
@Bean
public PulsarMessageChannelBinder pulsarMessageChannelBinder(PulsarTopicProvisioner pulsarTopicProvisioner,
PulsarTemplate<Object> pulsarTemplate, PulsarConsumerFactory<byte[]> pulsarConsumerFactory,
PulsarBinderConfigurationProperties binderConfigProps, PulsarExtendedBindingProperties bindingConfigProps,
SchemaResolver schemaResolver) {
SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
PulsarMessageChannelBinder pulsarMessageChannelBinder = new PulsarMessageChannelBinder(pulsarTopicProvisioner,
pulsarTemplate, pulsarConsumerFactory, binderConfigProps, schemaResolver);
pulsarTemplate, pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper);
pulsarMessageChannelBinder.setExtendedBindingProperties(bindingConfigProps);
return pulsarMessageChannelBinder;
}

48
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java
View File

@@ -49,6 +49,8 @@ import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.pulsar.autoconfigure.PulsarProperties;
import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
@@ -278,6 +280,52 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport {
}
@Nested
class CustomMessageHeaders {
@Test
void headersPropagatedSendAndReceive(CapturedOutput output) {
SpringApplication app = new SpringApplication(CustomHeadersConfig.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(),
"--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(),
"--spring.cloud.function.definition=springMessageSupplier;springMessageLogger",
"--spring.cloud.stream.bindings.springMessageLogger-in-0.destination=springMessageSupplier-out-0",
"--spring.cloud.stream.pulsar.bindings.springMessageLogger-in-0.consumer.subscription-name=pbit-cmh-sub1")) {
// Wait for a few of the messages to flow through (check for index = 5)
Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)).until(
() -> output.toString().contains("Hello binder: test-headers-msg-5 w/ custom-id: 5150-5"));
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
static class CustomHeadersConfig {
private final Logger logger = LoggerFactory.getLogger(getClass());
private int msgCount = 0;
@Bean
public Supplier<Message<String>> springMessageSupplier() {
return () -> {
msgCount++;
return MessageBuilder.withPayload("test-headers-msg-" + msgCount)
.setHeader("custom-id", "5150-" + msgCount).build();
};
}
@Bean
public Consumer<Message<String>> springMessageLogger() {
return s -> this.logger.info("Hello binder: {} w/ custom-id: {}", s.getPayload(),
s.getHeaders().get("custom-id"));
}
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
static class PrimitiveTextConfig {

52
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java
View File

@@ -58,6 +58,7 @@ import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBi
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarProducerProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
import org.springframework.pulsar.support.DefaultPulsarHeaderMapper;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
@@ -110,7 +111,7 @@ public class PulsarBinderTests extends
var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config);
if (this.binder == null) {
this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps,
new DefaultSchemaResolver());
new DefaultSchemaResolver(), new DefaultPulsarHeaderMapper());
}
return this.binder;
}
@@ -245,55 +246,6 @@ public class PulsarBinderTests extends
binding2.unbind();
}
@Test
@Override
@SuppressWarnings({ "unchecked", "rawtypes" })
public void testSendAndReceiveNoOriginalContentType(TestInfo testInfo) throws Exception {
Binder binder = getBinder();
BindingProperties producerBindingProperties = createProducerBindingProperties(
createProducerProperties(testInfo));
DirectChannel moduleOutputChannel = createBindableChannel("output", producerBindingProperties);
BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties());
DirectChannel moduleInputChannel = createBindableChannel("input", inputBindingProperties);
Binding<MessageChannel> producerBinding = binder.bindProducer(
String.format("bar%s0", getDestinationNameDelimiter()), moduleOutputChannel,
producerBindingProperties.getProducer());
Binding<MessageChannel> consumerBinding = binder.bindConsumer(
String.format("bar%s0", getDestinationNameDelimiter()), "testSendAndReceiveNoOriginalContentType",
moduleInputChannel, createConsumerProperties());
binderBindUnbindLatency();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
moduleOutputChannel.send(message);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<Message<byte[]>> inboundMessageRef = new AtomicReference<>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.set((Message<byte[]>) message1);
}
finally {
latch.countDown();
}
});
moduleOutputChannel.send(message);
Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
assertThat(inboundMessageRef.get()).isNotNull();
assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo".getBytes());
// TODO: The below content-type should be TEXT_PLAIN, but default to
// application/json
// This is because we don't currently preserve any message headers on send. We
// should look into this soon.
// Also, the content-type is "application/json" (with double quotes). We will have
// to fix that as well.
assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE).toString())
.contains(MimeTypeUtils.APPLICATION_JSON.toString());
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
@Override
@Disabled

3
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java
View File

@@ -39,6 +39,7 @@ import org.springframework.pulsar.core.Resolved;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBinderConfigurationProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
import org.springframework.pulsar.support.DefaultPulsarHeaderMapper;
/**
* Unit tests for {@link PulsarMessageChannelBinder#resolveSchema}.
@@ -52,7 +53,7 @@ public class PulsarMessageChannelBinderResolveSchemaTests {
@SuppressWarnings("unchecked")
private PulsarMessageChannelBinder binder = new PulsarMessageChannelBinder(mock(PulsarTopicProvisioner.class),
mock(PulsarTemplate.class), mock(PulsarConsumerFactory.class),
mock(PulsarBinderConfigurationProperties.class), resolver);
mock(PulsarBinderConfigurationProperties.class), resolver, new DefaultPulsarHeaderMapper());
@ParameterizedTest
@EnumSource(mode = Mode.MATCH_NONE, names = "^(AUTO.*|AVRO|JSON|KEY_VALUE|NONE|PROTOBUF.*)$")

5
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java
View File

@@ -24,6 +24,7 @@ import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBinderConfigurationProperties;
import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner;
import org.springframework.pulsar.support.PulsarHeaderMapper;
/**
* Test binder to exercise producer/consumer bindings in
@@ -37,11 +38,11 @@ public class PulsarTestBinder extends AbstractPulsarTestBinder {
@SuppressWarnings({ "unchecked" })
PulsarTestBinder(PulsarTopicProvisioner pulsarTopicProvisioner, PulsarTemplate<?> pulsarTemplate,
PulsarConsumerFactory<?> pulsarConsumerFactory, PulsarBinderConfigurationProperties binderConfigProps,
SchemaResolver schemaResolver) {
SchemaResolver schemaResolver, PulsarHeaderMapper headerMapper) {
try {
var binder = new PulsarMessageChannelBinder(pulsarTopicProvisioner, (PulsarTemplate<Object>) pulsarTemplate,
pulsarConsumerFactory, binderConfigProps, schemaResolver);
pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper);
var context = new AnnotationConfigApplicationContext(Config.class);
setApplicationContext(context);
binder.setApplicationContext(context);

4
spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/AbstractPulsarMessageToSpringMessageAdapter.java
View File

@@ -38,7 +38,7 @@ 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.support.DefaultPulsarMessageHeaderMapper;
import org.springframework.pulsar.support.DefaultPulsarHeaderMapper;
import org.springframework.pulsar.support.converter.PulsarMessageConverter;
import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter;
import org.springframework.util.Assert;
@@ -82,7 +82,7 @@ public abstract class AbstractPulsarMessageToSpringMessageAdapter<V> {
private boolean converterSet;
private PulsarMessageConverter<V> messageConverter = new PulsarRecordMessageConverter<V>(
new DefaultPulsarMessageHeaderMapper());
new DefaultPulsarHeaderMapper());
private Type fallbackType = Object.class;

73
spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapper.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright 2022-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import org.apache.pulsar.client.api.Message;
import org.springframework.messaging.MessageHeaders;
/**
* Default implementation of {@link PulsarHeaderMapper}.
*
* @author Soby Chacko
* @author Chris Bono
*/
public class DefaultPulsarHeaderMapper implements PulsarHeaderMapper {
@Override
public Map<String, String> fromSpringHeaders(MessageHeaders springHeaders) {
Objects.requireNonNull(springHeaders, "springHeaders must be specified");
var pulsarHeaders = new LinkedHashMap<String, String>();
springHeaders.forEach((k, v) -> pulsarHeaders.put(k, Objects.toString(v, null)));
return pulsarHeaders;
}
@Override
public MessageHeaders toSpringHeaders(Message<?> pulsarMessage) {
Objects.requireNonNull(pulsarMessage, "pulsarMessage must be specified");
var headersMap = new HashMap<String, Object>(pulsarMessage.getProperties());
if (pulsarMessage.hasKey()) {
headersMap.put(PulsarHeaders.KEY, pulsarMessage.getKey());
headersMap.put(PulsarHeaders.KEY_BYTES, pulsarMessage.getKeyBytes());
}
if (pulsarMessage.hasOrderingKey()) {
headersMap.put(PulsarHeaders.ORDERING_KEY, pulsarMessage.getOrderingKey());
}
if (pulsarMessage.hasIndex()) {
headersMap.put(PulsarHeaders.INDEX, pulsarMessage.getIndex());
}
headersMap.put(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId());
headersMap.put(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime());
headersMap.put(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime());
headersMap.put(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size());
headersMap.put(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName());
headersMap.put(PulsarHeaders.RAW_DATA, pulsarMessage.getData());
headersMap.put(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime());
headersMap.put(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount());
headersMap.put(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom());
headersMap.put(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion());
headersMap.put(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId());
headersMap.put(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName());
return new MessageHeaders(headersMap);
}
}

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

@@ -1,57 +0,0 @@
/*
* Copyright 2022-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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());
}
}

52
spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2022-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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;
import org.springframework.messaging.MessageHeaders;
/**
* Defines the contract for mapping Spring Messaging {@link MessageHeaders} to and from
* Pulsar message headers.
* <p>
* <b>NOTE:</b>Pulsar does not have the concept of message headers, but rather message
* metadata. The terms &quot;Pulsar message headers&quot; and &quot;Pulsar message
* metadata&quot; are used interchangeably.
*
* @author Soby Chacko
* @author Chris Bono
*/
public interface PulsarHeaderMapper {
/**
* Map from the given Spring Messaging headers to Pulsar message headers.
* @param springHeaders the Spring messaging headers
* @return map of Pulsar message headers or an empty map for no headers.
*/
Map<String, String> fromSpringHeaders(MessageHeaders springHeaders);
/**
* Map the headers from the given Pulsar message to Spring Messaging headers.
* @param pulsarMessage the Pulsar message containing the headers to map
* @return the Spring Messaging headers
*/
MessageHeaders toSpringHeaders(Message<?> pulsarMessage);
}

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

@@ -1,38 +0,0 @@
/*
* Copyright 2022-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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);
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java
View File

@@ -17,16 +17,13 @@
package org.springframework.pulsar.support.converter;
import java.lang.reflect.Type;
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.MessageBuilder;
import org.springframework.pulsar.support.PulsarMessageHeaderMapper;
import org.springframework.pulsar.support.PulsarHeaderMapper;
/**
*
@@ -35,26 +32,21 @@ import org.springframework.pulsar.support.PulsarMessageHeaderMapper;
*
* @param <V> message type
* @author Soby Chacko
* @author Chris Bono
*/
public class PulsarRecordMessageConverter<V> implements PulsarMessageConverter<V> {
private final PulsarMessageHeaderMapper pulsarMessageHeaderMapper;
private final PulsarHeaderMapper headerMapper;
private SmartMessageConverter messagingConverter;
public PulsarRecordMessageConverter(PulsarMessageHeaderMapper pulsarMessageHeaderMapper) {
this.pulsarMessageHeaderMapper = pulsarMessageHeaderMapper;
public PulsarRecordMessageConverter(PulsarHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
}
@Override
public Message<?> toMessage(org.apache.pulsar.client.api.Message<V> record, Consumer<V> consumer, Type type) {
Map<String, Object> messageHeaders = new HashMap<>();
this.pulsarMessageHeaderMapper.toHeaders(record, messageHeaders);
Message<?> message = MessageBuilder.createMessage(extractAndConvertValue(record),
new MessageHeaders(messageHeaders));
return message;
return MessageBuilder.createMessage(extractAndConvertValue(record), this.headerMapper.toSpringHeaders(record));
}
protected org.springframework.messaging.converter.MessageConverter getMessagingConverter() {

242
spring-pulsar/src/test/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapperTests.java Normal file
View File

@@ -0,0 +1,242 @@
/*
* Copyright 2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* 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 static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNullPointerException;
import static org.assertj.core.api.AssertionsForClassTypes.entry;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.springframework.messaging.MessageHeaders;
/**
* Tests for {@link DefaultPulsarHeaderMapper}.
*
* @author Chris Bono
*/
class DefaultPulsarHeaderMapperTests {
private DefaultPulsarHeaderMapper mapper = new DefaultPulsarHeaderMapper();
@SuppressWarnings("unchecked")
private Message<String> mockPulsarMessage(boolean includeOptionalMetadata, Map<String, String> userProperties) {
Message<String> pulsarMessage = (Message<String>) mock(Message.class);
// custom user properties
when(pulsarMessage.getProperties()).thenReturn(userProperties);
// optional metadata
when(pulsarMessage.hasKey()).thenReturn(includeOptionalMetadata);
when(pulsarMessage.getKey()).thenReturn("key");
when(pulsarMessage.getKeyBytes()).thenReturn("key".getBytes());
when(pulsarMessage.hasOrderingKey()).thenReturn(includeOptionalMetadata);
when(pulsarMessage.getOrderingKey()).thenReturn("orderingKey".getBytes());
when(pulsarMessage.hasIndex()).thenReturn(includeOptionalMetadata);
when(pulsarMessage.getIndex()).thenReturn(Optional.of(1L));
// required metadata
MessageId messageId = mock(MessageId.class);
when(pulsarMessage.getMessageId()).thenReturn(messageId);
when(pulsarMessage.getBrokerPublishTime()).thenReturn(Optional.of(100L));
when(pulsarMessage.getEventTime()).thenReturn(200L);
when(pulsarMessage.size()).thenReturn(300);
when(pulsarMessage.getProducerName()).thenReturn("producerName");
when(pulsarMessage.getData()).thenReturn("data".getBytes());
when(pulsarMessage.getPublishTime()).thenReturn(400L);
when(pulsarMessage.getRedeliveryCount()).thenReturn(500);
when(pulsarMessage.getReplicatedFrom()).thenReturn("replicatedFrom");
when(pulsarMessage.getSchemaVersion()).thenReturn("schemaVersion".getBytes());
when(pulsarMessage.getSequenceId()).thenReturn(600L);
when(pulsarMessage.getTopicName()).thenReturn("topicName");
return pulsarMessage;
}
@Nested
class FromSpringHeaders {
@Test
void nullSpringHeaders() {
assertThatNullPointerException().isThrownBy(() -> mapper.fromSpringHeaders(null))
.withMessage("springHeaders must be specified");
}
@Test
void emptySpringHeaders() {
assertThat(mapper.fromSpringHeaders(new MessageHeaders(Collections.emptyMap()))).containsOnlyKeys("id",
"timestamp");
}
@Test
void springHeadersWithNullValue() {
var headers = new HashMap<String, Object>();
headers.put("foo", "bar");
headers.put("uuid", null);
assertThat(mapper.fromSpringHeaders(new MessageHeaders(headers)))
.containsOnlyKeys("id", "timestamp", "foo", "uuid")
.contains(entry("foo", "bar"), entry("uuid", null));
}
@Test
void springHeadersWithValues() {
var uuid = UUID.randomUUID();
var headers = new HashMap<String, Object>();
headers.put("foo", "bar");
headers.put("uuid", uuid);
assertThat(mapper.fromSpringHeaders(new MessageHeaders(headers)))
.containsOnlyKeys("id", "timestamp", "foo", "uuid")
.contains(entry("foo", "bar"), entry("uuid", uuid.toString()));
}
}
@Nested
class ToSpringHeaders {
@Test
void nullPulsarMessage() {
assertThatNullPointerException().isThrownBy(() -> mapper.toSpringHeaders(null))
.withMessage("pulsarMessage must be specified");
}
@Test
void pulsarMessageWithOnlyRequiredMetadata() {
var pulsarMessage = mockPulsarMessage(false, Collections.emptyMap());
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
// @formatter:off
assertThat(springHeaders).contains(
entry(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId()),
entry(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime()),
entry(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime()),
entry(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size()),
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
entry(PulsarHeaders.RAW_DATA, pulsarMessage.getData()),
entry(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime()),
entry(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount()),
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
entry(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion()),
entry(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId()),
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
assertThat(springHeaders).doesNotContainKeys(
PulsarHeaders.KEY,
PulsarHeaders.KEY_BYTES,
PulsarHeaders.ORDERING_KEY,
PulsarHeaders.INDEX);
// @formatter:on
}
@Test
void pulsarMessageWithAllMetadataAndUserProperties() {
var uuid = UUID.randomUUID();
var customHeaders = new HashMap<String, String>();
customHeaders.put("foo", "bar");
customHeaders.put("uuid", uuid.toString());
var pulsarMessage = mockPulsarMessage(true, customHeaders);
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
// @formatter:off
assertThat(springHeaders).contains(
entry("foo", "bar"),
entry("uuid", uuid.toString()),
entry(PulsarHeaders.KEY, pulsarMessage.getKey()),
entry(PulsarHeaders.KEY_BYTES, pulsarMessage.getKeyBytes()),
entry(PulsarHeaders.ORDERING_KEY, pulsarMessage.getOrderingKey()),
entry(PulsarHeaders.INDEX, pulsarMessage.getIndex()),
entry(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId()),
entry(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime()),
entry(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime()),
entry(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size()),
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
entry(PulsarHeaders.RAW_DATA, pulsarMessage.getData()),
entry(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime()),
entry(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount()),
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
entry(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion()),
entry(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId()),
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
// @formatter:on
}
}
@Nested
class ToAndFromSpringHeaders {
@Test
void pulsarMessageWithOnlyRequiredMetadataRoundTripped() {
var pulsarMessage = mockPulsarMessage(false, Collections.emptyMap());
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
var pulsarHeaders = mapper.fromSpringHeaders(springHeaders);
// @formatter:off
assertThat(pulsarHeaders).contains(
entry(PulsarHeaders.MESSAGE_ID, Objects.toString(pulsarMessage.getMessageId())),
entry(PulsarHeaders.BROKER_PUBLISH_TIME, Objects.toString(pulsarMessage.getBrokerPublishTime())),
entry(PulsarHeaders.EVENT_TIME, Objects.toString(pulsarMessage.getEventTime())),
entry(PulsarHeaders.MESSAGE_SIZE, Objects.toString(pulsarMessage.size())),
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
entry(PulsarHeaders.RAW_DATA, Objects.toString(pulsarMessage.getData())),
entry(PulsarHeaders.PUBLISH_TIME, Objects.toString(pulsarMessage.getPublishTime())),
entry(PulsarHeaders.REDELIVERY_COUNT, Objects.toString(pulsarMessage.getRedeliveryCount())),
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
entry(PulsarHeaders.SCHEMA_VERSION, Objects.toString(pulsarMessage.getSchemaVersion())),
entry(PulsarHeaders.SEQUENCE_ID, Objects.toString(pulsarMessage.getSequenceId())),
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
// @formatter:on
}
@Test
void pulsarMessageWithAllMetadataAndUserPropertiesRoundTripped() {
var pulsarMessage = mockPulsarMessage(true, Collections.singletonMap("foo", "bar"));
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
var pulsarHeaders = mapper.fromSpringHeaders(springHeaders);
// @formatter:off
assertThat(pulsarHeaders).contains(
entry("foo", "bar"),
entry(PulsarHeaders.KEY, pulsarMessage.getKey()),
entry(PulsarHeaders.KEY_BYTES, Objects.toString(pulsarMessage.getKeyBytes())),
entry(PulsarHeaders.ORDERING_KEY, Objects.toString(pulsarMessage.getOrderingKey())),
entry(PulsarHeaders.INDEX, Objects.toString(pulsarMessage.getIndex())),
entry(PulsarHeaders.MESSAGE_ID, Objects.toString(pulsarMessage.getMessageId())),
entry(PulsarHeaders.BROKER_PUBLISH_TIME, Objects.toString(pulsarMessage.getBrokerPublishTime())),
entry(PulsarHeaders.EVENT_TIME, Objects.toString(pulsarMessage.getEventTime())),
entry(PulsarHeaders.MESSAGE_SIZE, Objects.toString(pulsarMessage.size())),
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
entry(PulsarHeaders.RAW_DATA, Objects.toString(pulsarMessage.getData())),
entry(PulsarHeaders.PUBLISH_TIME, Objects.toString(pulsarMessage.getPublishTime())),
entry(PulsarHeaders.REDELIVERY_COUNT, Objects.toString(pulsarMessage.getRedeliveryCount())),
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
entry(PulsarHeaders.SCHEMA_VERSION, Objects.toString(pulsarMessage.getSchemaVersion())),
entry(PulsarHeaders.SEQUENCE_ID, Objects.toString(pulsarMessage.getSequenceId())),
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
// @formatter:on
}
}
}

6
src/checkstyle/checkstyle-suppressions.xml
View File

@@ -6,11 +6,13 @@
<suppress files="package-info\.java" checks=".*" />
<suppress files="[\\/]test[\\/]" checks="RequireThis" />
<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp" />
<suppress files="PulsarMessageChannelBinderResolveSchemaTests" checks="AvoidStaticImport" />
<suppress files="DefaultPulsarHeaderMapperTests" checks="AvoidStaticImport" />
<suppress files="DefaultSchemaResolverTests" checks="AvoidStaticImport|MethodParamPad" />
<suppress files="DefaultTopicResolverTests" checks="AvoidStaticImport" />
<suppress files="PulsarBinderUtilsTest" checks="AvoidStaticImport" />
<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp" />
<suppress files="PulsarMessageChannelBinderResolveSchemaTests" checks="AvoidStaticImport" />
<suppress files="Proto" checks=".*"/>
<suppress files="ReactiveSpringPulsarBootApp" checks="HideUtilityClassConstructor"/>
<suppress files="[\\/]spring-pulsar-docs[\\/]" checks="JavadocPackage|JavadocType|JavadocVariable|SpringDeprecatedCheck" />