diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index 67d01488..8e5f489b 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -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 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 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 pulsarTemplate; + + private final Schema schema; + + private final String destination; + + private final ProducerBuilderCustomizer layeredProducerPropsCustomizer; + + private final PulsarHeaderMapper headerMapper; + private boolean running = true; - final PulsarTemplate pulsarTemplate; - - final Schema schema; - - final String destination; - - final ProducerBuilderCustomizer layeredProducerPropsCustomizer; - PulsarProducerConfigurationMessageHandler(PulsarTemplate pulsarTemplate, Schema schema, - String destination, ProducerBuilderCustomizer layeredProducerPropsCustomizer) { + String destination, ProducerBuilderCustomizer 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 applySpringHeadersAsPulsarProperties(MessageHeaders headers) { + return (mb) -> this.headerMapper.fromSpringHeaders(headers).forEach(mb::property); + } + } } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/config/PulsarBinderConfiguration.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/config/PulsarBinderConfiguration.java index e42b9261..537283c7 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/config/PulsarBinderConfiguration.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/config/PulsarBinderConfiguration.java @@ -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 pulsarTemplate, PulsarConsumerFactory 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; } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java index 20b40763..184ec64e 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java @@ -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> springMessageSupplier() { + return () -> { + msgCount++; + return MessageBuilder.withPayload("test-headers-msg-" + msgCount) + .setHeader("custom-id", "5150-" + msgCount).build(); + }; + } + + @Bean + public Consumer> springMessageLogger() { + return s -> this.logger.info("Hello binder: {} w/ custom-id: {}", s.getPayload(), + s.getHeaders().get("custom-id")); + } + + } + + } + @EnableAutoConfiguration @SpringBootConfiguration static class PrimitiveTextConfig { diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java index 6cbbabfc..27236a67 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java @@ -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 producerBinding = binder.bindProducer( - String.format("bar%s0", getDestinationNameDelimiter()), moduleOutputChannel, - producerBindingProperties.getProducer()); - Binding 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> inboundMessageRef = new AtomicReference<>(); - moduleInputChannel.subscribe(message1 -> { - try { - inboundMessageRef.set((Message) 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 diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java index bd9cc39f..1f602ebb 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java @@ -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.*)$") diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java index 9373abc0..b8c297d2 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java @@ -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) pulsarTemplate, - pulsarConsumerFactory, binderConfigProps, schemaResolver); + pulsarConsumerFactory, binderConfigProps, schemaResolver, headerMapper); var context = new AnnotationConfigApplicationContext(Config.class); setApplicationContext(context); binder.setApplicationContext(context); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/AbstractPulsarMessageToSpringMessageAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/AbstractPulsarMessageToSpringMessageAdapter.java index c8218db3..ee891d5c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/AbstractPulsarMessageToSpringMessageAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/AbstractPulsarMessageToSpringMessageAdapter.java @@ -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 { private boolean converterSet; private PulsarMessageConverter messageConverter = new PulsarRecordMessageConverter( - new DefaultPulsarMessageHeaderMapper()); + new DefaultPulsarHeaderMapper()); private Type fallbackType = Object.class; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapper.java new file mode 100644 index 00000000..7171d6f6 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapper.java @@ -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 fromSpringHeaders(MessageHeaders springHeaders) { + Objects.requireNonNull(springHeaders, "springHeaders must be specified"); + var pulsarHeaders = new LinkedHashMap(); + 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(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); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java deleted file mode 100644 index 5595d2da..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java +++ /dev/null @@ -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 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()); - } - -} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java new file mode 100644 index 00000000..29c2d961 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java @@ -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. + *

+ * NOTE:Pulsar does not have the concept of message headers, but rather message + * metadata. The terms "Pulsar message headers" and "Pulsar message + * metadata" 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 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); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java deleted file mode 100644 index 3a98331c..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java +++ /dev/null @@ -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 target); - -} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java index 087af7a5..52a57c09 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java @@ -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 message type * @author Soby Chacko + * @author Chris Bono */ public class PulsarRecordMessageConverter implements PulsarMessageConverter { - 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 record, Consumer consumer, Type type) { - - Map 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() { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapperTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapperTests.java new file mode 100644 index 00000000..42c0be6f --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapperTests.java @@ -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 mockPulsarMessage(boolean includeOptionalMetadata, Map userProperties) { + Message pulsarMessage = (Message) 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(); + 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(); + 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(); + 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 + } + + } + +} diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index a4a508ed..6e6785e9 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -6,11 +6,13 @@ - - + + + +