Change Rabbit Message delivery Mode (2)

Author: Daniel Bustamante Ospina

async/async-commons/src/main/java/org/reactivecommons/async/impl/communications/ReactiveMessageSender.java

@@ -51,7 +51,7 @@ private AMQP.BasicProperties buildMessageProperties(Message message, Map<String,
             .contentType(properties.getContentType())
             .appId(sourceApplication)
             .contentEncoding(properties.getContentEncoding())
-            .deliveryMode(1)
+            .deliveryMode(2)
             .timestamp(new Date())
             .messageId(UUID.randomUUID().toString())
             .headers(baseHeaders).build();

async/async-commons/src/main/java/org/reactivecommons/async/impl/converters/JacksonMessageConverter.java

@@ -73,7 +73,7 @@ public <T> T readValue(Message message, Class<T> valueClass) {
     public <T> Command<T> readCommandStructure(Message message) {
         try {
             final CommandJson commandJson = objectMapper.readValue(message.getBody(), CommandJson.class);
-            return new Command<>(commandJson.getName(), commandJson.getCommandId(), null);
+            return new Command<>(commandJson.getName(), commandJson.getCommandId(), (T)commandJson.getData());
         } catch (IOException e) {
             throw new MessageConversionException("Failed to convert Message content", e);
         }

async/async-commons/src/main/java/org/reactivecommons/async/impl/converters/MessageConverter.java

@@ -15,6 +15,8 @@ public interface MessageConverter {
 
     <T> T readValue(Message message, Class<T> valueClass);
 
+    <T> Command<T> readCommandStructure(Message message);
+
     Message toMessage(Object object);
 
 }

async/async-commons/src/main/java/org/reactivecommons/async/impl/listeners/ApplicationCommandListener.java

@@ -51,9 +51,8 @@ protected Function<Message, Mono<Object>> rawMessageHandler(String executorPath)
         return msj -> executor.execute(msj).cast(Object.class);
     }
 
-    //TODO: replace with interface
     protected String getExecutorPath(AcknowledgableDelivery msj) {
-        final Command<Object> command = ((JacksonMessageConverter) messageConverter).readCommandStructure(RabbitMessage.fromDelivery(msj));
+        final Command<Object> command = messageConverter.readCommandStructure(RabbitMessage.fromDelivery(msj));
         return command.getName();
     }
 

domain/domain-events/src/main/java/org/reactivecommons/api/domain/Command.java

@@ -1,5 +1,6 @@
 package org.reactivecommons.api.domain;
 
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 

samples/async/receiver-responder/src/main/resources/application.properties

@@ -1,2 +1 @@
-app.async.reply.prefix=sample-receiver.
 spring.application.name=Receiver2

samples/async/sender-client/src/main/java/sample/SampleSenderApp.java

@@ -1,6 +1,7 @@
 package sample;
 
 import lombok.extern.java.Log;
+import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.impl.config.annotations.EnableDirectAsyncGateway;
@@ -25,6 +26,8 @@ public static void main(String[] args) throws InterruptedException {
 
     @Bean
     public CommandLineRunner run(MemberRegistrySender sender, DirectAsyncGateway asyncGateway) {
+        Command<String> command0 = new Command<>("test", "01", "Daniel");
+        asyncGateway.sendCommand(command0, "Receiver2").repeat(5).subscribe();
         return args -> Flux.interval(Duration.ofSeconds(1)).concatMap(n -> {
             AddMemberCommand command = new AddMemberCommand("Daniel " + n, n+"");
             return asyncGateway.requestReply(new AsyncQuery<>("serveQuery.empty", "test"), "Receiver2", String.class)

samples/async/sender-client/src/main/resources/application.properties

@@ -1,6 +1 @@
-app.async.reply.prefix=sender-client.
-spring.application.name=Sender
-
-spring.cloud.stream.bindings.input.group=sample1
-spring.cloud.stream.bindings.input.destination=Sender
-spring.cloud.stream.bindings.input.consumer.concurrency=12
+spring.application.name=Sender