Set message metadata via Reactor context

.github/dependabot.yml

@@ -0,0 +1,16 @@
+version: 2
+updates:
+- package-ecosystem: maven
+  directory: "/"
+  schedule:
+    interval: daily
+  open-pull-requests-limit: 5
+  # Specify labels for pull requests
+  labels:
+    - "Type: Dependency Upgrade"
+    - "Priority 1: Must"
+  # Add reviewers
+  reviewers:
+    - "m1l4n54v1c"
+    - "saratry"
+    - "stefanvozd"

.travis.yml

@@ -1,32 +1,41 @@
-# We want to run this on Ubuntu 18.04 LTS
 dist: bionic
 
 language: java
 
 sudo: required
 
 services:
-  - docker
-
-jobs:
-  include:
-    - name: "JDK 8"
-      jdk: openjdk8
-      env: MVN_BUILD_SCRIPT="clean verify"
-    - name: "JDK 11 with Sonar analysis"
-      jdk: openjdk11
-      env: MVN_BUILD_SCRIPT="-Pcoverage clean verify sonar:sonar -Dsonar.projectKey=AxonFramework_extension-reactor"
+- docker
+
+#jobs:
+#  include:
+#  - name: JDK 8
+#    jdk: openjdk8
+#    env: MVN_BUILD_SCRIPT="clean verify"
+#  - name: JDK 11 with Sonar analysis
+#    jdk: openjdk11
+#    env: MVN_BUILD_SCRIPT="-Pcoverage clean verify sonar:sonar -Dsonar.projectKey=AxonFramework_extension-reactor"
+#install: true
+#cache:
+#  directories:
+#  - "$HOME/.m2"
+#script:
+#  - mvn -version
+#  - mvn -B -U $MVN_BUILD_SCRIPT
+jdk:
+  - openjdk8
 
 install: true
 
 cache:
   directories:
     - $HOME/.m2
 
-# We switch JDK by setting the JAVA_HOME environment variable. All JDKs are under the same base directory.
 script:
   - mvn -version
-  - mvn -B -U $MVN_BUILD_SCRIPT
+  - mvn -B clean verify
+  - JAVA_HOME=$(dirname ${JAVA_HOME})/openjdk11 mvn -version
+  - JAVA_HOME=$(dirname ${JAVA_HOME})/openjdk11 mvn -B -Pcoverage clean verify sonar:sonar -Dsonar.projectKey=AxonFramework_extension-reactor
 
 notifications:
   slack:
@@ -36,4 +45,4 @@ addons:
   sonarcloud:
     organization: "axonframework"
     token:
-      secure: "htYEusSkwB7bvgURS8FU8CBFmIlJcfSVhA6dZQ5uF3m3KYltT8D6zC4/O9aRw/2dId0lPMC5T6XiRLkmJsa3hLJhsbnrOW+8UXyzWq6SZfuVngRAXKeo5sqrv+7XXyIir5kbq0WSgs4UBJfxUIIDynfDupE9Y4MfZAQRA0FrupkEwwV4sp+QQKIymYFO84iPgwvlHpcn+KDMI0LNtLa6+kKJuuL5u4HsKHA4tTWBDP5aSBZL2dRNt1YM8+poO3pXH+fqTW21X7qq2GfoDfyLdfWtodT9TfREBQWgJcX6dXc3yLruPMnY7wzvNZUVF7bDDjIg9rcfARLLyBDutDGtiovqilKjes5JbsjEweRw9CpLq86UqIN1Sn1dHf40ieuvuU6hDyOCRAYwbP2/DZtN5eFyVJgUKY4NpcAewOODI3KMVTrhEeVjET8uvFsKT1Phsph3Ii/qhamED5+Yqvu+0974hQq0C3jXtDaM9xKC9ViaK1Bw1Q8P6EwGUvTEDFzsEPsSpg03G0EXQoOlikk5AbgjIEhDzeD9/UQvXv8qL5DM9N9nD/fH8ELa0ma5GHzMwK9WrkjjsYmeppjyT/wJHvb0MHAZxkJvq4izHvAZmvkusBEAOGpZHGRhDdGSR59zDMT1B7d8SLLk8OH9bFVEYh7rtFeNzwl6mzaGPFO7aWc="
+      secure: "Byhy5L5cGZ7dlymrXg6/5SRSmCSbBLq2DbmBDHnWKTq2TfbhCgNvkrjI7R4i4xwkvIcNEuFblULUzGSz7+lGcTxpNsMt9MZKxnH8Hj+PAWNOLdHQGty9+EXyQCF9q4x+td/Sqp/EuCFBYvDyVCVCIVbyMKe6bD9J6hfC8QWpK1Q3eDOVGI9N9AtawkDVxoRYH2Br4tkxrSAQcNA8w53oqycNPotDeIQS0K/780AULXzb7i8fYO/Uz4x+pnh348O35009WTS2eUU8yjoZ4UJmBJ1ocN3Xtj7NyllPPr9e5QZfPi+xAz2YecWLSrwpyPiS4nPmzAyPcYtohYc/HiSntZvP9mZYmCVPb4GbZAIPrgXSjtPX2RMd9FVwLH2PJ2VX4uxQ2B9maCuETj7zV8CW6zCUNJJMRtpHzIvNY/Nj+9H8n0EKepMKFxU0uWq34JJcWTvLX/fITGs6KljNivMq/w3Hue/akW2b9XoqLzfryTUYIFShZWp/rhQdset67ReBvtLFE+lTH+XoNkvd+psV37j51nmnnFUNEfITpvIi+3zStytCK213coyNU2cKCaa93Jfvgt90OJBNBf9M1RpP+oyQ++kSWkUFY8u/QMW1walR8sFlKmkw6NRaupgEc/QpbkGzSdcT5laiDrhxisNsaRCTuF42QmlT9nQhP0ZG+EQ="

