Use MessageConverter in replyListener

Fixes: gh-10, gh-11

Author: Daniel Bustamante Ospina

async/async-commons-standalone/src/main/java/org/reactivecommons/async/impl/config/DirectAsyncGatewayConfig.java

@@ -3,6 +3,7 @@
 import org.reactivecommons.async.impl.RabbitDirectAsyncGateway;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
+import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationReplyListener;
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
 
@@ -20,8 +21,8 @@ public DirectAsyncGatewayConfig(String directMessagesExchangeName, String appNam
         this.appName = appName;
     }
 
-    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName);
+    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter) throws Exception {
+        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName, converter);
     }
 
     public ApplicationReplyListener msgListener(ReactiveReplyRouter router, BrokerConfig config, ReactiveMessageListener listener)  {

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/ReplyCommandSender.java

@@ -18,8 +18,6 @@ public abstract class ReplyCommandSender {
     @Autowired
     DirectAsyncGateway asyncGateway;
 
-    private final ObjectMapper mapper = new ObjectMapper();
-
     protected <R, C> Mono<R> sendCommand(C command, String commandId, Class<R> type) {
         AsyncQuery<C> asyncQuery = new AsyncQuery<C>(commandId, command);
         return asyncGateway.requestReply(asyncQuery, target(), type);

async/async-commons-starter/src/main/java/org/reactivecommons/async/impl/config/DirectAsyncGatewayConfig.java

@@ -5,6 +5,7 @@
 import org.reactivecommons.async.impl.RabbitDirectAsyncGateway;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
+import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.listeners.ApplicationReplyListener;
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
 import org.springframework.beans.factory.annotation.Value;
@@ -14,7 +15,6 @@
 import org.springframework.util.Base64Utils;
 
 import java.nio.ByteBuffer;
-import java.util.Base64;
 import java.util.UUID;
 
 @Configuration
@@ -30,8 +30,8 @@ public class DirectAsyncGatewayConfig {
     private String appName;
 
     @Bean
-    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender) throws Exception {
-        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName);
+    public RabbitDirectAsyncGateway rabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender rSender, MessageConverter converter) throws Exception {
+        return new RabbitDirectAsyncGateway(config, router, rSender, directMessagesExchangeName, converter);
     }
 
     @Bean

async/async-commons/src/main/java/org/reactivecommons/async/impl/RabbitDirectAsyncGateway.java

@@ -1,11 +1,11 @@
 package org.reactivecommons.async.impl;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.reactivecommons.api.domain.Command;
 import org.reactivecommons.async.api.AsyncQuery;
 import org.reactivecommons.async.api.DirectAsyncGateway;
 import org.reactivecommons.async.impl.communications.ReactiveMessageSender;
 import org.reactivecommons.async.impl.config.BrokerConfig;
+import org.reactivecommons.async.impl.converters.MessageConverter;
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
 import reactor.core.publisher.Mono;
 
@@ -20,19 +20,19 @@
 
 public class RabbitDirectAsyncGateway implements DirectAsyncGateway {
 
-    private final ObjectMapper mapper = new ObjectMapper();
-
     private final BrokerConfig config;
     private final ReactiveReplyRouter router;
     private final ReactiveMessageSender sender;
     private final String exchange;
+    private final MessageConverter converter;
 
 
-    public RabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender sender, String exchange) {
+    public RabbitDirectAsyncGateway(BrokerConfig config, ReactiveReplyRouter router, ReactiveMessageSender sender, String exchange, MessageConverter converter) {
         this.config = config;
         this.router = router;
         this.sender = sender;
         this.exchange = exchange;
+        this.converter = converter;
     }
 
     @Override
@@ -46,7 +46,7 @@ public <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class
 
         final Mono<R> replyHolder = router.register(correlationID)
             .timeout(Duration.ofSeconds(15))
-            .flatMap(s -> fromCallable(() -> String.class.equals(type) ? type.cast(s) : mapper.readValue(s, type)));
+            .flatMap(s -> fromCallable(() -> (R)converter.readAsyncQuery(s, type)));
 
         Map<String, Object> headers = new HashMap<>();
         headers.put(REPLY_ID, config.getRoutingKey());

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

@@ -60,6 +60,15 @@ public <T> Command<T> readCommand(Message message, Class<T> bodyClass) {
         }
     }
 
+    @Override
+    public <T> T readValue(Message message, Class<T> valueClass) {
+        try {
+            return objectMapper.readValue(message.getBody(), valueClass);
+        } catch (IOException e) {
+            throw new MessageConversionException("Failed to convert Message content", e);
+        }
+    }
+
     //TODO: pull definition up to interface
     public <T> Command<T> readCommandStructure(Message message) {
         try {

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

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

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

@@ -1,6 +1,8 @@
 package org.reactivecommons.async.impl.listeners;
 
 import lombok.extern.java.Log;
+import org.reactivecommons.async.impl.RabbitMessage;
+import org.reactivecommons.async.impl.communications.Message;
 import org.reactivecommons.async.impl.communications.ReactiveMessageListener;
 import org.reactivecommons.async.impl.communications.TopologyCreator;
 import org.reactivecommons.async.impl.reply.ReactiveReplyRouter;
@@ -37,7 +39,7 @@ public void startListening(String routeKey) {
                     if (isEmpty) {
                         router.routeEmpty(correlationID);
                     } else {
-                        router.routeReply(correlationID, new String(delivery.getBody()));
+                        router.routeReply(correlationID, RabbitMessage.fromDelivery(delivery));
                     }
                 } catch (Exception e) {
                     log.log(Level.SEVERE, "Error in reply reception", e);

async/async-commons/src/main/java/org/reactivecommons/async/impl/reply/ReactiveReplyRouter.java

@@ -1,37 +1,38 @@
 package org.reactivecommons.async.impl.reply;
 
+import org.reactivecommons.async.impl.communications.Message;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.UnicastProcessor;
 import reactor.util.concurrent.Queues;
 
 import java.util.concurrent.ConcurrentHashMap;
 
 public class ReactiveReplyRouter {
-    private final ConcurrentHashMap<String, UnicastProcessor<String>> processors = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<String, UnicastProcessor<Message>> processors = new ConcurrentHashMap<>();
 
-    public Mono<String> register(String correlationID) {
-        final UnicastProcessor<String> processor = UnicastProcessor.create(Queues.<String>one().get());
+    public Mono<Message> register(String correlationID) {
+        final UnicastProcessor<Message> processor = UnicastProcessor.create(Queues.<Message>one().get());
         processors.put(correlationID, processor);
         return processor.singleOrEmpty();
     }
 
-    public void routeReply(String correlationID, String data) {
-        final UnicastProcessor<String> processor = processors.remove(correlationID);
+    public void routeReply(String correlationID, Message data) {
+        final UnicastProcessor<Message> processor = processors.remove(correlationID);
         if (processor != null) {
             processor.onNext(data);
             processor.onComplete();
         }
     }
 
     public <E> void routeError(String correlationID, String data) {
-        final UnicastProcessor<String> processor = processors.remove(correlationID);
+        final UnicastProcessor<Message> processor = processors.remove(correlationID);
         if (processor != null) {
             processor.onError(new RuntimeException(data));
         }
     }
 
     public void routeEmpty(String correlationID) {
-        final UnicastProcessor<String> processor = processors.remove(correlationID);
+        final UnicastProcessor<Message> processor = processors.remove(correlationID);
         if (processor != null) {
             processor.onComplete();
         }

async/async-commons/src/test/java/org/reactivecommons/async/impl/converters/JacksonMessageConverterTest.java

@@ -53,7 +53,22 @@ public void shouldConvertWithUnknownProperties() throws JsonProcessingException
 
         assertThat(command.getData()).extracting(SampleClass::getId, SampleClass::getName)
             .containsExactly("23", "one");
+    }
 
+    @Test
+    public void readValue() {
+        Date date = new Date();
+        final Message message = converter.toMessage(new SampleClass("35", "name1", date));
+        final SampleClass value = converter.readValue(message, SampleClass.class);
+        assertThat(value).extracting(SampleClass::getId, SampleClass::getName, SampleClass::getDate)
+            .containsExactly("35", "name1", date);
+    }
+
+    @Test
+    public void readValueString() {
+        final Message message = converter.toMessage("Hi!");
+        final String value = converter.readValue(message, String.class);
+        assertThat(value).isEqualTo("Hi!");
     }
 
     @RequiredArgsConstructor