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 8e5f489b..fc0428c4 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 @@ -59,7 +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; +import org.springframework.pulsar.support.header.PulsarHeaderMapper; /** * {@link Binder} implementation for Apache Pulsar. @@ -314,7 +314,7 @@ public class PulsarMessageChannelBinder extends } private TypedMessageBuilderCustomizer applySpringHeadersAsPulsarProperties(MessageHeaders headers) { - return (mb) -> this.headerMapper.fromSpringHeaders(headers).forEach(mb::property); + return (mb) -> this.headerMapper.toPulsarHeaders(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 537283c7..e41c7090 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,8 +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; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper; +import org.springframework.pulsar.support.header.PulsarHeaderMapper; /** * Pulsar binder {@link Configuration}. @@ -52,7 +52,7 @@ public class PulsarBinderConfiguration { @Bean public PulsarHeaderMapper pulsarHeaderMapper() { - return new DefaultPulsarHeaderMapper(); + return JsonPulsarHeaderMapper.builder().build(); } @Bean 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 27236a67..630ddfb2 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,7 +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.support.header.JsonPulsarHeaderMapper; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; import org.springframework.util.Assert; import org.springframework.util.MimeTypeUtils; @@ -111,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 DefaultPulsarHeaderMapper()); + new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build()); } return this.binder; } 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 1f602ebb..8b5db905 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,7 +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; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper; /** * Unit tests for {@link PulsarMessageChannelBinder#resolveSchema}. @@ -53,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, new DefaultPulsarHeaderMapper()); + mock(PulsarBinderConfigurationProperties.class), resolver, JsonPulsarHeaderMapper.builder().build()); @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 b8c297d2..7b01b20f 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,7 +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; +import org.springframework.pulsar.support.header.PulsarHeaderMapper; /** * Test binder to exercise producer/consumer bindings in 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 ee891d5c..29601565 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,9 +38,9 @@ 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.DefaultPulsarHeaderMapper; import org.springframework.pulsar.support.converter.PulsarMessageConverter; import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper; import org.springframework.util.Assert; /** @@ -82,7 +82,7 @@ public abstract class AbstractPulsarMessageToSpringMessageAdapter { private boolean converterSet; private PulsarMessageConverter messageConverter = new PulsarRecordMessageConverter( - new DefaultPulsarHeaderMapper()); + JsonPulsarHeaderMapper.builder().build()); 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 deleted file mode 100644 index 7171d6f6..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapper.java +++ /dev/null @@ -1,73 +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.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/converter/PulsarRecordMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarRecordMessageConverter.java index 52a57c09..b167c331 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 @@ -23,7 +23,7 @@ import org.apache.pulsar.client.api.Consumer; import org.springframework.messaging.Message; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.pulsar.support.PulsarHeaderMapper; +import org.springframework.pulsar.support.header.PulsarHeaderMapper; /** * diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapper.java new file mode 100644 index 00000000..158cc82c --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapper.java @@ -0,0 +1,284 @@ +/* + * Copyright 2018-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.header; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.function.BiConsumer; +import java.util.function.Supplier; + +import org.apache.pulsar.client.api.Message; + +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.support.PulsarHeaders; +import org.springframework.pulsar.support.header.PulsarHeaderMatcher.NeverMatch; +import org.springframework.pulsar.support.header.PulsarHeaderMatcher.PatternMatch; + +/** + * Base {@code PulsarHeaderMapper} implementation that constrains which headers are mapped + * via {@link PulsarHeaderMatcher header matchers}. + * + *

+ * Concrete implementations only need to specify how to convert a header value to and from + * Spring Messaging and Pulsar, by implementing the abstract {@link #toPulsarHeaderValue} + * and {@link #toSpringHeaderValue} methods. + * + * @param type of context object used in the + * {@link #toPulsarHeaders} API + * @param type of context object used in the + * {@link #toSpringHeaders} API + * @author Chris Bono + */ +public abstract class AbstractPulsarHeaderMapper + implements PulsarHeaderMapper { + + protected final LogAccessor logger = new LogAccessor(this.getClass()); + + private final List inboundMatchers = new ArrayList<>(); + + private final List outboundMatchers = new ArrayList<>(); + + /** + * Construct a mapper that will match the supplied inbound and outbound patterns. + *

+ * NOTE: Internal framework headers are never mapped + * outbound. By default, the {@code "id"} and {@code "timestamp"} headers are also + * excluded from outbound mapping but can be included by adding them to + * {@code outboundPatterns}. + *

+ * NOTE: The patterns are applied in order, stopping on the first + * match (positive or negative). When no pattern is specified, the {@code "*"} pattern + * is added last. However, once a pattern is specified, the {@code "*"} is not added + * and must be added to the specified patterns if desired. + * @param inboundPatterns the inbound patterns to match - or empty to match all + * @param outboundPatterns the outbound patterns to match - or empty to match all + * (except internal framework headers and id/timestamp) + * + * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String) + */ + public AbstractPulsarHeaderMapper(List inboundPatterns, List outboundPatterns) { + Objects.requireNonNull(inboundPatterns, "inboundPatterns must be specified"); + Objects.requireNonNull(outboundPatterns, "outboundPatterns must be specified"); + inboundPatterns.forEach((p) -> this.inboundMatchers.add(PatternMatch.fromPatternString(p))); + // @formatter:off + this.outboundMatchers.add(new NeverMatch( + PulsarHeaders.KEY, + PulsarHeaders.KEY_BYTES, + PulsarHeaders.ORDERING_KEY, + PulsarHeaders.INDEX, + PulsarHeaders.MESSAGE_ID, + PulsarHeaders.BROKER_PUBLISH_TIME, + PulsarHeaders.EVENT_TIME, + PulsarHeaders.MESSAGE_SIZE, + PulsarHeaders.PRODUCER_NAME, + PulsarHeaders.RAW_DATA, + PulsarHeaders.PUBLISH_TIME, + PulsarHeaders.REDELIVERY_COUNT, + PulsarHeaders.REPLICATED_FROM, + PulsarHeaders.SCHEMA_VERSION, + PulsarHeaders.SEQUENCE_ID, + PulsarHeaders.TOPIC_NAME)); + // @formatter:on + if (outboundPatterns.isEmpty()) { + this.outboundMatchers.add(PatternMatch.fromPatternString("!id")); + this.outboundMatchers.add(PatternMatch.fromPatternString("!timestamp")); + this.outboundMatchers.add(PatternMatch.fromPatternString("*")); + } + else { + outboundPatterns.forEach((p) -> this.outboundMatchers.add(PatternMatch.fromPatternString(p))); + this.outboundMatchers.add(PatternMatch.fromPatternString("!id")); + this.outboundMatchers.add(PatternMatch.fromPatternString("!timestamp")); + } + } + + @Override + public Map toPulsarHeaders(MessageHeaders springHeaders) { + Objects.requireNonNull(springHeaders, "springHeaders must not be null"); + var pulsarHeaders = new LinkedHashMap(); + var context = toPulsarHeadersOnStarted(springHeaders); + springHeaders.forEach((key, rawValue) -> { + if (matchesForOutbound(key)) { + var value = toPulsarHeaderValue(key, rawValue, context); + pulsarHeaders.put(key, value); + } + }); + toPulsarHeadersOnCompleted(springHeaders, pulsarHeaders, context); + return pulsarHeaders; + } + + /** + * Called at the beginning of every {@link #toPulsarHeaders} invocation and the + * returned value is passed into each {@link #toPulsarHeaderValue} invocation as well + * as the final {@link #toPulsarHeadersOnCompleted}. Allows concrete implementations + * the ability to create an arbitrary context object and have it passed through the + * mapping process. + * @param springHeaders the Spring Messaging headers that are being converted + * @return optional context to pass through the mapping invocation + */ + @Nullable + protected ToPulsarHeadersContextType toPulsarHeadersOnStarted(MessageHeaders springHeaders) { + return null; + } + + /** + * Determine the Pulsar header value to use for a Spring Messaging header. + * @param name the Spring Messaging header name + * @param value the Spring Messaging header value + * @param context the optional context used for the mapping invocation + * @return the Pulsar header value to use + */ + protected abstract String toPulsarHeaderValue(String name, Object value, + @Nullable ToPulsarHeadersContextType context); + + /** + * Called at the end of a successful {@link #toSpringHeaders} invocation. Allows + * concrete implementations the ability to do final logic/cleanup. + * @param springHeaders the Spring Messaging headers that were mapped + * @param pulsarHeaders the resulting map of Pulsar message headers + * @param context the optional context used for the mapping invocation + */ + protected void toPulsarHeadersOnCompleted(MessageHeaders springHeaders, Map pulsarHeaders, + @Nullable ToPulsarHeadersContextType context) { + } + + @Override + public MessageHeaders toSpringHeaders(Message pulsarMessage) { + Objects.requireNonNull(pulsarMessage, "pulsarMessage must not be null"); + var context = toSpringHeadersOnStarted(pulsarMessage); + var headersMap = new LinkedHashMap(); + + // custom user properties (headers) + pulsarMessage.getProperties().forEach((name, value) -> { + if (matchesForInbound(name)) { + var valueToUse = toSpringHeaderValue(name, value, context); + headersMap.put(name, valueToUse); + } + }); + // built-in Pulsar metadata headers + if (pulsarMessage.hasKey()) { + addToHeadersMapIfAllowed(PulsarHeaders.KEY, pulsarMessage::getKey, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.KEY_BYTES, pulsarMessage::getKeyBytes, headersMap::put); + } + if (pulsarMessage.hasOrderingKey()) { + addToHeadersMapIfAllowed(PulsarHeaders.ORDERING_KEY, pulsarMessage::getOrderingKey, headersMap::put); + } + if (pulsarMessage.hasIndex()) { + addToHeadersMapIfAllowed(PulsarHeaders.INDEX, pulsarMessage::getIndex, headersMap::put); + } + addToHeadersMapIfAllowed(PulsarHeaders.MESSAGE_ID, pulsarMessage::getMessageId, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage::getBrokerPublishTime, + headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.EVENT_TIME, pulsarMessage::getEventTime, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.MESSAGE_SIZE, pulsarMessage::size, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.PRODUCER_NAME, pulsarMessage::getProducerName, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.RAW_DATA, pulsarMessage::getData, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.PUBLISH_TIME, pulsarMessage::getPublishTime, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage::getRedeliveryCount, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.REPLICATED_FROM, pulsarMessage::getReplicatedFrom, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.SCHEMA_VERSION, pulsarMessage::getSchemaVersion, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.SEQUENCE_ID, pulsarMessage::getSequenceId, headersMap::put); + addToHeadersMapIfAllowed(PulsarHeaders.TOPIC_NAME, pulsarMessage::getTopicName, headersMap::put); + + var springHeaders = new MessageHeaders(headersMap); + toSpringHeadersOnCompleted(pulsarMessage, springHeaders, context); + return springHeaders; + } + + /** + * Called at the beginning of every {@link #toSpringHeaders} invocation and the + * returned value is passed into each {@link #toSpringHeaderValue} invocation as well + * as the final {@link #toSpringHeadersOnCompleted}. Allows concrete implementations + * the ability to create an arbitrary context object and have it passed through the + * mapping process. + * @param pulsarMessage the Pulsar message whose headers are being mapped + * @return optional context to pass through the mapping invocation + */ + @Nullable + protected ToSpringHeadersContextType toSpringHeadersOnStarted(Message pulsarMessage) { + return null; + } + + /** + * Determine the Spring Messaging header value to use for a Pulsar header. + * @param name the Pulsar header name + * @param value the Pulsar header value + * @param context the optional context used for the mapping invocation + * @return the Spring Messaging header value to use + */ + protected abstract Object toSpringHeaderValue(String name, String value, + @Nullable ToSpringHeadersContextType context); + + /** + * Called at the end of a successful {@link #toPulsarHeaders} invocation. Allows + * concrete implementations the ability to do final logic/cleanup. + * @param pulsarMessage the Pulsar message whose headers were mapped + * @param springHeaders the resulting Spring Messaging headers + * @param context the optional context used for the mapping invocation + */ + protected void toSpringHeadersOnCompleted(Message pulsarMessage, MessageHeaders springHeaders, + @Nullable ToSpringHeadersContextType context) { + } + + /** + * Determine if a header name matches any of the currently configured list of outbound + * matchers. + * @param header the name of the header to check + * @return whether the header matches for outbound + */ + protected boolean matchesForOutbound(String header) { + if (this.outboundMatchers.isEmpty()) { + return true; + } + return matchesAny(header, this.outboundMatchers); + } + + /** + * Determine if a header name matches any of the currently configured list of inbound + * matchers. + * @param header the name of the header to check + * @return whether the header matches for inbound + */ + protected boolean matchesForInbound(String header) { + if (this.inboundMatchers.isEmpty()) { + return true; + } + return matchesAny(header, this.inboundMatchers); + } + + private boolean matchesAny(String header, List matchers) { + for (PulsarHeaderMatcher matcher : matchers) { + if (matcher.matchHeader(header)) { + return !matcher.isNegated(); + } + } + this.logger.debug(() -> "header (%s) WILL NOT be mapped; matched no patterns".formatted(header)); + return false; + } + + private void addToHeadersMapIfAllowed(String key, Supplier valueSupplier, + BiConsumer mapConsumer) { + if (matchesForInbound(key)) { + mapConsumer.accept(key, valueSupplier.get()); + } + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JacksonUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JacksonUtils.java new file mode 100644 index 00000000..8d586798 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JacksonUtils.java @@ -0,0 +1,85 @@ +/* + * Copyright 2019-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.header; + +import org.springframework.util.ClassUtils; + +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.MapperFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.json.JsonMapper; + +/** + * The utilities for Jackson {@link ObjectMapper} instances. + * + * @author Artem Bilan + * @author Chris Bono + */ +public final class JacksonUtils { + + private static final boolean JDK8_MODULE_PRESENT = ClassUtils + .isPresent("com.fasterxml.jackson.datatype.jdk8.Jdk8Module", null); + + private static final boolean JAVA_TIME_MODULE_PRESENT = ClassUtils + .isPresent("com.fasterxml.jackson.datatype.jsr310.JavaTimeModule", null); + + private static final boolean JODA_MODULE_PRESENT = ClassUtils + .isPresent("com.fasterxml.jackson.datatype.joda.JodaModule", null); + + /** + * Factory for {@link ObjectMapper} instances with registered well-known modules and + * disabled {@link MapperFeature#DEFAULT_VIEW_INCLUSION} and + * {@link DeserializationFeature#FAIL_ON_UNKNOWN_PROPERTIES} features. + * @return the {@link ObjectMapper} instance. + */ + public static ObjectMapper enhancedObjectMapper() { + ObjectMapper objectMapper = JsonMapper.builder().configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false) + .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false).build(); + if (JDK8_MODULE_PRESENT) { + objectMapper.registerModule(Jdk8ModuleProvider.MODULE); + } + if (JAVA_TIME_MODULE_PRESENT) { + objectMapper.registerModule(JavaTimeModuleProvider.MODULE); + } + if (JODA_MODULE_PRESENT) { + objectMapper.registerModule(JodaModuleProvider.MODULE); + } + return objectMapper; + } + + private JacksonUtils() { + } + + private static final class Jdk8ModuleProvider { + + static final com.fasterxml.jackson.databind.Module MODULE = new com.fasterxml.jackson.datatype.jdk8.Jdk8Module(); + + } + + private static final class JavaTimeModuleProvider { + + static final com.fasterxml.jackson.databind.Module MODULE = new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule(); + + } + + private static final class JodaModuleProvider { + + static final com.fasterxml.jackson.databind.Module MODULE = new com.fasterxml.jackson.datatype.joda.JodaModule(); + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapper.java new file mode 100644 index 00000000..15761087 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapper.java @@ -0,0 +1,393 @@ +/* + * Copyright 2017-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.header; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +import org.apache.pulsar.client.api.Message; + +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper.ToPulsarHeadersContext; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper.ToSpringHeadersContext; +import org.springframework.util.ClassUtils; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; + +/** + * A {@code PulsarHeaderMapper} implementation that writes headers as JSON. + * + *

+ * Allows user to constrain which classes are {@code "trusted"} for deserialization as + * well as configure a list of classes to serialize with a simple {@code "toString"}. + * + * @author Chris Bono + */ +public class JsonPulsarHeaderMapper extends AbstractPulsarHeaderMapper { + + private static final Set TRUSTED_ARRAY_TYPES = new HashSet<>( + Arrays.asList("[B", "[I", "[J", "[F", "[D", "[C")); + + private static final List DEFAULT_TO_STRING_CLASSES = Arrays.asList("org.springframework.util.MimeType", + "org.springframework.http.MediaType"); + + /** + * Header name for java types of other headers. + */ + public static final String JSON_TYPES = "spring_json_header_types"; + + private final ObjectMapper objectMapper; + + private final Set trustedPackages = new LinkedHashSet<>(); + + private final Set toStringClasses = new LinkedHashSet<>(DEFAULT_TO_STRING_CLASSES); + + /** + * Construct an instance with the provided specifications. + *

+ * NOTE: Internal framework headers are never mapped + * outbound. By default, the {@code "id"} and {@code "timestamp"} headers are also + * excluded from outbound mapping but can be included by adding them to + * {@code outboundPatterns}. + *

+ * NOTE: The patterns are applied in order, stopping on the first + * match (positive or negative). When no pattern is specified, the {@code "*"} pattern + * is added last. However, once a pattern is specified, the {@code "*"} is not added + * and must be added to the specified patterns if desired. + * @param objectMapper the object mapper to use to read/write JSON + * @param inboundPatterns the inbound patterns to match - or empty to match all + * @param outboundPatterns the outbound patterns to match - or empty to match all + * (except internal framework headers and id/timestamp) + * @param trustedPackages the additional packages to trust (allow to be deserialized) + * @param toStringClasses the additional classes to use {@code toString} for + * serialization + * + * @see org.springframework.util.PatternMatchUtils#simpleMatch(String, String) + */ + JsonPulsarHeaderMapper(ObjectMapper objectMapper, List inboundPatterns, List outboundPatterns, + Set trustedPackages, Set toStringClasses) { + super(inboundPatterns, outboundPatterns); + this.objectMapper = Objects.requireNonNull(objectMapper, "objectMapper must not be null"); + Objects.requireNonNull(trustedPackages, "trustedPackages must not be null"); + Objects.requireNonNull(toStringClasses, "toStringClasses must not be null"); + for (var trusted : trustedPackages) { + if ("*".equals(trusted)) { + this.trustedPackages.clear(); + break; + } + this.trustedPackages.add(trusted); + } + this.toStringClasses.addAll(toStringClasses); + } + + /** + * Construct a builder instance that can be used to construct a + * {@link JsonPulsarHeaderMapper}. + * @return builder instance + */ + public static JsonPulsarHeaderMapperBuilder builder() { + return new JsonPulsarHeaderMapperBuilder(); + } + + protected ObjectMapper getObjectMapper() { + return this.objectMapper; + } + + protected Set getTrustedPackages() { + return this.trustedPackages; + } + + protected Set getToStringClasses() { + return this.toStringClasses; + } + + // ToPulsarHeaders ------ + + @Override + protected ToPulsarHeadersContext toPulsarHeadersOnStarted(MessageHeaders springHeaders) { + var jsonHeaders = new LinkedHashMap(); + return new ToPulsarHeadersContext(jsonHeaders); + } + + @Override + protected String toPulsarHeaderValue(String name, Object rawValue, ToPulsarHeadersContext context) { + if (rawValue == null) { + return null; + } + if (rawValue instanceof String) { + return (String) rawValue; + } + String className = rawValue.getClass().getName(); + if (this.toStringClasses.contains(className)) { + return rawValue.toString(); + } + try { + var valueToAdd = getObjectMapper().writeValueAsString(rawValue); + context.jsonTypes().put(name, className); + return valueToAdd; + } + catch (Exception e) { + logger.debug(e, () -> "Could not map %s with type %s (will instead map w/ toString()) reason: %s" + .formatted(name, className, e.getMessage())); + } + return rawValue.toString(); + } + + @Override + protected void toPulsarHeadersOnCompleted(MessageHeaders springHeaders, Map pulsarHeaders, + ToPulsarHeadersContext context) { + var jsonHeaders = context.jsonTypes(); + if (jsonHeaders.size() > 0) { + try { + pulsarHeaders.put(JSON_TYPES, getObjectMapper().writeValueAsString(jsonHeaders)); + } + catch (Exception e) { + logger.error(e, () -> "Could not add json types header due to: %s".formatted(e.getMessage())); + } + } + } + + // ToSpringHeaders ------ + + @Override + protected boolean matchesForInbound(String header) { + return !header.equals(JSON_TYPES) && super.matchesForInbound(header); + } + + @NonNull + @Override + protected ToSpringHeadersContext toSpringHeadersOnStarted(Message pulsarMessage) { + Map types = new HashMap<>(); + if (pulsarMessage.hasProperty(JSON_TYPES)) { + String jsonTypesStr = pulsarMessage.getProperty(JSON_TYPES); + try { + types = getObjectMapper().readValue(jsonTypesStr, new TypeReference<>() { + }); + } + catch (IOException e) { + logger.error(e, + () -> "Could not decode json types: %s due to: %s".formatted(jsonTypesStr, e.getMessage())); + } + } + return new ToSpringHeadersContext(types); + } + + @Override + protected Object toSpringHeaderValue(String name, String value, ToSpringHeadersContext context) { + var jsonTypes = context.jsonTypes(); + if (jsonTypes != null && jsonTypes.containsKey(name)) { + String requestedType = jsonTypes.get(name); + return toJsonHeaderValue(name, value, requestedType); + } + return value; + } + + private Object toJsonHeaderValue(String name, String value, String requestedType) { + if (!trusted(requestedType)) { + return new NonTrustedHeaderType(value, requestedType); + } + + final Class type; + try { + type = ClassUtils.forName(requestedType, null); + } + catch (Exception e) { + logger.error(e, () -> "Could not load type (%s) for header (%s) due to: %s".formatted(requestedType, name, + e.getMessage())); + return value; + } + + try { + return decodeValue(name, value, type); + } + catch (IOException e) { + logger.error(e, () -> "Could not decode type (%s) for header (%s) using value (%s) due to: %s" + .formatted(type, name, value, e.getMessage())); + } + return value; + } + + private Object decodeValue(String name, String value, Class type) throws IOException { + Object decodedValue = getObjectMapper().readValue(value, type); + if (!type.equals(NonTrustedHeaderType.class)) { + return decodedValue; + } + // Upstream NTHT propagated; may be trusted here... + NonTrustedHeaderType nth = (NonTrustedHeaderType) decodedValue; + if (!trusted(nth.untrustedType())) { + return nth; + } + try { + decodedValue = getObjectMapper().readValue(nth.headerValue(), + ClassUtils.forName(nth.untrustedType(), null)); + } + catch (Exception e) { + logger.error(e, + () -> "Could not decode non-trusted header type (%s) for header (%s) using value (%s) due to: %s" + .formatted(nth.untrustedType(), name, nth.headerValue(), e.getMessage())); + } + return decodedValue; + } + + // Trusted ------ + + protected boolean trusted(String requestedType) { + if (requestedType.equals(NonTrustedHeaderType.class.getName())) { + return true; + } + if (TRUSTED_ARRAY_TYPES.contains(requestedType)) { + return true; + } + if (this.trustedPackages.isEmpty()) { + return true; + } + var type = requestedType.startsWith("[") ? requestedType.substring(2) : requestedType; + var lastDot = type.lastIndexOf('.'); + if (lastDot < 0) { + return false; + } + var packageName = type.substring(0, lastDot); + for (var trustedPackage : this.trustedPackages) { + if (packageName.equals(trustedPackage) || packageName.startsWith(trustedPackage + ".")) { + return true; + } + } + return false; + } + + /** + * Represents a header that could not be decoded due to an untrusted type. + * @param headerValue the header value that could not be decoded + * @param untrustedType the class name of the encoded header value + */ + public record NonTrustedHeaderType(String headerValue, String untrustedType) { + } + + /** + * Context used for {@link #toPulsarHeaders} (outbound) that includes the cumulative + * map of header name to classname for headers that are serialized with JSON. This map + * is ultimately included as a separate {@link #JSON_TYPES} header on the outgoing + * Pulsar message. This allows for the inbound consumer to decode the headers when/if + * the message is later consumed. + * @param jsonTypes cumulative map of header name to classname for types that are + * serialized as JSON + */ + public record ToPulsarHeadersContext(Map jsonTypes) { + } + + /** + * Context used for {@link #toSpringHeaders} (inbound) that includes the cumulative + * map of header name to classname for headers in the incoming message that were + * serialized with JSON. This is used to decode the headers in the incoming message. + * @param jsonTypes cumulative map of header name to classname for types that were + * serialized as JSON + */ + public record ToSpringHeadersContext(Map jsonTypes) { + } + + public static class JsonPulsarHeaderMapperBuilder { + + private ObjectMapper objectMapper; + + private final Set trustedPackages = new HashSet<>(); + + private final Set toStringClasses = new HashSet<>(); + + private final List inboundPatterns = new ArrayList<>(); + + private final List outboundPatterns = new ArrayList<>(); + + /** + * Sets the object mapper to use to read/write header values as JSON. + * @param objectMapper the object mapper + * @return current builder + */ + public JsonPulsarHeaderMapperBuilder objectMapper(@Nullable ObjectMapper objectMapper) { + this.objectMapper = objectMapper; + return this; + } + + /** + * Add packages to the list of trusted packages used when constructing objects + * from JSON. + *

+ * NOTE:If a class for a non-trusted package is encountered, the + * header is returned to the application with value of type + * {@link NonTrustedHeaderType}. + * @param packages the packages to include in the trusted list - if any entry is + * {@code "*"} all packages are trusted + * @return current builder + */ + public JsonPulsarHeaderMapperBuilder trustedPackages(String... packages) { + this.trustedPackages.addAll(List.of(packages)); + return this; + } + + /** + * Add class names to the list of classes that should be serialized using their + * {@link #toString()} method. + * @param classNames the class names to add to the 'toString' list + * @return current builder + */ + public JsonPulsarHeaderMapperBuilder toStringClasses(String... classNames) { + this.toStringClasses.addAll(List.of(classNames)); + return this; + } + + /** + * Adds to the list of patterns to be used for inbound header matching. + * @param patterns inbound patterns to add + * @return current builder + */ + public JsonPulsarHeaderMapperBuilder inboundPatterns(String... patterns) { + this.inboundPatterns.addAll(List.of(patterns)); + return this; + } + + /** + * Adds to the list of patterns to be used for outbound header matching. + * @param patterns outbound patterns to add + * @return current builder + */ + public JsonPulsarHeaderMapperBuilder outboundPatterns(String... patterns) { + this.outboundPatterns.addAll(List.of(patterns)); + return this; + } + + public JsonPulsarHeaderMapper build() { + if (this.objectMapper == null) { + this.objectMapper = JacksonUtils.enhancedObjectMapper(); + } + return new JsonPulsarHeaderMapper(this.objectMapper, this.inboundPatterns, this.outboundPatterns, + this.trustedPackages, this.toStringClasses); + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMapper.java similarity index 80% rename from spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java rename to spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMapper.java index 29c2d961..4f3b6b02 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaderMapper.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMapper.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.pulsar.support; +package org.springframework.pulsar.support.header; import java.util.Map; @@ -37,13 +37,19 @@ public interface PulsarHeaderMapper { /** * Map from the given Spring Messaging headers to Pulsar message headers. + *

+ * Commonly used in the outbound flow when a Spring message is being converted to a + * Pulsar message in order to be written out to Pulsar topic (outbound). * @param springHeaders the Spring messaging headers * @return map of Pulsar message headers or an empty map for no headers. */ - Map fromSpringHeaders(MessageHeaders springHeaders); + Map toPulsarHeaders(MessageHeaders springHeaders); /** * Map the headers from the given Pulsar message to Spring Messaging headers. + *

+ * Commonly used in the inbound flow when an incoming Pulsar message is being + * converted to a Spring message. * @param pulsarMessage the Pulsar message containing the headers to map * @return the Spring Messaging headers */ diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMatcher.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMatcher.java new file mode 100644 index 00000000..95ee67ba --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/PulsarHeaderMatcher.java @@ -0,0 +1,120 @@ +/* + * 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.header; + +import java.util.Set; + +import org.springframework.core.log.LogAccessor; +import org.springframework.util.Assert; +import org.springframework.util.PatternMatchUtils; + +/** + * Defines the contract for matching message headers. + * + *

+ * Used by the header mapper to determine whether an incoming or outgoing message header + * should be included in the target. + * + * @author Chris Bono + */ +public interface PulsarHeaderMatcher { + + /** + * Determine if the header matches. + * @param headerName the header name + * @return whether the header matches + */ + boolean matchHeader(String headerName); + + /** + * Determine if this matcher is a negative matcher where a match means that the header + * should not be included. + * @return whether this matcher is a negative matcher + */ + boolean isNegated(); + + /** + * A matcher that never matches a set of headers. + */ + class NeverMatch implements PulsarHeaderMatcher { + + private final Set neverMatchHeaders; + + public NeverMatch(String... headers) { + Assert.notEmpty(headers, "headers must contain at least 1 non-null entry"); + Assert.noNullElements(headers, "headers must not contain null entries"); + this.neverMatchHeaders = Set.of(headers); + } + + @Override + public boolean matchHeader(String headerName) { + return this.neverMatchHeaders.contains(headerName); + } + + @Override + public boolean isNegated() { + return true; + } + + } + + /** + * A pattern-based header matcher that matches if the specified header matches the + * specified simple pattern. + * + * @see PatternMatchUtils#simpleMatch(String, String) + */ + class PatternMatch implements PulsarHeaderMatcher { + + private static final LogAccessor LOGGER = new LogAccessor(PatternMatch.class); + + private final String pattern; + + private final boolean negate; + + public static PatternMatch fromPatternString(String pattern) { + return new PatternMatch(pattern.startsWith("!") ? pattern.substring(1) : pattern, pattern.startsWith("!")); + } + + public PatternMatch(String pattern, boolean negate) { + Assert.notNull(pattern, "Pattern must not be null"); + this.pattern = pattern.toLowerCase(); + this.negate = negate; + } + + @Override + public boolean matchHeader(String headerName) { + if (!PatternMatchUtils.simpleMatch(this.pattern, headerName.toLowerCase())) { + return false; + } + LOGGER.debug(() -> "headerName=[%s] WILL %s be mapped, matched pattern=%s".formatted(headerName, + this.negate ? "NOT " : "", toPatternString())); + return true; + } + + @Override + public boolean isNegated() { + return this.negate; + } + + public String toPatternString() { + return (this.negate ? "!" : "") + this.pattern; + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapper.java new file mode 100644 index 00000000..612da6b9 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapper.java @@ -0,0 +1,59 @@ +/* + * 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.header; + +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * A {@link PulsarHeaderMapper} that converts header values using simple 'toString'. + * + * @author Chris Bono + */ +public class ToStringPulsarHeaderMapper extends AbstractPulsarHeaderMapper { + + /** + * Construct a 'ToString' mapper that will match all inbound headers and all outbound + * headers (except internal framework headers and id/timestamp). + */ + public ToStringPulsarHeaderMapper() { + super(Collections.emptyList(), Collections.emptyList()); + } + + /** + * Construct a 'ToString' mapper that will match the supplied inbound and outbound + * patterns. + * @param inboundPatterns the inbound patterns to match - or empty map to match all + * @param outboundPatterns the outbound patterns to match - or empty to match all + * (except internal framework headers and id/timestamp) + */ + public ToStringPulsarHeaderMapper(List inboundPatterns, List outboundPatterns) { + super(inboundPatterns, outboundPatterns); + } + + @Override + protected String toPulsarHeaderValue(String name, Object value, Object context) { + return Objects.toString(value, null); + } + + @Override + protected Object toSpringHeaderValue(String headerName, String rawHeader, Object context) { + return rawHeader; + } + +} 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 deleted file mode 100644 index 42c0be6f..00000000 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/support/DefaultPulsarHeaderMapperTests.java +++ /dev/null @@ -1,242 +0,0 @@ -/* - * 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/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapperTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapperTests.java new file mode 100644 index 00000000..a46b559b --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/AbstractPulsarHeaderMapperTests.java @@ -0,0 +1,281 @@ +/* + * 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.header; + +import static java.util.Map.entry; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNullPointerException; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.springframework.pulsar.support.header.PulsarHeaderMapperTestUtil.assertSpringHeadersHavePulsarMetadata; +import static org.springframework.pulsar.support.header.PulsarHeaderMapperTestUtil.mockPulsarMessage; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Objects; + +import org.apache.pulsar.client.api.Message; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.support.PulsarHeaders; + +/** + * Tests for {@link AbstractPulsarHeaderMapper}. + * + * @author Chris Bono + */ +abstract class AbstractPulsarHeaderMapperTests { + + abstract AbstractPulsarHeaderMapper mapper(); + + abstract AbstractPulsarHeaderMapper mapperWithInboundPatterns(String... patterns); + + abstract AbstractPulsarHeaderMapper mapperWithOutboundPatterns(String... patterns); + + @Nested + class ToSpringHeadersInboundTests { + + @Test + void nullPulsarMessage() { + assertThatNullPointerException().isThrownBy(() -> mapper().toSpringHeaders(null)) + .withMessage("pulsarMessage must not be null"); + } + + @Test + void pulsarMessageWitRequiredMetadataOnly() { + var pulsarMessage = mockPulsarMessage(false, Collections.emptyMap()); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertSpringHeadersHavePulsarMetadata(springHeaders, pulsarMessage, false); + } + + @Test + void pulsarMessageWithAllMetadataAndSimpleUserProperties() { + var pulsarHeaders = Collections.singletonMap("foo", "bar"); + var pulsarMessage = mockPulsarMessage(true, pulsarHeaders); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertSpringHeadersHavePulsarMetadata(springHeaders, pulsarMessage, true); + assertThat(springHeaders).containsEntry("foo", "bar"); + } + + @Test + void pulsarHeaderValuesFilteredByInboundPatterns() { + var mapper = mapperWithInboundPatterns("!foo", "!" + PulsarHeaders.MESSAGE_ID, "*"); + var pulsarHeaders = new HashMap(); + pulsarHeaders.put("foo", "bar"); + pulsarHeaders.put("info", "5150"); + var pulsarMessage = mockPulsarMessage(true, pulsarHeaders); + var springHeaders = mapper.toSpringHeaders(pulsarMessage); + // check metadata - sanity check metadata pulled in by existence of + // MESSAGE_SIZE + assertThat(springHeaders).containsKey(PulsarHeaders.MESSAGE_SIZE) + .doesNotContainKey(PulsarHeaders.MESSAGE_ID); + // check user properties + assertThat(springHeaders).containsKey("info").doesNotContainKey("foo"); + } + + @Test + void ensureCallbacksInvoked() { + var inboundContext = "inboundContext"; + TestPulsarHeaderMapper testMapper = new TestPulsarHeaderMapper(inboundContext, null); + TestPulsarHeaderMapper spyTestMapper = spy(testMapper); + var pulsarHeaders = new HashMap(); + pulsarHeaders.put("foo", "bar"); + pulsarHeaders.put("info", "5150"); + var pulsarMessage = mockPulsarMessage(true, pulsarHeaders); + + var springHeaders = spyTestMapper.toSpringHeaders(pulsarMessage); + + verify(spyTestMapper).toSpringHeadersOnStarted(pulsarMessage); + verify(spyTestMapper).toSpringHeaderValue("foo", "bar", inboundContext); + verify(spyTestMapper).toSpringHeaderValue("info", "5150", inboundContext); + verify(spyTestMapper).toSpringHeadersOnCompleted(pulsarMessage, springHeaders, inboundContext); + var numPulsarHeaders = springHeaders.size() - 2; // id/timestamp not pulsar + // headers + verify(spyTestMapper, times(numPulsarHeaders)).matchesForInbound(anyString()); + } + + } + + @Nested + class ToPulsarHeadersOutboundTests { + + @Test + void nullSpringHeaders() { + assertThatNullPointerException().isThrownBy(() -> mapper().toPulsarHeaders(null)) + .withMessage("springHeaders must not be null"); + } + + @Test + void emptySpringHeaders() { + assertThat(mapper().toPulsarHeaders(new MessageHeaders(Collections.emptyMap()))).isEmpty(); + } + + @Test + void springHeadersWithOnlyPulsarMetadata() { + var pulsarMessage = mockPulsarMessage(true, Collections.emptyMap()); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertSpringHeadersHavePulsarMetadata(springHeaders, pulsarMessage, true); + assertThat(mapper().toPulsarHeaders(springHeaders)).isEmpty(); + } + + @Test + void springHeadersWithSimpleValues() { + var headers = new HashMap(); + headers.put("foo", "bar"); + assertThat(mapper().toPulsarHeaders(new MessageHeaders(headers))).containsOnly(entry("foo", "bar")); + } + + @Test + void springHeadersWithNullValue() { + var headers = Collections.singletonMap("foo", null); + assertThat(mapper().toPulsarHeaders(new MessageHeaders(headers))).containsEntry("foo", null); + } + + @Test + void springHeadersFilteredByOutboundPatterns() { + var mapper = mapperWithOutboundPatterns("!foo", "*"); + var springHeaders = new HashMap(); + springHeaders.put("foo", "bar"); + springHeaders.put("info", "5150"); + assertThat(mapper.toPulsarHeaders(new MessageHeaders(springHeaders))).doesNotContainKey("foo") + .containsKeys("id", "timestamp"); // '*' allows id/ts + } + + @Test + void springHeadersFilteredByOutboundPatternsIdAndTsExcluded() { + var mapper = mapperWithOutboundPatterns("!foo", "info"); + var springHeaders = new HashMap(); + springHeaders.put("foo", "bar"); + springHeaders.put("info", "5150"); + assertThat(mapper.toPulsarHeaders(new MessageHeaders(springHeaders))).containsOnlyKeys("info"); + } + + @Test + void springHeadersFilteredByOutboundPatterns_NoneAllowed() { + var mapper = mapperWithOutboundPatterns("!foo"); + var springHeaders = new HashMap(); + springHeaders.put("foo", "bar"); + springHeaders.put("info", "5150"); + assertThat(mapper.toPulsarHeaders(new MessageHeaders(springHeaders))).isEmpty(); + } + + @Test + void springHeadersCanAllowIdAndTimestampOutbound() { + var mapper = mapperWithOutboundPatterns("id", "timestamp"); + var springHeaders = new HashMap(); + springHeaders.put("foo", "bar"); + assertThat(mapper.toPulsarHeaders(new MessageHeaders(springHeaders))).containsKeys("id", "timestamp") + .doesNotContainKey("foo"); + } + + @Test + void ensureCallbacksInvoked() { + var outboundContext = "outboundContext"; + TestPulsarHeaderMapper testMapper = new TestPulsarHeaderMapper(null, outboundContext); + TestPulsarHeaderMapper spyTestMapper = spy(testMapper); + var headersMap = new HashMap(); + headersMap.put("foo", "bar"); + headersMap.put("info", "5150"); + var springHeaders = new MessageHeaders(headersMap); + + var pulsarHeaders = spyTestMapper.toPulsarHeaders(springHeaders); + + verify(spyTestMapper).toPulsarHeadersOnStarted(springHeaders); + verify(spyTestMapper).toPulsarHeaderValue("foo", "bar", outboundContext); + verify(spyTestMapper).toPulsarHeaderValue("info", "5150", outboundContext); + verify(spyTestMapper).toPulsarHeadersOnCompleted(springHeaders, pulsarHeaders, outboundContext); + verify(spyTestMapper, times(springHeaders.size())).matchesForOutbound(anyString()); + } + + } + + @Nested + class RoundtripTests { + + @Test + void inboundToSpringOutboundToPulsarInboundToSpring() { + // Inbound initial + var customHeaders = new HashMap(); + customHeaders.put("foo", "bar"); + var inboundPulsarMsg = mockPulsarMessage(true, customHeaders); + // Comes into spring msg w/ headers (foo=bar) + pulsar msg metadata + var inboundSpringHeaders = mapper().toSpringHeaders(inboundPulsarMsg); + assertThat(inboundSpringHeaders).containsEntry("foo", "bar"); + assertSpringHeadersHavePulsarMetadata(inboundSpringHeaders, inboundPulsarMsg, true); + // Modify spring msg headers -> 'foo=bar-bar' and add uuid=UUID(1234) + var mutableSpringHeadersMap = new HashMap<>(inboundSpringHeaders); + var fooValue = mutableSpringHeadersMap.get("foo"); + mutableSpringHeadersMap.put("foo", fooValue + "-" + fooValue); + + // Outbound + var outboundPulsarHeaders = mapper().toPulsarHeaders(new MessageHeaders(mutableSpringHeadersMap)); + assertThat(outboundPulsarHeaders).containsEntry("foo", "bar-bar"); + + // Inbound again: modified foo=bar-bar, UUID as object, pulsar msg metadata + inboundPulsarMsg = mockPulsarMessage(true, outboundPulsarHeaders); + inboundSpringHeaders = mapper().toSpringHeaders(inboundPulsarMsg); + assertThat(inboundSpringHeaders).containsEntry("foo", "bar-bar"); + assertSpringHeadersHavePulsarMetadata(inboundSpringHeaders, inboundPulsarMsg, true); + } + + } + + /** + * A test header mapper that helps w/ lifecycle callback verification. + */ + static class TestPulsarHeaderMapper extends AbstractPulsarHeaderMapper { + + private String toSpringHeadersContext; + + private String toPulsarHeadersContext; + + TestPulsarHeaderMapper(String toSpringHeadersContext, String toPulsarHeadersContext) { + super(Collections.emptyList(), Collections.emptyList()); + this.toSpringHeadersContext = toSpringHeadersContext; + this.toPulsarHeadersContext = toPulsarHeadersContext; + } + + @Nullable + @Override + protected String toSpringHeadersOnStarted(Message pulsarMessage) { + return this.toSpringHeadersContext; + } + + @Override + protected Object toSpringHeaderValue(String name, String value, @Nullable String context) { + return value; + } + + @Nullable + @Override + protected String toPulsarHeadersOnStarted(MessageHeaders springHeaders) { + return this.toPulsarHeadersContext; + } + + @Override + protected String toPulsarHeaderValue(String name, Object value, @Nullable String context) { + return Objects.toString(value, null); + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapperTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapperTests.java new file mode 100644 index 00000000..878830c7 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/JsonPulsarHeaderMapperTests.java @@ -0,0 +1,187 @@ +/* + * 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.header; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.pulsar.support.header.JsonPulsarHeaderMapper.JSON_TYPES; +import static org.springframework.pulsar.support.header.PulsarHeaderMapperTestUtil.assertSpringHeadersHavePulsarMetadata; +import static org.springframework.pulsar.support.header.PulsarHeaderMapperTestUtil.mockPulsarMessage; + +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper.NonTrustedHeaderType; + +import com.fasterxml.jackson.core.JsonProcessingException; + +/** + * Tests for {@link JsonPulsarHeaderMapper}. + * + * @author Chris Bono + */ +class JsonPulsarHeaderMapperTests extends AbstractPulsarHeaderMapperTests { + + @Override + AbstractPulsarHeaderMapper mapper() { + return JsonPulsarHeaderMapper.builder().build(); + } + + @Override + AbstractPulsarHeaderMapper mapperWithInboundPatterns(String... patterns) { + return JsonPulsarHeaderMapper.builder().inboundPatterns(patterns).build(); + } + + @Override + AbstractPulsarHeaderMapper mapperWithOutboundPatterns(String... patterns) { + return JsonPulsarHeaderMapper.builder().outboundPatterns(patterns).build(); + } + + @Nested + class ToSpringHeadersJsonMapperSpecificInboundTests { + + @Test + void pulsarMessageWithObjectUserProperty() { + var uuid = UUID.randomUUID(); + var customHeaders = new HashMap(); + customHeaders.put("uuid", "\"%s\"".formatted(uuid.toString())); + customHeaders.put(JSON_TYPES, "{\"uuid\":\"%s\"}".formatted(UUID.class.getName())); + var pulsarMessage = mockPulsarMessage(true, customHeaders); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertThat(springHeaders).containsEntry("uuid", uuid); + } + + @Test + void pulsarMessageWithUntrustedUserProperty() { + // Once trustedPackages is set, all others are untrusted + var mapper = JsonPulsarHeaderMapper.builder().trustedPackages("com.acme").build(); + var uuid = UUID.randomUUID(); + var serializedUuid = "\"%s\"".formatted(uuid.toString()); + var customHeaders = new HashMap(); + customHeaders.put("uuid", serializedUuid); + customHeaders.put(JSON_TYPES, "{\"uuid\":\"%s\"}".formatted(UUID.class.getName())); + var pulsarMessage = mockPulsarMessage(true, customHeaders); + var springHeaders = mapper.toSpringHeaders(pulsarMessage); + assertThat(springHeaders).containsEntry("uuid", + new NonTrustedHeaderType(serializedUuid, UUID.class.getName())); + } + + @ParameterizedTest + @ValueSource(booleans = { true, false }) + void pulsarMessageWithUpstreamNth(boolean nowTrusted) throws JsonProcessingException { + var mapper = JsonPulsarHeaderMapper.builder() + .trustedPackages(nowTrusted ? UUID.class.getPackageName() : "com.acme").build(); + var uuid = UUID.randomUUID(); + var serializedUuid = "\"%s\"".formatted(uuid.toString()); + var upstreamNth = new NonTrustedHeaderType(serializedUuid, UUID.class.getName()); + var serializedUpstreamNth = JacksonUtils.enhancedObjectMapper().writeValueAsString(upstreamNth); + var customHeaders = new HashMap(); + customHeaders.put("uuid", serializedUpstreamNth); + customHeaders.put(JSON_TYPES, "{\"uuid\":\"%s\"}".formatted(NonTrustedHeaderType.class.getName())); + var pulsarMessage = mockPulsarMessage(true, customHeaders); + var springHeaders = mapper.toSpringHeaders(pulsarMessage); + assertThat(springHeaders).containsEntry("uuid", nowTrusted ? uuid : upstreamNth); + } + + } + + @Nested + class ToPulsarHeadersJsonMapperSpecificOutboundTests { + + @Test + void springHeadersWithObjectValues() { + var uuid = UUID.randomUUID(); + var headers = new HashMap(); + headers.put("foo", "bar"); + headers.put("uuid", uuid); + assertThat(mapper().toPulsarHeaders(new MessageHeaders(headers))).containsEntry("foo", "bar") + .containsEntry("uuid", "\"%s\"".formatted(uuid.toString())) + .extractingByKey(JSON_TYPES, InstanceOfAssertFactories.STRING) + .contains("\"uuid\":\"java.util.UUID\""); + } + + @Test + void springHeadersWithValuesInToStringClasses() { + var mapper = JsonPulsarHeaderMapper.builder().toStringClasses(UUID.class.getName()).build(); + var uuid = UUID.randomUUID(); + var headers = Collections.singletonMap("uuid", uuid); + assertThat(mapper.toPulsarHeaders(new MessageHeaders(headers))).containsEntry("uuid", uuid.toString()) + .doesNotContainKey(JSON_TYPES); + } + + } + + @Nested + class RoundtripJsonMapperSpecificTests { + + @Test + void inboundToSpringOutboundToPulsarInboundToSpring() { + var mapper = mapper(); + // Inbound initial + var uuid = UUID.randomUUID(); + var customHeaders = new HashMap(); + customHeaders.put("foo", "bar"); + var inboundPulsarMsg = mockPulsarMessage(true, customHeaders); + // Comes into spring msg w/ headers (foo=bar) + pulsar msg metadata + var inboundSpringHeaders = mapper.toSpringHeaders(inboundPulsarMsg); + assertThat(inboundSpringHeaders).containsEntry("foo", "bar"); + assertSpringHeadersHavePulsarMetadata(inboundSpringHeaders, inboundPulsarMsg, true); + + // Modify spring msg headers -> 'foo=bar-bar' and add uuid=UUID(1234) + var mutableSpringHeadersMap = new HashMap<>(inboundSpringHeaders); + var fooValue = mutableSpringHeadersMap.get("foo"); + mutableSpringHeadersMap.put("foo", fooValue + "-" + fooValue); + mutableSpringHeadersMap.put("uuid", uuid); + + // Outbound + var outboundPulsarHeaders = mapper.toPulsarHeaders(new MessageHeaders(mutableSpringHeadersMap)); + assertThat(outboundPulsarHeaders).containsEntry("foo", "bar-bar"); + assertThat(outboundPulsarHeaders).containsEntry("uuid", "\"%s\"".formatted(uuid.toString())); + assertThat(outboundPulsarHeaders).containsOnlyKeys("foo", "uuid", JSON_TYPES); + + // Inbound again: modified foo=bar-bar, UUID as object, pulsar msg metadata + inboundPulsarMsg = mockPulsarMessage(true, outboundPulsarHeaders); + inboundSpringHeaders = mapper.toSpringHeaders(inboundPulsarMsg); + assertThat(inboundSpringHeaders).containsEntry("foo", "bar-bar").containsEntry("uuid", uuid); + assertSpringHeadersHavePulsarMetadata(inboundSpringHeaders, inboundPulsarMsg, true); + } + + @Test + void roundtripWithByteArrayValue() { + var payload = "payload5150".getBytes(); + var payloadEnc = Base64.getEncoder().encodeToString(payload); + var headers = Map.of("payload", payload); + var pulsarHeaders = mapper().toPulsarHeaders(new MessageHeaders(headers)); + assertThat(pulsarHeaders).containsEntry("payload", "\"%s\"".formatted(payloadEnc)).containsEntry(JSON_TYPES, + "{\"payload\":\"[B\"}"); + var pulsarMessage = mockPulsarMessage(true, pulsarHeaders); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertThat(springHeaders).containsEntry("payload", payload); + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/PulsarHeaderMapperTestUtil.java b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/PulsarHeaderMapperTestUtil.java new file mode 100644 index 00000000..b0ce837e --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/PulsarHeaderMapperTestUtil.java @@ -0,0 +1,112 @@ +/* + * 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.header; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.AssertionsForClassTypes.entry; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.springframework.pulsar.support.header.JsonPulsarHeaderMapper.JSON_TYPES; + +import java.util.Map; +import java.util.Optional; + +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; + +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.support.PulsarHeaders; + +/** + * Utilities for testing header mapper related functionality. + * + * @author Chris Bono + */ +final class PulsarHeaderMapperTestUtil { + + private PulsarHeaderMapperTestUtil() { + } + + @SuppressWarnings("unchecked") + static Message mockPulsarMessage(boolean includeOptionalMetadata, Map userProperties) { + Message pulsarMessage = (Message) mock(Message.class); + + // custom user properties + when(pulsarMessage.getProperties()).thenReturn(userProperties); + when(pulsarMessage.hasProperty(JSON_TYPES)).thenReturn(userProperties.containsKey(JSON_TYPES)); + when(pulsarMessage.getProperty(JSON_TYPES)).thenReturn(userProperties.get(JSON_TYPES)); + + // 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; + } + + static void assertSpringHeadersHavePulsarMetadata(MessageHeaders springHeaders, Message pulsarMessage, + boolean shouldHaveOptionalFields) { + // @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())); + if (shouldHaveOptionalFields) { + assertThat(springHeaders).contains( + entry(PulsarHeaders.KEY, pulsarMessage.getKey()), + entry(PulsarHeaders.KEY_BYTES, pulsarMessage.getKeyBytes()), + entry(PulsarHeaders.ORDERING_KEY, pulsarMessage.getOrderingKey()), + entry(PulsarHeaders.INDEX, pulsarMessage.getIndex())); + } + else { + assertThat(springHeaders).doesNotContainKeys( + PulsarHeaders.KEY, + PulsarHeaders.KEY_BYTES, + PulsarHeaders.ORDERING_KEY, + PulsarHeaders.INDEX); + } + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapperTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapperTests.java new file mode 100644 index 00000000..a7b26afd --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/support/header/ToStringPulsarHeaderMapperTests.java @@ -0,0 +1,79 @@ +/* + * 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.header; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.pulsar.support.header.PulsarHeaderMapperTestUtil.mockPulsarMessage; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.junit.jupiter.api.Test; + +import org.springframework.messaging.MessageHeaders; + +/** + * Tests for {@link ToStringPulsarHeaderMapper}. + * + * @author Chris Bono + */ +class ToStringPulsarHeaderMapperTests extends AbstractPulsarHeaderMapperTests { + + @Override + AbstractPulsarHeaderMapper mapper() { + return new ToStringPulsarHeaderMapper(); + } + + @Override + AbstractPulsarHeaderMapper mapperWithInboundPatterns(String... patterns) { + return new ToStringPulsarHeaderMapper(List.of(patterns), Collections.emptyList()); + } + + @Override + AbstractPulsarHeaderMapper mapperWithOutboundPatterns(String... patterns) { + return new ToStringPulsarHeaderMapper(Collections.emptyList(), List.of(patterns)); + } + + @Test + void roundtripWithObjectHeaderValue() { + var uuid = UUID.randomUUID(); + var customHeaders = new HashMap(); + customHeaders.put("uuid", uuid.toString()); + var pulsarMessage = mockPulsarMessage(true, customHeaders); + + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + var pulsarHeaders = mapper().toPulsarHeaders(springHeaders); + assertThat(springHeaders).containsEntry("uuid", uuid.toString()); + assertThat(pulsarHeaders).containsEntry("uuid", "%s".formatted(uuid.toString())); + } + + @Test + void roundtripWithByteArrayHeaderValue() { + var payload = "payload5150".getBytes(); + var headers = Map.of("payload", payload); + var pulsarHeaders = mapper().toPulsarHeaders(new MessageHeaders(headers)); + assertThat(pulsarHeaders).containsEntry("payload", payload.toString()); + + var pulsarMessage = mockPulsarMessage(true, pulsarHeaders); + var springHeaders = mapper().toSpringHeaders(pulsarMessage); + assertThat(springHeaders).containsEntry("payload", payload.toString()); + } + +} diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 6e6785e9..1b4fa8b2 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -4,15 +4,9 @@ "https://checkstyle.org/dtds/suppressions_1_2.dtd"> - - - - - - + + - -