Add filter/encode ability to header mappers (#366)
- Add simple 'ToString' mapper - Add 'json' mapper for complex types - Delete 'default' mapper - Use 'json' mapper as default mapper - Inbound/outbound filter rules See #361
This commit is contained in:
Chris Bono
@@ -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<Object> applySpringHeadersAsPulsarProperties(MessageHeaders headers) {
|
||||
return (mb) -> this.headerMapper.fromSpringHeaders(headers).forEach(mb::property);
|
||||
return (mb) -> this.headerMapper.toPulsarHeaders(headers).forEach(mb::property);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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.*)$")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<V> {
|
||||
private boolean converterSet;
|
||||
|
||||
private PulsarMessageConverter<V> messageConverter = new PulsarRecordMessageConverter<V>(
|
||||
new DefaultPulsarHeaderMapper());
|
||||
JsonPulsarHeaderMapper.builder().build());
|
||||
|
||||
private Type fallbackType = Object.class;
|
||||
|
||||
|
||||
@@ -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<String, String> fromSpringHeaders(MessageHeaders springHeaders) {
|
||||
Objects.requireNonNull(springHeaders, "springHeaders must be specified");
|
||||
var pulsarHeaders = new LinkedHashMap<String, String>();
|
||||
springHeaders.forEach((k, v) -> pulsarHeaders.put(k, Objects.toString(v, null)));
|
||||
return pulsarHeaders;
|
||||
}
|
||||
|
||||
@Override
|
||||
public MessageHeaders toSpringHeaders(Message<?> pulsarMessage) {
|
||||
Objects.requireNonNull(pulsarMessage, "pulsarMessage must be specified");
|
||||
var headersMap = new HashMap<String, Object>(pulsarMessage.getProperties());
|
||||
if (pulsarMessage.hasKey()) {
|
||||
headersMap.put(PulsarHeaders.KEY, pulsarMessage.getKey());
|
||||
headersMap.put(PulsarHeaders.KEY_BYTES, pulsarMessage.getKeyBytes());
|
||||
}
|
||||
if (pulsarMessage.hasOrderingKey()) {
|
||||
headersMap.put(PulsarHeaders.ORDERING_KEY, pulsarMessage.getOrderingKey());
|
||||
}
|
||||
if (pulsarMessage.hasIndex()) {
|
||||
headersMap.put(PulsarHeaders.INDEX, pulsarMessage.getIndex());
|
||||
}
|
||||
headersMap.put(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId());
|
||||
headersMap.put(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime());
|
||||
headersMap.put(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime());
|
||||
headersMap.put(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size());
|
||||
headersMap.put(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName());
|
||||
headersMap.put(PulsarHeaders.RAW_DATA, pulsarMessage.getData());
|
||||
headersMap.put(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime());
|
||||
headersMap.put(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount());
|
||||
headersMap.put(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom());
|
||||
headersMap.put(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion());
|
||||
headersMap.put(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId());
|
||||
headersMap.put(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName());
|
||||
return new MessageHeaders(headersMap);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/**
|
||||
*
|
||||
|
||||
@@ -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}.
|
||||
*
|
||||
* <p>
|
||||
* 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 <ToPulsarHeadersContextType> type of context object used in the
|
||||
* {@link #toPulsarHeaders} API
|
||||
* @param <ToSpringHeadersContextType> type of context object used in the
|
||||
* {@link #toSpringHeaders} API
|
||||
* @author Chris Bono
|
||||
*/
|
||||
public abstract class AbstractPulsarHeaderMapper<ToPulsarHeadersContextType, ToSpringHeadersContextType>
|
||||
implements PulsarHeaderMapper {
|
||||
|
||||
protected final LogAccessor logger = new LogAccessor(this.getClass());
|
||||
|
||||
private final List<PulsarHeaderMatcher> inboundMatchers = new ArrayList<>();
|
||||
|
||||
private final List<PulsarHeaderMatcher> outboundMatchers = new ArrayList<>();
|
||||
|
||||
/**
|
||||
* Construct a mapper that will match the supplied inbound and outbound patterns.
|
||||
* <p>
|
||||
* <strong>NOTE:</strong> Internal framework headers are <em>never</em> 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}.
|
||||
* <p>
|
||||
* <strong>NOTE:</strong> 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<String> inboundPatterns, List<String> 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<String, String> toPulsarHeaders(MessageHeaders springHeaders) {
|
||||
Objects.requireNonNull(springHeaders, "springHeaders must not be null");
|
||||
var pulsarHeaders = new LinkedHashMap<String, String>();
|
||||
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<String, String> 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<String, Object>();
|
||||
|
||||
// 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<PulsarHeaderMatcher> 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<Object> valueSupplier,
|
||||
BiConsumer<String, Object> mapConsumer) {
|
||||
if (matchesForInbound(key)) {
|
||||
mapConsumer.accept(key, valueSupplier.get());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* 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<ToPulsarHeadersContext, ToSpringHeadersContext> {
|
||||
|
||||
private static final Set<String> TRUSTED_ARRAY_TYPES = new HashSet<>(
|
||||
Arrays.asList("[B", "[I", "[J", "[F", "[D", "[C"));
|
||||
|
||||
private static final List<String> 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<String> trustedPackages = new LinkedHashSet<>();
|
||||
|
||||
private final Set<String> toStringClasses = new LinkedHashSet<>(DEFAULT_TO_STRING_CLASSES);
|
||||
|
||||
/**
|
||||
* Construct an instance with the provided specifications.
|
||||
* <p>
|
||||
* <strong>NOTE:</strong> Internal framework headers are <em>never</em> 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}.
|
||||
* <p>
|
||||
* <strong>NOTE:</strong> 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<String> inboundPatterns, List<String> outboundPatterns,
|
||||
Set<String> trustedPackages, Set<String> 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<String> getTrustedPackages() {
|
||||
return this.trustedPackages;
|
||||
}
|
||||
|
||||
protected Set<String> getToStringClasses() {
|
||||
return this.toStringClasses;
|
||||
}
|
||||
|
||||
// ToPulsarHeaders ------
|
||||
|
||||
@Override
|
||||
protected ToPulsarHeadersContext toPulsarHeadersOnStarted(MessageHeaders springHeaders) {
|
||||
var jsonHeaders = new LinkedHashMap<String, String>();
|
||||
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<String, String> 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<String, String> 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<String, String> 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<String, String> jsonTypes) {
|
||||
}
|
||||
|
||||
public static class JsonPulsarHeaderMapperBuilder {
|
||||
|
||||
private ObjectMapper objectMapper;
|
||||
|
||||
private final Set<String> trustedPackages = new HashSet<>();
|
||||
|
||||
private final Set<String> toStringClasses = new HashSet<>();
|
||||
|
||||
private final List<String> inboundPatterns = new ArrayList<>();
|
||||
|
||||
private final List<String> 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.
|
||||
* <p>
|
||||
* <strong>NOTE:</strong>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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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.
|
||||
* <p>
|
||||
* 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<String, String> fromSpringHeaders(MessageHeaders springHeaders);
|
||||
Map<String, String> toPulsarHeaders(MessageHeaders springHeaders);
|
||||
|
||||
/**
|
||||
* Map the headers from the given Pulsar message to Spring Messaging headers.
|
||||
* <p>
|
||||
* 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
|
||||
*/
|
||||
@@ -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.
|
||||
*
|
||||
* <p>
|
||||
* 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 <strong>not</strong> 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<String> 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Object, Object> {
|
||||
|
||||
/**
|
||||
* 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<String> inboundPatterns, List<String> 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String> mockPulsarMessage(boolean includeOptionalMetadata, Map<String, String> userProperties) {
|
||||
Message<String> pulsarMessage = (Message<String>) mock(Message.class);
|
||||
|
||||
// custom user properties
|
||||
when(pulsarMessage.getProperties()).thenReturn(userProperties);
|
||||
|
||||
// optional metadata
|
||||
when(pulsarMessage.hasKey()).thenReturn(includeOptionalMetadata);
|
||||
when(pulsarMessage.getKey()).thenReturn("key");
|
||||
when(pulsarMessage.getKeyBytes()).thenReturn("key".getBytes());
|
||||
when(pulsarMessage.hasOrderingKey()).thenReturn(includeOptionalMetadata);
|
||||
when(pulsarMessage.getOrderingKey()).thenReturn("orderingKey".getBytes());
|
||||
when(pulsarMessage.hasIndex()).thenReturn(includeOptionalMetadata);
|
||||
when(pulsarMessage.getIndex()).thenReturn(Optional.of(1L));
|
||||
|
||||
// required metadata
|
||||
MessageId messageId = mock(MessageId.class);
|
||||
when(pulsarMessage.getMessageId()).thenReturn(messageId);
|
||||
when(pulsarMessage.getBrokerPublishTime()).thenReturn(Optional.of(100L));
|
||||
when(pulsarMessage.getEventTime()).thenReturn(200L);
|
||||
when(pulsarMessage.size()).thenReturn(300);
|
||||
when(pulsarMessage.getProducerName()).thenReturn("producerName");
|
||||
when(pulsarMessage.getData()).thenReturn("data".getBytes());
|
||||
when(pulsarMessage.getPublishTime()).thenReturn(400L);
|
||||
when(pulsarMessage.getRedeliveryCount()).thenReturn(500);
|
||||
when(pulsarMessage.getReplicatedFrom()).thenReturn("replicatedFrom");
|
||||
when(pulsarMessage.getSchemaVersion()).thenReturn("schemaVersion".getBytes());
|
||||
when(pulsarMessage.getSequenceId()).thenReturn(600L);
|
||||
when(pulsarMessage.getTopicName()).thenReturn("topicName");
|
||||
|
||||
return pulsarMessage;
|
||||
}
|
||||
|
||||
@Nested
|
||||
class FromSpringHeaders {
|
||||
|
||||
@Test
|
||||
void nullSpringHeaders() {
|
||||
assertThatNullPointerException().isThrownBy(() -> mapper.fromSpringHeaders(null))
|
||||
.withMessage("springHeaders must be specified");
|
||||
}
|
||||
|
||||
@Test
|
||||
void emptySpringHeaders() {
|
||||
assertThat(mapper.fromSpringHeaders(new MessageHeaders(Collections.emptyMap()))).containsOnlyKeys("id",
|
||||
"timestamp");
|
||||
}
|
||||
|
||||
@Test
|
||||
void springHeadersWithNullValue() {
|
||||
var headers = new HashMap<String, Object>();
|
||||
headers.put("foo", "bar");
|
||||
headers.put("uuid", null);
|
||||
assertThat(mapper.fromSpringHeaders(new MessageHeaders(headers)))
|
||||
.containsOnlyKeys("id", "timestamp", "foo", "uuid")
|
||||
.contains(entry("foo", "bar"), entry("uuid", null));
|
||||
}
|
||||
|
||||
@Test
|
||||
void springHeadersWithValues() {
|
||||
var uuid = UUID.randomUUID();
|
||||
var headers = new HashMap<String, Object>();
|
||||
headers.put("foo", "bar");
|
||||
headers.put("uuid", uuid);
|
||||
assertThat(mapper.fromSpringHeaders(new MessageHeaders(headers)))
|
||||
.containsOnlyKeys("id", "timestamp", "foo", "uuid")
|
||||
.contains(entry("foo", "bar"), entry("uuid", uuid.toString()));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Nested
|
||||
class ToSpringHeaders {
|
||||
|
||||
@Test
|
||||
void nullPulsarMessage() {
|
||||
assertThatNullPointerException().isThrownBy(() -> mapper.toSpringHeaders(null))
|
||||
.withMessage("pulsarMessage must be specified");
|
||||
}
|
||||
|
||||
@Test
|
||||
void pulsarMessageWithOnlyRequiredMetadata() {
|
||||
var pulsarMessage = mockPulsarMessage(false, Collections.emptyMap());
|
||||
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
|
||||
// @formatter:off
|
||||
assertThat(springHeaders).contains(
|
||||
entry(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId()),
|
||||
entry(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime()),
|
||||
entry(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime()),
|
||||
entry(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size()),
|
||||
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
|
||||
entry(PulsarHeaders.RAW_DATA, pulsarMessage.getData()),
|
||||
entry(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime()),
|
||||
entry(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount()),
|
||||
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
|
||||
entry(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion()),
|
||||
entry(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId()),
|
||||
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
|
||||
assertThat(springHeaders).doesNotContainKeys(
|
||||
PulsarHeaders.KEY,
|
||||
PulsarHeaders.KEY_BYTES,
|
||||
PulsarHeaders.ORDERING_KEY,
|
||||
PulsarHeaders.INDEX);
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
@Test
|
||||
void pulsarMessageWithAllMetadataAndUserProperties() {
|
||||
var uuid = UUID.randomUUID();
|
||||
var customHeaders = new HashMap<String, String>();
|
||||
customHeaders.put("foo", "bar");
|
||||
customHeaders.put("uuid", uuid.toString());
|
||||
var pulsarMessage = mockPulsarMessage(true, customHeaders);
|
||||
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
|
||||
// @formatter:off
|
||||
assertThat(springHeaders).contains(
|
||||
entry("foo", "bar"),
|
||||
entry("uuid", uuid.toString()),
|
||||
entry(PulsarHeaders.KEY, pulsarMessage.getKey()),
|
||||
entry(PulsarHeaders.KEY_BYTES, pulsarMessage.getKeyBytes()),
|
||||
entry(PulsarHeaders.ORDERING_KEY, pulsarMessage.getOrderingKey()),
|
||||
entry(PulsarHeaders.INDEX, pulsarMessage.getIndex()),
|
||||
entry(PulsarHeaders.MESSAGE_ID, pulsarMessage.getMessageId()),
|
||||
entry(PulsarHeaders.BROKER_PUBLISH_TIME, pulsarMessage.getBrokerPublishTime()),
|
||||
entry(PulsarHeaders.EVENT_TIME, pulsarMessage.getEventTime()),
|
||||
entry(PulsarHeaders.MESSAGE_SIZE, pulsarMessage.size()),
|
||||
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
|
||||
entry(PulsarHeaders.RAW_DATA, pulsarMessage.getData()),
|
||||
entry(PulsarHeaders.PUBLISH_TIME, pulsarMessage.getPublishTime()),
|
||||
entry(PulsarHeaders.REDELIVERY_COUNT, pulsarMessage.getRedeliveryCount()),
|
||||
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
|
||||
entry(PulsarHeaders.SCHEMA_VERSION, pulsarMessage.getSchemaVersion()),
|
||||
entry(PulsarHeaders.SEQUENCE_ID, pulsarMessage.getSequenceId()),
|
||||
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Nested
|
||||
class ToAndFromSpringHeaders {
|
||||
|
||||
@Test
|
||||
void pulsarMessageWithOnlyRequiredMetadataRoundTripped() {
|
||||
var pulsarMessage = mockPulsarMessage(false, Collections.emptyMap());
|
||||
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
|
||||
var pulsarHeaders = mapper.fromSpringHeaders(springHeaders);
|
||||
// @formatter:off
|
||||
assertThat(pulsarHeaders).contains(
|
||||
entry(PulsarHeaders.MESSAGE_ID, Objects.toString(pulsarMessage.getMessageId())),
|
||||
entry(PulsarHeaders.BROKER_PUBLISH_TIME, Objects.toString(pulsarMessage.getBrokerPublishTime())),
|
||||
entry(PulsarHeaders.EVENT_TIME, Objects.toString(pulsarMessage.getEventTime())),
|
||||
entry(PulsarHeaders.MESSAGE_SIZE, Objects.toString(pulsarMessage.size())),
|
||||
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
|
||||
entry(PulsarHeaders.RAW_DATA, Objects.toString(pulsarMessage.getData())),
|
||||
entry(PulsarHeaders.PUBLISH_TIME, Objects.toString(pulsarMessage.getPublishTime())),
|
||||
entry(PulsarHeaders.REDELIVERY_COUNT, Objects.toString(pulsarMessage.getRedeliveryCount())),
|
||||
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
|
||||
entry(PulsarHeaders.SCHEMA_VERSION, Objects.toString(pulsarMessage.getSchemaVersion())),
|
||||
entry(PulsarHeaders.SEQUENCE_ID, Objects.toString(pulsarMessage.getSequenceId())),
|
||||
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
@Test
|
||||
void pulsarMessageWithAllMetadataAndUserPropertiesRoundTripped() {
|
||||
var pulsarMessage = mockPulsarMessage(true, Collections.singletonMap("foo", "bar"));
|
||||
var springHeaders = mapper.toSpringHeaders(pulsarMessage);
|
||||
var pulsarHeaders = mapper.fromSpringHeaders(springHeaders);
|
||||
// @formatter:off
|
||||
assertThat(pulsarHeaders).contains(
|
||||
entry("foo", "bar"),
|
||||
entry(PulsarHeaders.KEY, pulsarMessage.getKey()),
|
||||
entry(PulsarHeaders.KEY_BYTES, Objects.toString(pulsarMessage.getKeyBytes())),
|
||||
entry(PulsarHeaders.ORDERING_KEY, Objects.toString(pulsarMessage.getOrderingKey())),
|
||||
entry(PulsarHeaders.INDEX, Objects.toString(pulsarMessage.getIndex())),
|
||||
entry(PulsarHeaders.MESSAGE_ID, Objects.toString(pulsarMessage.getMessageId())),
|
||||
entry(PulsarHeaders.BROKER_PUBLISH_TIME, Objects.toString(pulsarMessage.getBrokerPublishTime())),
|
||||
entry(PulsarHeaders.EVENT_TIME, Objects.toString(pulsarMessage.getEventTime())),
|
||||
entry(PulsarHeaders.MESSAGE_SIZE, Objects.toString(pulsarMessage.size())),
|
||||
entry(PulsarHeaders.PRODUCER_NAME, pulsarMessage.getProducerName()),
|
||||
entry(PulsarHeaders.RAW_DATA, Objects.toString(pulsarMessage.getData())),
|
||||
entry(PulsarHeaders.PUBLISH_TIME, Objects.toString(pulsarMessage.getPublishTime())),
|
||||
entry(PulsarHeaders.REDELIVERY_COUNT, Objects.toString(pulsarMessage.getRedeliveryCount())),
|
||||
entry(PulsarHeaders.REPLICATED_FROM, pulsarMessage.getReplicatedFrom()),
|
||||
entry(PulsarHeaders.SCHEMA_VERSION, Objects.toString(pulsarMessage.getSchemaVersion())),
|
||||
entry(PulsarHeaders.SEQUENCE_ID, Objects.toString(pulsarMessage.getSequenceId())),
|
||||
entry(PulsarHeaders.TOPIC_NAME, pulsarMessage.getTopicName()));
|
||||
// @formatter:on
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, String>();
|
||||
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<String, String>();
|
||||
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<String, Object>();
|
||||
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<String, Object>();
|
||||
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<String, Object>();
|
||||
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<String, Object>();
|
||||
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<String, Object>();
|
||||
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<String, Object>();
|
||||
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<String, String>();
|
||||
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<String, String> {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, String>();
|
||||
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<String, String>();
|
||||
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<String, String>();
|
||||
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<String, Object>();
|
||||
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.<String, Object>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<String, String>();
|
||||
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.<String, Object>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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String> mockPulsarMessage(boolean includeOptionalMetadata, Map<String, String> userProperties) {
|
||||
Message<String> pulsarMessage = (Message<String>) 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<String> 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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<String, String>();
|
||||
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.<String, Object>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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -4,15 +4,9 @@
|
||||
"https://checkstyle.org/dtds/suppressions_1_2.dtd">
|
||||
<suppressions>
|
||||
<suppress files="package-info\.java" checks=".*" />
|
||||
<suppress files="[\\/]test[\\/]" checks="RequireThis" />
|
||||
<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
|
||||
<suppress files="DefaultPulsarHeaderMapperTests" checks="AvoidStaticImport" />
|
||||
<suppress files="DefaultSchemaResolverTests" checks="AvoidStaticImport|MethodParamPad" />
|
||||
<suppress files="DefaultTopicResolverTests" checks="AvoidStaticImport" />
|
||||
<suppress files="PulsarBinderUtilsTest" checks="AvoidStaticImport" />
|
||||
<suppress files="[\\/]test[\\/]" checks="RequireThis|AvoidStaticImport|Javadoc*" />
|
||||
<suppress files="DefaultSchemaResolverTests" checks="MethodParamPad" />
|
||||
<suppress files="PulsarFunctionAdministrationIntegrationTests" checks="Regexp" />
|
||||
<suppress files="PulsarMessageChannelBinderResolveSchemaTests" checks="AvoidStaticImport" />
|
||||
|
||||
<suppress files="Proto" checks=".*"/>
|
||||
<suppress files="ReactiveSpringPulsarBootApp" checks="HideUtilityClassConstructor"/>
|
||||
<suppress files="[\\/]spring-pulsar-docs[\\/]" checks="JavadocPackage|JavadocType|JavadocVariable|SpringDeprecatedCheck" />
|
||||
|
||||
Reference in New Issue
Block a user