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

GH-855 - Support to add headers in event externalization.

Browse Source 
EventExternalizationConfiguration now exposes a ….headers(Class<T>, Function<T, Map<String, Object>) to allow to define a function that extracts headers from the event that are supposed to added to the message to be sent out. The Kafka and AMQP implementations have been augmented to consider those configurations.

Furthermore, if the mapping step prior to the externalization creates a Spring Message<?>, we add routing information as fallback and send it out as is.
This commit is contained in:
Oliver Drotbohm
2024-10-08 15:04:54 +02:00
parent 747834754e
commit 84e9f38b07
10 changed files with 193 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-modulith-events/spring-modulith-events-amqp/src/main/java/org/springframework/modulith/events/amqp/RabbitEventExternalizerConfiguration.java
View File

@@ -62,8 +62,9 @@ class RabbitEventExternalizerConfiguration {
return new DelegatingEventExternalizer(configuration, (target, payload) -> {
var routing = BrokerRouting.of(target, context);
var headers = configuration.getHeadersFor(payload);
operations.convertAndSend(routing.getTarget(), routing.getKey(payload), payload);
operations.convertAndSend(routing.getTarget(), routing.getKey(payload), payload, headers);
return CompletableFuture.completedFuture(null);
});

18
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/DefaultEventExternalizationConfiguration.java
View File

@@ -15,6 +15,7 @@
*/
package org.springframework.modulith.events;
import java.util.Map;
import java.util.function.Function;
import java.util.function.Predicate;
@@ -31,6 +32,7 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
private final Predicate<Object> filter;
private final Function<Object, Object> mapper;
private final Function<Object, RoutingTarget> router;
private final Function<Object, Map<String, Object>> headers;
/**
* Creates a new {@link DefaultEventExternalizationConfiguration}
@@ -38,9 +40,10 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
* @param filter must not be {@literal null}.
* @param mapper must not be {@literal null}.
* @param router must not be {@literal null}.
* @param headers must not be {@literal null}.
*/
DefaultEventExternalizationConfiguration(Predicate<Object> filter, Function<Object, Object> mapper,
Function<Object, RoutingTarget> router) {
Function<Object, RoutingTarget> router, Function<Object, Map<String, Object>> headers) {
Assert.notNull(filter, "Filter must not be null!");
Assert.notNull(mapper, "Mapper must not be null!");
@@ -49,6 +52,7 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
this.filter = filter;
this.mapper = mapper;
this.router = router;
this.headers = headers;
}
/**
@@ -95,4 +99,16 @@ class DefaultEventExternalizationConfiguration implements EventExternalizationCo
return router.apply(event).verify();
}
/*
* (non-Javadoc)
* @see org.springframework.modulith.events.EventExternalizationConfiguration#getHeadersFor(java.lang.Object)
*/
@Override
public Map<String, Object> getHeadersFor(Object event) {
Assert.notNull(event, "Event must not be null!");
return headers.apply(event);
}
}

87
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventExternalizationConfiguration.java
View File