README.md

@@ -31,9 +31,10 @@ There are a couple of things to consider when you're traversing anything Axon:
  as the majority of possible scenarios you might encounter when using Axon should be covered there.
 * If the Reference Guide does not cover a specific topic you would've expected,
  we'd appreciate if you could file an [issue](https://github.com/AxonIQ/reference-guide/issues) about it for us. 
-* There is a a [public mailing list](https://groups.google.com/forum/#!forum/axonframework) to support you in the case 
- the reference guide did not sufficiently answer your question.
-* Next to the mailing list we also monitor Stack Overflow for any questions which are tagged with `axon`.
+* There is a [forum](https://discuss.axoniq.io/) to support you in the case the reference guide did not sufficiently answer your question.
+Axon Framework and Server developers will help out on a best effort basis.
+Know that any support from contributors on posted question is very much appreciated on the forum.
+* Next to the forum we also monitor Stack Overflow for any questions which are tagged with `axon`.
 
 ## Feature requests and issue reporting
 

axon-reactor-spring-boot-autoconfigure/pom.xml

@@ -19,7 +19,7 @@
     <parent>
         <artifactId>axon-reactor-parent</artifactId>
         <groupId>org.axonframework.extensions.reactor</groupId>
-        <version>4.4.3-SNAPSHOT</version>
+        <version>4.5-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

axon-reactor-spring-boot-starter/pom.xml

@@ -26,7 +26,7 @@
 
     <groupId>org.axonframework.extensions.reactor</groupId>
     <artifactId>axon-reactor-spring-boot-starter</artifactId>
-    <version>4.4.3-SNAPSHOT</version>
+    <version>4.5-SNAPSHOT</version>
 
     <name>Axon Framework Reactor Extension - Spring Boot Starter</name>
     <description>
@@ -125,7 +125,7 @@
             </plugin>
             <plugin>
                 <artifactId>maven-javadoc-plugin</artifactId>
-                <version>2.10.4</version>
+                <version>3.2.0</version>
                 <executions>
                     <execution>
                         <id>attach-javadoc</id>

axon-reactor/pom.xml

@@ -19,7 +19,7 @@
     <parent>
         <artifactId>axon-reactor-parent</artifactId>
         <groupId>org.axonframework.extensions.reactor</groupId>
-        <version>4.4.3-SNAPSHOT</version>
+        <version>4.5-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

axon-reactor/src/main/java/org/axonframework/extensions/reactor/commandhandling/gateway/DefaultReactorCommandGateway.java

@@ -1,17 +1,14 @@
 package org.axonframework.extensions.reactor.commandhandling.gateway;
 
-import org.axonframework.commandhandling.CommandBus;
-import org.axonframework.commandhandling.CommandCallback;
-import org.axonframework.commandhandling.CommandMessage;
-import org.axonframework.commandhandling.CommandResultMessage;
-import org.axonframework.commandhandling.GenericCommandMessage;
+import org.axonframework.commandhandling.*;
 import org.axonframework.commandhandling.gateway.RetryScheduler;
 import org.axonframework.commandhandling.gateway.RetryingCallback;
 import org.axonframework.common.AxonConfigurationException;
 import org.axonframework.common.Registration;
 import org.axonframework.extensions.reactor.messaging.ReactorMessageDispatchInterceptor;
 import org.axonframework.extensions.reactor.messaging.ReactorResultHandlerInterceptor;
 import org.axonframework.extensions.reactor.commandhandling.callbacks.ReactorCallback;
+import org.axonframework.messaging.MetaData;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -72,13 +69,27 @@ public static Builder builder() {
     @Override
     public <R> Mono<R> send(Object command) {
         //noinspection unchecked
-        return Mono.<CommandMessage<?>>just(GenericCommandMessage.asCommandMessage(command))
+        return createCommandMessage(command)
                 .transform(this::processCommandInterceptors)
                 .flatMap(this::dispatchCommand)
                 .flatMap(this::processResultsInterceptors)
                 .transform(this::getPayload);
     }
 
+    private Mono<CommandMessage<?>> createCommandMessage(Object command) {
+        return Mono.just(command)
+                .zipWith(metaDataFromContext())
+                .map(commandAndMeta -> GenericCommandMessage.asCommandMessage(commandAndMeta.getT1())
+                        .andMetaData(commandAndMeta.getT2()));
+    }
+
+    private Mono<MetaData> metaDataFromContext() {
+        return Mono.subscriberContext()
+                .handle((ctx,sink) -> sink.next(Objects.requireNonNull(
+                        ctx.getOrDefault(MetaData.class, MetaData.emptyInstance())
+                )));
+    }
+
     @Override
     public Registration registerDispatchInterceptor(ReactorMessageDispatchInterceptor<CommandMessage<?>> interceptor) {
         dispatchInterceptors.add(interceptor);
@@ -100,19 +111,23 @@ private Mono<CommandMessage<?>> processCommandInterceptors(Mono<CommandMessage<?
 
     private <C, R> Mono<Tuple2<CommandMessage<C>, Flux<CommandResultMessage<? extends R>>>> dispatchCommand(
             CommandMessage<C> commandMessage) {
-        ReactorCallback<C, R> reactorCallback = new ReactorCallback<>();
-        CommandCallback<C, R> callback = reactorCallback;
-        if (retryScheduler != null) {
-            callback = new RetryingCallback<>(callback, retryScheduler, commandBus);
-        }
-        commandBus.dispatch(commandMessage, callback);
-        return Mono.just(commandMessage).zipWith(Mono.just(Flux.from(reactorCallback)));
+        return Mono.defer(()-> {
+            ReactorCallback<C, R> reactorCallback = new ReactorCallback<>();
+            CommandCallback<C, R> callback = reactorCallback;
+            if (retryScheduler != null) {
+                callback = new RetryingCallback<>(callback, retryScheduler, commandBus);
+            }
+            commandBus.dispatch(commandMessage, callback);
+            return Mono.just(commandMessage).zipWith(Mono.just(Flux.from(reactorCallback)));
+        });
     }
 
     private <C> Mono<? extends CommandResultMessage<?>> processResultsInterceptors(
             Tuple2<CommandMessage<C>, Flux<CommandResultMessage<?>>> commandWithResults) {
         CommandMessage<?> commandMessage = commandWithResults.getT1();
-        Flux<CommandResultMessage<?>> commandResultMessages = commandWithResults.getT2();
+        Flux<CommandResultMessage<?>> commandResultMessages = commandWithResults.getT2()
+                .flatMapSequential(this::mapExceptionalResult);
+
         return Flux.fromIterable(resultInterceptors)
                    .reduce(commandResultMessages,
                            (result, interceptor) -> interceptor.intercept(commandMessage, result))
@@ -126,6 +141,10 @@ private <R> Mono<R> getPayload(Mono<? extends CommandResultMessage<?>> commandRe
                 .map(it -> (R) it.getPayload());
     }
 
+    private Mono<? extends CommandResultMessage<?>> mapExceptionalResult(CommandResultMessage<?> response) {
+        return response.isExceptional() ? Mono.error(response.exceptionResult()) : Mono.just(response);
+    }
+
     /**
      * Builder class to instantiate {@link DefaultReactorCommandGateway}.
      * <p>

axon-reactor/src/main/java/org/axonframework/extensions/reactor/commandhandling/gateway/ReactorCommandGateway.java

@@ -52,4 +52,5 @@ default Flux<Object> sendAll(Publisher<?> commands) {
         return Flux.from(commands)
                    .concatMap(this::send);
     }
+
 }

axon-reactor/src/main/java/org/axonframework/extensions/reactor/eventhandling/gateway/DefaultReactorEventGateway.java

@@ -6,10 +6,12 @@
 import org.axonframework.eventhandling.EventMessage;
 import org.axonframework.eventhandling.GenericEventMessage;
 import org.axonframework.extensions.reactor.messaging.ReactorMessageDispatchInterceptor;
+import org.axonframework.messaging.MetaData;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CopyOnWriteArrayList;
 
 import static java.util.Arrays.asList;
@@ -57,11 +59,25 @@ public static Builder builder() {
     @Override
     public Flux<EventMessage<?>> publish(List<?> events) {
         return Flux.fromIterable(events)
-                   .map(event -> Mono.<EventMessage<?>>just(GenericEventMessage.asEventMessage(event)))
+                   .map(this::createEventMessage)
                    .flatMap(this::processEventInterceptors)
                    .flatMap(this::publishEvent);
     }
 
+    private Mono<EventMessage<?>> createEventMessage(Object event) {
+        return Mono.just(event)
+                .zipWith(metaDataFromContext())
+                .map(eventAndMeta -> GenericEventMessage.asEventMessage(eventAndMeta.getT1())
+                        .andMetaData(eventAndMeta.getT2()));
+    }
+
+    private Mono<MetaData> metaDataFromContext() {
+        return Mono.subscriberContext()
+                .handle((ctx,sink) -> sink.next(Objects.requireNonNull(
+                        ctx.getOrDefault(MetaData.class, MetaData.emptyInstance())
+                )));
+    }
+
     @Override
     public Registration registerDispatchInterceptor(ReactorMessageDispatchInterceptor<EventMessage<?>> interceptor) {
         dispatchInterceptors.add(interceptor);

axon-reactor/src/main/java/org/axonframework/extensions/reactor/queryhandling/gateway/DefaultReactorQueryGateway.java

@@ -4,18 +4,10 @@
 import org.axonframework.common.Registration;
 import org.axonframework.extensions.reactor.messaging.ReactorMessageDispatchInterceptor;
 import org.axonframework.extensions.reactor.messaging.ReactorResultHandlerInterceptor;
+import org.axonframework.messaging.MetaData;
 import org.axonframework.messaging.ResultMessage;
 import org.axonframework.messaging.responsetypes.ResponseType;
-import org.axonframework.queryhandling.DefaultSubscriptionQueryResult;
-import org.axonframework.queryhandling.GenericQueryMessage;
-import org.axonframework.queryhandling.GenericSubscriptionQueryMessage;
-import org.axonframework.queryhandling.QueryBus;
-import org.axonframework.queryhandling.QueryMessage;
-import org.axonframework.queryhandling.QueryResponseMessage;
-import org.axonframework.queryhandling.SubscriptionQueryBackpressure;
-import org.axonframework.queryhandling.SubscriptionQueryMessage;
-import org.axonframework.queryhandling.SubscriptionQueryResult;
-import org.axonframework.queryhandling.SubscriptionQueryUpdateMessage;
+import org.axonframework.queryhandling.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -91,16 +83,29 @@ public Registration registerResultHandlerInterceptor(
 
     @Override
     public <R, Q> Mono<R> query(String queryName, Q query, ResponseType<R> responseType) {
-        return Mono.<QueryMessage<?, ?>>fromCallable(() -> new GenericQueryMessage<>(asMessage(query),
-                                                                                     queryName,
-                                                                                     responseType))
+        return createQueryMessage(queryName, query, responseType)
                 .transform(this::processDispatchInterceptors)
                 .flatMap(this::dispatchQuery)
                 .flatMapMany(this::processResultsInterceptors)
                 .<R>transform(this::getPayload)
                 .next();
     }
 
+
+    public <R, Q> Mono<QueryMessage<?, ?>> createQueryMessage(String queryName, Q query, ResponseType<R> responseType) {
+        return Mono.fromCallable(() -> new GenericQueryMessage<>(asMessage(query), queryName, responseType))
+                .zipWith(metaDataFromContext())
+                .map(queryAndMeta -> queryAndMeta.getT1().andMetaData(queryAndMeta.getT2()));
+    }
+
+
+    private Mono<MetaData> metaDataFromContext() {
+        return Mono.subscriberContext()
+                .handle((ctx,sink) -> sink.next(Objects.requireNonNull(
+                        ctx.getOrDefault(MetaData.class, MetaData.emptyInstance())
+                )));
+    }
+
     @Override
     public <R, Q> Flux<R> scatterGather(String queryName, Q query, ResponseType<R> responseType, Duration timeout) {
         return Mono.<QueryMessage<?, ?>>fromCallable(() -> new GenericQueryMessage<>(asMessage(query),

axon-reactor/src/test/java/org/axonframework/extensions/reactor/commandhandling/gateway/DefaultReactorCommandGatewayComponentTest.java

@@ -7,7 +7,10 @@
 import org.axonframework.commandhandling.gateway.RetryScheduler;
 import org.axonframework.common.Registration;
 import org.axonframework.messaging.MessageHandler;
+import org.axonframework.messaging.MetaData;
+import org.axonframework.queryhandling.QueryMessage;
 import org.junit.jupiter.api.*;
+import org.mockito.ArgumentCaptor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
@@ -23,6 +26,7 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
+import static reactor.util.context.Context.of;
 
 /**
  * Tests for {@link DefaultReactorCommandGateway} all together with Command Bus.
@@ -100,7 +104,7 @@ void testSendContext() throws Exception {
         Mono<String> result = reactiveCommandGateway.send("command");
         verifyNoMoreInteractions(commandMessageHandler);
 
-        Context context = Context.of("k1", "v1");
+        Context context = of("k1", "v1");
 
         StepVerifier.create(result.subscriberContext(context))
                     .expectNext("handled")
@@ -112,6 +116,33 @@ void testSendContext() throws Exception {
         verifyNoMoreInteractions(mockRetryScheduler);
     }
 
+    @Test
+    void testQuerySetMetaDataViaContext() throws Exception {
+        Mono<String> result = reactiveCommandGateway.send("command");
+        verifyNoMoreInteractions(commandMessageHandler);
+
+        Context context = of(MetaData.class, MetaData.with("k","v"));
+
+        StepVerifier.create(result.subscriberContext(context))
+                .expectNext("handled")
+                .expectAccessibleContext()
+                .containsOnly(context)
+                .then()
+                .verifyComplete();
+
+        ArgumentCaptor<CommandMessage> commandMessageCaptor =  ArgumentCaptor.forClass(CommandMessage.class);
+
+
+        verify(commandBus).dispatch(commandMessageCaptor.capture(),any());
+        CommandMessage commandMessage = commandMessageCaptor.getValue();
+
+        assertTrue(commandMessage.getMetaData().containsKey("k"));
+        assertTrue(commandMessage.getMetaData().containsValue("v"));
+
+        verify(commandMessageHandler).handle(any());
+        verifyNoMoreInteractions(mockRetryScheduler);
+    }
+
     @Test
     void testSendVoidHandler() throws Exception {
         Mono<String> result = reactiveCommandGateway.send(1L);
@@ -186,7 +217,7 @@ void testSendWithDispatchInterceptor() {
 
     @Test
     void testSendWithDispatchInterceptorWithContext() {
-        Context context = Context.of("security", true);
+        Context context = of("security", true);
 
         reactiveCommandGateway
                 .registerDispatchInterceptor(cmdMono -> cmdMono
@@ -207,7 +238,7 @@ void testSendWithDispatchInterceptorWithContext() {
 
     @Test
     void testSendWithDispatchInterceptorWithContextFiltered() {
-        Context context = Context.of("security", false);
+        Context context = of("security", false);
 
         reactiveCommandGateway
                 .registerDispatchInterceptor(cmdMono -> cmdMono