Publish events to Spring Cloud Stream channels #5
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?> | ||
| <!-- | ||
| ~ Copyright (c) 2010-2020. Axon Framework | ||
|
There was a problem hiding this comment. The notice should extend to 2021, as this commit is from 2021. |
||
| ~ | ||
| ~ 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 | ||
| ~ | ||
| ~ http://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. | ||
| --> | ||
|
|
||
| <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" | ||
| xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> | ||
| <modelVersion>4.0.0</modelVersion> | ||
| <parent> | ||
| <artifactId>axon-springcloud-parent</artifactId> | ||
| <groupId>org.axonframework.extensions.springcloud</groupId> | ||
| <version>4.5-SNAPSHOT</version> | ||
| </parent> | ||
|
|
||
| <artifactId>axon-springcloud-streams</artifactId> | ||
|
|
||
| <name>Axon Framework Spring Cloud Stream Extension</name> | ||
| <description> | ||
| Module containing Message Adapter implementations for distributing events over Spring Cloud Stream. | ||
| </description> | ||
|
|
||
| <packaging>jar</packaging> | ||
|
|
||
| <dependencies> | ||
|
|
||
| <dependency> | ||
| <groupId>org.axonframework</groupId> | ||
| <artifactId>axon-messaging</artifactId> | ||
| <version>${axon.version}</version> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.springframework</groupId> | ||
| <artifactId>spring-messaging</artifactId> | ||
| <version>${spring-messaging.version}</version> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.springframework.integration</groupId> | ||
| <artifactId>spring-integration-core</artifactId> | ||
| <version>${spring-integration.version}</version> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.springframework.cloud</groupId> | ||
| <artifactId>spring-cloud-stream</artifactId> | ||
| </dependency> | ||
|
|
||
| <dependency> | ||
| <groupId>com.fasterxml.jackson.core</groupId> | ||
| <artifactId>jackson-databind</artifactId> | ||
| <version>${jackson.version}</version> | ||
| <optional>true</optional> | ||
| </dependency> | ||
|
|
||
| <dependency> | ||
| <groupId>com.fasterxml.jackson.datatype</groupId> | ||
| <artifactId>jackson-datatype-jsr310</artifactId> | ||
| <version>${jackson.version}</version> | ||
| <scope>test</scope> | ||
| </dependency> | ||
|
|
||
| <dependency> | ||
| <groupId>org.springframework.cloud</groupId> | ||
| <artifactId>spring-cloud-stream-test-support</artifactId> | ||
| <scope>test</scope> | ||
| </dependency> | ||
|
|
||
| </dependencies> | ||
|
|
||
| <build> | ||
| <resources> | ||
| <resource> | ||
| <directory>src/main/resources</directory> | ||
| <filtering>true</filtering> | ||
| </resource> | ||
| </resources> | ||
| <plugins> | ||
| <plugin> | ||
| <artifactId>maven-jar-plugin</artifactId> | ||
| <configuration> | ||
| <archive> | ||
| <manifestEntries> | ||
| <Automatic-Module-Name>org.axonframework.extensions.springcloud</Automatic-Module-Name> | ||
| </manifestEntries> | ||
| </archive> | ||
| </configuration> | ||
| </plugin> | ||
| </plugins> | ||
| </build> | ||
| </project> | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,207 @@ | ||
| /* | ||
| * Copyright (c) 2010-2021. Axon Framework | ||
| * | ||
| * 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 | ||
| * | ||
| * http://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.axonframework.extensions.springcloud.eventhandling; | ||
|
|
||
| import org.axonframework.common.AxonConfigurationException; | ||
| import org.axonframework.common.Registration; | ||
| import org.axonframework.eventhandling.EventBus; | ||
| import org.axonframework.eventhandling.EventMessage; | ||
| import org.axonframework.extensions.springcloud.eventhandling.converter.SpringMessageConverter; | ||
| import org.axonframework.messaging.SubscribableMessageSource; | ||
| import org.springframework.integration.handler.AbstractMessageProducingHandler; | ||
| import org.springframework.messaging.Message; | ||
| import org.springframework.messaging.MessageChannel; | ||
| import org.springframework.messaging.support.GenericMessage; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.CopyOnWriteArrayList; | ||
| import java.util.function.Consumer; | ||
| import java.util.function.Predicate; | ||
|
|
||
| import static java.util.Collections.singletonList; | ||
| import static org.axonframework.common.BuilderUtils.assertNonNull; | ||
|
|
||
| /** | ||
| * Initialize the adapter to publish all incoming events to the subscribed processors. Note that this instance should | ||
| * be registered as a consumer of a Spring Message Channel. | ||
| * | ||
| * @author Mehdi chitforoosh | ||
| * @since 4.5 | ||
| */ | ||
| public class SpringStreamMessageProcessor extends AbstractMessageProducingHandler implements SubscribableMessageSource<EventMessage<?>> { | ||
|
There was a problem hiding this comment. Could we call it the Calling it a processor to closely aligns with Axon's Event Processor logic. There was a problem hiding this comment. Alright, I notice there already is a |
||
|
|
||
| private final SubscribableMessageSource<EventMessage<?>> messageSource; | ||
| private final CopyOnWriteArrayList<Consumer<List<? extends EventMessage<?>>>> messageProcessors = new CopyOnWriteArrayList<>(); | ||
| private Predicate<? super EventMessage<?>> filter; | ||
| private final SpringMessageConverter converter; | ||
|
|
||
| /** | ||
| * Instantiate a {@link SpringStreamMessageProcessor} based on the fields contained in the {@link SpringStreamMessageProcessor.Builder}. | ||
| * The {@link SubscribableMessageSource} is a <b>hard requirement</b> and thus should be provided. | ||
| * The {@link SpringMessageConverter} is a <b>hard requirement</b> and thus should be provided. | ||
| * <p> | ||
| * Will validate that the {@link SubscribableMessageSource} and {@link SpringMessageConverter} and {@link EventBus} are not {@code null}, | ||
| * and will throw an {@link AxonConfigurationException} if for either of them this holds. | ||
| * | ||
| * @param builder the {@link SpringStreamMessageProcessor.Builder} used to instantiate a {@link SpringStreamMessageProcessor} instance | ||
| */ | ||
| protected SpringStreamMessageProcessor(SpringStreamMessageProcessor.Builder builder) { | ||
| builder.validate(); | ||
| this.filter = builder.filter; | ||
| this.messageSource = builder.messageSource; | ||
| this.converter = builder.converter; | ||
| if (builder.eventBus != null) { | ||
| this.messageProcessors.addAll(singletonList(builder.eventBus::publish)); | ||
|
There was a problem hiding this comment. I wonder if we should add EventBus by default here? @smcvb what do you think? In my opinion, we could leave this decision to the users of the API, but I am afraid that in that case, we are missing something like There was a problem hiding this comment. Honestly, I think this is a bad idea. Simply publishing the Spring Messages as new events on the local |
||
| } | ||
| // Subscribes this event message handler to the message source. | ||
| this.messageSource.subscribe(this::handle); | ||
| } | ||
|
|
||
| /** | ||
| * Instantiate a Builder to be able to create a {@link SpringStreamMessageProcessor}. | ||
| * The {@link SubscribableMessageSource} is a <b>hard requirement</b> and as such should be provided. | ||
| * The {@link SpringMessageConverter} is a <b>hard requirement</b> and thus should be provided. | ||
| * | ||
| * @return a Builder to be able to create a {@link SpringStreamMessageProcessor}. | ||
| */ | ||
| public static SpringStreamMessageProcessor.Builder builder() { | ||
| return new SpringStreamMessageProcessor.Builder(); | ||
| } | ||
|
|
||
| @Override | ||
| public Registration subscribe(Consumer<List<? extends EventMessage<?>>> messageProcessor) { | ||
| messageProcessors.add(messageProcessor); | ||
| return () -> messageProcessors.remove(messageProcessor); | ||
| } | ||
|
|
||
| /** | ||
| * If allows by the filter, wraps the given {@link EventMessage} in a {@link GenericMessage} ands sends it to the | ||
| * configured {@link MessageChannel}. | ||
| * | ||
| * @param events the events to handle | ||
| */ | ||
| protected void handle(List<? extends EventMessage<?>> events) { | ||
| events.stream() | ||
| .filter(filter) | ||
| .forEach(event -> this.sendOutput(converter.createSpringMessage(event), null, false)); | ||
| } | ||
|
|
||
| /** | ||
| * Handles the given {@link Message}. If the filter refuses the message, it is ignored. | ||
| * | ||
| * @param message The message containing the event to publish | ||
| */ | ||
| @Override | ||
| protected void handleMessageInternal(Message<?> message) { | ||
| Optional<EventMessage<?>> optional = converter.readSpringMessage(message); | ||
| optional.ifPresent(eventMessage -> { | ||
| List<? extends EventMessage<?>> messages = singletonList(eventMessage); | ||
| for (Consumer<List<? extends EventMessage<?>>> messageProcessor : messageProcessors) { | ||
| messageProcessor.accept(messages); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Set filter to remove filtered messages | ||
| * | ||
| * @param filter | ||
| */ | ||
| public void setFilter(Predicate<? super EventMessage<?>> filter) { | ||
| this.filter = filter; | ||
| } | ||
|
|
||
| /** | ||
| * Builder class to instantiate a {@link SpringStreamMessageProcessor}. | ||
| * The {@link SubscribableMessageSource} is a <b>hard requirement</b> and thus should be provided. | ||
| * The {@link SpringMessageConverter} is a <b>hard requirement</b> and thus should be provided. | ||
| */ | ||
| public static class Builder { | ||
|
|
||
| private SubscribableMessageSource<EventMessage<?>> messageSource; | ||
| private EventBus eventBus; | ||
| private Predicate<? super EventMessage<?>> filter = m -> true; | ||
| private SpringMessageConverter converter; | ||
|
|
||
| /** | ||
| * Sets the filter to send specific event messages. | ||
| * | ||
| * @param filter The filter to send filtered event messages | ||
| * @return the current Builder instance, for fluent interfacing | ||
| */ | ||
| public SpringStreamMessageProcessor.Builder filter(Predicate<? super EventMessage<?>> filter) { | ||
| this.filter = filter; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Sets the messageSource for subscribing handler. | ||
| * | ||
| * @param messageSource The messageSource to subscribe | ||
| * @return the current Builder instance, for fluent interfacing | ||
| */ | ||
| public SpringStreamMessageProcessor.Builder messageSource(SubscribableMessageSource<EventMessage<?>> messageSource) { | ||
| assertNonNull(messageSource, "messageSource may not be null"); | ||
| this.messageSource = messageSource; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Sets the eventBus to publish received event message. | ||
| * | ||
| * @param eventBus The eventBus to publish event messages | ||
| * @return the current Builder instance, for fluent interfacing | ||
| */ | ||
| public SpringStreamMessageProcessor.Builder eventBus(EventBus eventBus) { | ||
| this.eventBus = eventBus; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Sets the converter to convert spring messages to event messages and versa. | ||
| * | ||
| * @param converter The converter to convert messages | ||
| * @return the current Builder instance, for fluent interfacing | ||
| */ | ||
| public SpringStreamMessageProcessor.Builder converter(SpringMessageConverter converter) { | ||
| assertNonNull(converter, "converter may not be null"); | ||
| this.converter = converter; | ||
| return this; | ||
| } | ||
|
|
||
| /** | ||
| * Initializes a {@link SpringStreamMessageProcessor} as specified through this Builder. | ||
| * | ||
| * @return a {@link SpringStreamMessageProcessor} as specified through this Builder | ||
| */ | ||
| public SpringStreamMessageProcessor build() { | ||
| return new SpringStreamMessageProcessor(this); | ||
| } | ||
|
|
||
| /** | ||
| * Validate whether the fields contained in this Builder as set accordingly. | ||
| * | ||
| * @throws AxonConfigurationException if one field is asserted to be incorrect according to the Builder's specifications | ||
| */ | ||
| protected void validate() { | ||
| assertNonNull(messageSource, "The MessageSource is a hard requirement and should be provided"); | ||
| assertNonNull(converter, "The Converter is a hard requirement and should be provided"); | ||
| } | ||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is there an actual need to have a separate
spring-messagingandspring-integrationproperty, or could we tag along withspring.version?