@@ -19,7 +19,9 @@ import static org.springframework.core.annotation.AnnotatedElementUtils.*;
import java.lang.annotation.Annotation;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.BiFunction;
import java.util.function.BiPredicate;
@@ -188,6 +190,15 @@ public interface EventExternalizationConfiguration {
*/
RoutingTarget determineTarget(Object event);
/**
* Returns the headers to be attached to the message sent out for the given event.
*
* @param event must not be {@literal null}.
* @return will never be {@literal null}.
* @since 1.3
*/
Map<String, Object> getHeadersFor(Object event);
/**
* API to define which events are supposed to be selected for externalization.
*
@@ -367,6 +378,7 @@ public interface EventExternalizationConfiguration {
private final Predicate<Object> filter;
private final Function<Object, Object> mapper;
private final Function<Object, RoutingTarget> router;
private final Function<Object, Map<String, Object>> headers;
/**
* Creates a new {@link Router} for the given selector {@link Predicate} and mapper and router {@link Function}s.
@@ -374,16 +386,20 @@ public interface EventExternalizationConfiguration {
* @param filter must not be {@literal null}.
* @param mapper must not be {@literal null}.
* @param router must not be {@literal null}.
* @param headers must not be {@literal null}.
*/
Router(Predicate<Object> filter, Function<Object, Object> mapper, Function<Object, RoutingTarget> router) {
Router(Predicate<Object> filter, Function<Object, Object> mapper, Function<Object, RoutingTarget> router,
Function<Object, Map<String, Object>> headers) {
Assert.notNull(filter, "Selector must not be null!");
Assert.notNull(mapper, "Mapper must not be null!");
Assert.notNull(router, "Router must not be null!");
Assert.notNull(headers, "Headers extractor must not be null!");
this.filter = filter;
this.mapper = mapper;
this.router = router;
this.headers = headers;
}
/**
@@ -392,7 +408,7 @@ public interface EventExternalizationConfiguration {
* @param filter must not be {@literal null}.
*/
Router(Predicate<Object> filter) {
this(filter, Function.identity(), DEFAULT_ROUTER);
this(filter, Function.identity(), DEFAULT_ROUTER, it -> Collections.emptyMap());
}
/**
@@ -406,7 +422,7 @@ public interface EventExternalizationConfiguration {
Assert.notNull(mapper, "Mapper must not be null!");
return new Router(filter, mapper, router);
return new Router(filter, mapper, router, headers);
}
/**
@@ -428,7 +444,42 @@ public interface EventExternalizationConfiguration {
.map(mapper::apply)
.orElse(it);
return new Router(filter, this.mapper.compose(combined), router);
return new Router(filter, this.mapper.compose(combined), router, headers);
}
/**
* Registers the given function to extract headers from the events to be externalized. Will reset the entire header
* extractor arrangement. For type-specific extractions, see {@link #headers(Class, Function)}.
*
* @param extractor must not be {@literal null}.
* @return will never be {@literal null}.
* @see #headers(Class, Function)
* @since 1.3
*/
public Router headers(Function<Object, Map<String, Object>> extractor) {
Assert.notNull(extractor, "Headers extractor must not be null!");
return new Router(filter, mapper, router, extractor);
}
/**
* Registers the given type-specific function to extract headers from the events to be externalized.
*
* @param extractor must not be {@literal null}.
* @return will never be {@literal null}.
* @since 1.3
*/
public <T> Router headers(Class<T> type, Function<T, Map<String, Object>> extractor) {
Assert.notNull(type, "Type must not be null!");
Assert.notNull(extractor, "Headers extractor must not be null!");
Function<Object, Map<String, Object>> combined = it -> toOptional(type, it)
.map(extractor::apply)
.orElseGet(() -> this.headers.apply(it));
return new Router(filter, mapper, router, combined);
}
/**
@@ -437,7 +488,7 @@ public interface EventExternalizationConfiguration {
* @return will never be {@literal null}.
*/
public Router routeMapped() {
return new Router(filter, mapper, router.compose(mapper));
return new Router(filter, mapper, router.compose(mapper), headers);
}
/**
@@ -450,7 +501,7 @@ public interface EventExternalizationConfiguration {
Assert.notNull(router, "Router must not be null!");
return new Router(filter, mapper, router);
return new Router(filter, mapper, router, headers);
}
/**
@@ -466,9 +517,11 @@ public interface EventExternalizationConfiguration {
Assert.notNull(type, "Type must not be null!");
Assert.notNull(router, "Router must not be null!");
return new Router(filter, mapper, it -> toOptional(type, it)
Function<Object, RoutingTarget> adapted = it -> toOptional(type, it)
.map(router::apply)
.orElseGet(() -> this.router.apply(it)));
.orElseGet(() -> this.router.apply(it));
return new Router(filter, mapper, adapted, headers);
}
/**
@@ -487,9 +540,11 @@ public interface EventExternalizationConfiguration {
Assert.notNull(type, "Type must not be null!");
Assert.notNull(extractor, "Extractor must not be null!");
return new Router(filter, mapper, it -> toOptional(type, it)
Function<Object, RoutingTarget> adapted = it -> toOptional(type, it)
.map(t -> this.router.apply(t).withKey(extractor.apply(t)))
.orElseGet(() -> this.router.apply(it)));
.orElseGet(() -> this.router.apply(it));
return new Router(filter, mapper, adapted, headers);
}
/**
@@ -503,9 +558,9 @@ public interface EventExternalizationConfiguration {
Assert.notNull(router, "Router must not be null!");
return new Router(filter, mapper, it -> router.apply(it)
.orElseGet(() -> this.router.apply(it)))
.build();
Function<Object, RoutingTarget> adapted = it -> router.apply(it).orElseGet(() -> this.router.apply(it));
return new Router(filter, mapper, adapted, headers).build();
}
/**
@@ -533,16 +588,16 @@ public interface EventExternalizationConfiguration {
Assert.notNull(router, "Router must not be null!");
return new Router(filter, mapper, it -> router.apply(it.getClass()));
return new Router(filter, mapper, it -> router.apply(it.getClass()), headers);
}
/**
* Creates a new {@link EventExternalizationConfiguration} refelcting the current configuration.
* Creates a new {@link EventExternalizationConfiguration} reflecting the current configuration.
*
* @return will never be {@literal null}.
*/
public EventExternalizationConfiguration build() {
return new DefaultEventExternalizationConfiguration(filter, mapper, router);
return new DefaultEventExternalizationConfiguration(filter, mapper, router, headers);
}
private static <T> Optional<T> toOptional(Class<T> type, Object source) {

17
spring-modulith-events/spring-modulith-events-api/src/test/java/org/springframework/modulith/events/EventExternalizationConfigurationUnitTests.java
View File

@@ -22,6 +22,8 @@ import lombok.RequiredArgsConstructor;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Test;
@@ -130,6 +132,21 @@ class EventExternalizationConfigurationUnitTests {
assertThat(target.getKey()).isNull();
}
@Test // GH-855
void registersHeaderExtractor() {
var configuration = defaults("org.springframework.modulith")
.headers(AnotherSampleEvent.class, it -> Map.of("another", "anotherValue"))
.headers(SampleEvent.class, it -> Map.of("sample", "value"))
.build();
assertThat(configuration.getHeadersFor(new SampleEvent()))
.containsEntry("sample", "value");
assertThat(configuration.getHeadersFor(new AnotherSampleEvent()))
.containsEntry("another", "anotherValue");
}
@Retention(RetentionPolicy.RUNTIME)
@interface CustomExternalized {
String value() default "";

15
spring-modulith-events/spring-modulith-events-kafka/src/main/java/org/springframework/modulith/events/kafka/KafkaEventExternalizerConfiguration.java
View File

@@ -27,6 +27,9 @@ import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.modulith.events.EventExternalizationConfiguration;
import org.springframework.modulith.events.config.EventExternalizationAutoConfiguration;
import org.springframework.modulith.events.support.BrokerRouting;
@@ -61,7 +64,17 @@ class KafkaEventExternalizerConfiguration {
return new DelegatingEventExternalizer(configuration, (target, payload) -> {
var routing = BrokerRouting.of(target, context);
return operations.send(routing.getTarget(), routing.getKey(payload), payload);
var builder = payload instanceof Message<?> message
? MessageBuilder.fromMessage(message)
: MessageBuilder.withPayload(payload).copyHeaders(configuration.getHeadersFor(payload));
var message = builder
.setHeaderIfAbsent(KafkaHeaders.KEY, routing.getKey(payload))
.setHeaderIfAbsent(KafkaHeaders.TOPIC, routing.getTarget())
.build();
return operations.send(message);
});
}
}

5
spring-modulith-events/spring-modulith-events-kafka/src/main/java/org/springframework/modulith/events/kafka/KafkaJacksonConfiguration.java
View File

@@ -22,6 +22,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.PropertySource;
import org.springframework.kafka.support.converter.ByteArrayJsonMessageConverter;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -41,7 +42,7 @@ class KafkaJacksonConfiguration {
@Bean
@ConditionalOnBean(ObjectMapper.class)
@ConditionalOnMissingBean(JsonMessageConverter.class)
JsonMessageConverter jsonMessageConverter(ObjectMapper mapper) {
return new JsonMessageConverter(mapper);
ByteArrayJsonMessageConverter jsonMessageConverter(ObjectMapper mapper) {
return new ByteArrayJsonMessageConverter(mapper);
}
}

2
spring-modulith-events/spring-modulith-events-kafka/src/main/resources/kafka-json.properties
View File

@@ -1,2 +1,2 @@
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.ByteArraySerializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

68
spring-modulith-events/spring-modulith-events-kafka/src/test/java/org/springframework/modulith/events/kafka/KafkaEventExternalizerConfigurationIntegrationTests.java
View File

@@ -18,11 +18,20 @@ package org.springframework.modulith.events.kafka;
import static org.assertj.core.api.Assertions.*;
import static org.mockito.Mockito.*;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.modulith.events.EventExternalizationConfiguration;
import org.springframework.modulith.events.Externalized;
import org.springframework.modulith.events.support.DelegatingEventExternalizer;
/**
@@ -33,6 +42,8 @@ import org.springframework.modulith.events.support.DelegatingEventExternalizer;
*/
class KafkaEventExternalizerConfigurationIntegrationTests {
private KafkaOperations<?, ?> operations = mock(KafkaOperations.class);
@Test // GH-342
void registersExternalizerByDefault() {
@@ -52,12 +63,61 @@ class KafkaEventExternalizerConfigurationIntegrationTests {
});
}
@Test // GH-855
void addsHeadersIfConfigured() {
var config = EventExternalizationConfiguration.defaults("org")
.headers(Sample.class, __ -> Map.of("key", "value"))
.build();
assertMessage(config, it -> {
assertThat(it.getHeaders()).containsKey("key");
});
}
@Test // GH-855
void sendsMessageAsIsIfMappingTarget() {
var config = EventExternalizationConfiguration.defaults("org")
.mapping(Sample.class, it -> MessageBuilder.withPayload(it).setHeader("key", "value").build())
.build();
assertMessage(config, it -> {
assertThat(it.getHeaders()).contains(Map.entry("key", "value"));
});
}
private void assertMessage(EventExternalizationConfiguration configuration, Consumer<Message<?>> assertions) {
basicSetup(configuration)
.run(ctxt -> {
ctxt.getBean(DelegatingEventExternalizer.class).externalize(new Sample());
var captor = ArgumentCaptor.forClass(Message.class);
verify(operations).send(captor.capture());
assertions.accept(captor.getValue());
});
}
private ApplicationContextRunner basicSetup() {
return basicSetup(null);
}
private ApplicationContextRunner basicSetup(@Nullable EventExternalizationConfiguration config) {
Supplier<EventExternalizationConfiguration> configProvider = () -> config == null
? EventExternalizationConfiguration.disabled()
: config;
return new ApplicationContextRunner()
.withConfiguration(
AutoConfigurations.of(KafkaEventExternalizerConfiguration.class))
.withBean(EventExternalizationConfiguration.class, () -> EventExternalizationConfiguration.disabled())
.withBean(KafkaOperations.class, () -> mock(KafkaOperations.class));
.withConfiguration(AutoConfigurations.of(KafkaEventExternalizerConfiguration.class))
.withBean(EventExternalizationConfiguration.class, configProvider)
.withBean(KafkaOperations.class, () -> operations);
}
@Externalized
record Sample() {}
}

1
spring-modulith-events/spring-modulith-events-messaging/src/main/java/org/springframework/modulith/events/messaging/SpringMessagingEventExternalizerConfiguration.java
View File

@@ -66,6 +66,7 @@ class SpringMessagingEventExternalizerConfiguration {
var message = MessageBuilder
.withPayload(payload)
.setHeader(MODULITH_ROUTING_HEADER, target.toString())
.copyHeadersIfAbsent(configuration.getHeadersFor(payload))
.build();
if (logger.isDebugEnabled()) {

5
spring-modulith-examples/spring-modulith-example-kafka/src/test/java/example/TestApplication.java
View File

@@ -28,6 +28,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.messaging.Message;
/**
* @author Oliver Drotbohm
@@ -44,9 +45,9 @@ public class TestApplication {
var mock = mock(KafkaOperations.class);
when(mock.send(any(), any())).then(invocation -> {
when(mock.send(any(Message.class))).then(invocation -> {
logger.info("Sending message {} to {}.", invocation.getArguments()[1], invocation.getArguments()[0]);
logger.info("Sending message {}.", invocation.getArguments()[0]);
return null;
});