Update props

juancgalvis · juancgalvis · commit 53f9716e4c70 · 2023-06-30T06:09:14.000-05:00
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/DirectAsyncGatewayConfig.java
@@ -8,13 +8,16 @@
 import org.reactivecommons.async.commons.converters.MessageConverter;
 import org.reactivecommons.async.commons.reply.ReactiveReplyRouter;
 import org.reactivecommons.async.rabbit.RabbitDirectAsyncGateway;
+import org.reactivecommons.async.rabbit.config.props.AsyncProps;
 import org.reactivecommons.async.rabbit.config.props.BrokerConfigProps;
 import org.reactivecommons.async.rabbit.listeners.ApplicationReplyListener;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 import static org.reactivecommons.async.api.HandlerRegistry.DEFAULT_DOMAIN;
 
 @Configuration
@@ -29,11 +32,22 @@ public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, Re
         return new RabbitEDADirectAsyncGateway(config, router, manager, props.getDirectMessagesExchangeName(), converter, meterRegistry);
     }
 
-    @Bean // TODO: Listen replies from n domains if enabled in config
-    public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ConnectionManager manager) {
-        final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(DEFAULT_DOMAIN), props.getReplyQueue());
-        replyListener.startListening(config.getRoutingKey());
-        return replyListener;
+    @Bean
+    public ApplicationReplyListener msgListener(ReactiveReplyRouter router, AsyncProps asyncProps, BrokerConfig config, ConnectionManager manager) {
+        asyncProps.getListenRepliesFrom().add(DEFAULT_DOMAIN);
+        AtomicReference<ApplicationReplyListener> localListener = new AtomicReference<>();
+
+        asyncProps.getListenRepliesFrom().forEach(domain -> {
+
+            final ApplicationReplyListener replyListener = new ApplicationReplyListener(router, manager.getListener(domain), props.getReplyQueue());
+            replyListener.startListening(config.getRoutingKey());
+
+            if (DEFAULT_DOMAIN.equals(domain)) {
+                localListener.set(replyListener);
+            }
+        });
+
+        return localListener.get();
     }
 
     @Bean
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/AsyncProps.java
@@ -6,6 +6,8 @@
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 
@@ -33,6 +35,8 @@ public class AsyncProps {
     @NestedConfigurationProperty
     private Map<String, RabbitProperties> connections = new TreeMap<>();
 
+    private List<String> listenRepliesFrom = new ArrayList<>();
+
     private Integer maxRetries = 10;
 
     private Integer prefetchCount = 250;
diff --git a/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/BrokerConfigProps.java b/async/async-rabbit-starter-eda/src/main/java/org/reactivecommons/async/rabbit/config/props/BrokerConfigProps.java
@@ -39,7 +39,7 @@ public String getCommandsQueue() {
     public String getReplyQueue() {
         final String name = replyQueueName.get();
         if (name == null) {
-            final String replyName = NameGenerator.generateNameFrom(appName);
+            final String replyName = NameGenerator.generateNameFrom(appName, "replies");
             if (replyQueueName.compareAndSet(null, replyName)) {
                 return replyName;
             } else {
diff --git a/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java b/samples/async/receiver-responder/src/main/java/sample/SampleReceiverApp.java
@@ -11,6 +11,7 @@
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.api.HandlerRegistry;
 import org.reactivecommons.async.api.handlers.QueryHandler;
+import org.reactivecommons.async.impl.config.annotations.EnableCommandListeners;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableDomainEventBus;
 import org.reactivecommons.async.impl.config.annotations.EnableEventListeners;
@@ -31,6 +32,7 @@
 
 @SpringBootApplication
 @EnableEventListeners
+@EnableCommandListeners
 @EnableDomainEventBus
 @EnableDirectAsyncGateway
 @Log
@@ -60,9 +62,9 @@ public HandlerRegistry handlerRegistrySubs(/*DirectAsyncGateway gateway*/) {
         return HandlerRegistry.register()
 
 //                .handleDynamicEvents("dynamic.*", message -> Mono.empty(), Object.class)
-                .listenEvent("fixed.event", message -> Mono.empty(), Object.class)
-                .listenDomainEvent("accounts", "account.created", message -> Mono.empty(), Object.class)
-                .listenDomainEvent("deposits", "transfer.xxx", message -> Mono.empty(), Object.class)
+//                .listenEvent("fixed.event", message -> Mono.empty(), Object.class)
+//                .listenDomainEvent("accounts", "account.created", message -> Mono.empty(), Object.class)
+//                .listenDomainEvent("deposits", "transfer.xxx", message -> Mono.empty(), Object.class)
 //                .serveQuery("query1", message -> {
 //                    log.info("resolving from direct query");
 //                    return just(new RespQuery1("Ok", message));
@@ -76,11 +78,11 @@ public HandlerRegistry handlerRegistrySubs(/*DirectAsyncGateway gateway*/) {
 //                    return gateway.reply(new RespQuery1("Ok", message), from).then();
 //                }, Call.class);
 
-                .handleDynamicEvents("dynamic.*", message -> Mono.empty(), Object.class)
-                .listenEvent("event", message -> {
-                    log.info(message.getData().toString());
-                    return Mono.empty();
-                }, CloudEvent.class)
+//                .handleDynamicEvents("dynamic.*", message -> Mono.empty(), Object.class)
+//                .listenEvent("event", message -> {
+//                    log.info(message.getData().toString());
+//                    return Mono.empty();
+//                }, CloudEvent.class)
                 .handleCommand("command", message -> {
                     log.info(message.getData().toString());
                     return Mono.empty();
@@ -96,11 +98,11 @@ public HandlerRegistry handlerRegistrySubs(/*DirectAsyncGateway gateway*/) {
                             .withData("application/json", CloudEventBuilderExt.asBytes(mapData))
                             .build();
                     return just(response);
-                }, CloudEvent.class)
-                .serveQuery("sample.query.*", message -> {
-                    log.info("resolving from direct query");
-                    return just(new RespQuery1("Ok", message));
-                }, Call.class);
+                }, CloudEvent.class);
+//                .serveQuery("sample.query.*", message -> {
+//                    log.info("resolving from direct query");
+//                    return just(new RespQuery1("Ok", message));
+//                }, Call.class);
                 /*.serveQuery("query2", (from, message) -> {
                     log.info("resolving from delegate query");
                     return gateway.reply(new RespQuery1("Ok", message), from).then();
diff --git a/samples/async/receiver-responder/src/main/resources/application.yaml b/samples/async/receiver-responder/src/main/resources/application.yaml
@@ -10,8 +10,8 @@ app:
     retry-delay: 1000 # son milisegundos
     connections:
       app:
-        virtualHost: app1
-      accounts:
-        virtualHost: domain-a
-      deposits:
         virtualHost: domain-a
+#      accounts:
+#        virtualHost: domain-a
+#      deposits:
+#        virtualHost: domain-a
diff --git a/samples/async/sender-client/src/main/java/sample/SampleRestController.java b/samples/async/sender-client/src/main/java/sample/SampleRestController.java
@@ -47,7 +47,7 @@ public Mono<CloudEvent> sampleService(@RequestBody Call call) throws JsonProcess
                 .withData("application/json", CloudEventBuilderExt.asBytes(call))
                 .build();
 
-        return directAsyncGateway.requestReply(query, target, CloudEvent.class);
+        return directAsyncGateway.requestReply(query, target, CloudEvent.class, "accounts");
     }
     @PostMapping(path = "/sample/event", consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
     public Mono<String> sampleServiceEvent(@RequestBody Call call) throws JsonProcessingException {
@@ -74,7 +74,7 @@ public Mono<String> sampleServiceCommand(@RequestBody Call call) throws JsonProc
                 .withData("application/json", CloudEventBuilderExt.asBytes(call))
                 .build();
 
-        return directAsyncGateway.sendCommand(command, target).thenReturn("command");
+        return directAsyncGateway.sendCommand(command, target, "accounts").thenReturn("command");
     }
     @PostMapping(path = "/sample/match", consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
     public Mono<RespQuery1> sampleServices(@RequestBody Call call) {
diff --git a/samples/async/sender-client/src/main/resources/application.yaml b/samples/async/sender-client/src/main/resources/application.yaml
@@ -15,9 +15,10 @@ management:
         include: health,prometheus
 app:
   async:
+    listenRepliesFrom: [accounts]
     connections:
       app:
-        virtualHost: app1
+        virtualHost: /
       accounts:
         virtualHost: domain-a
 
