Implementation of @handler annotation replaced with new jerarchy of classes for Handlers. In application properties 'listener' term replaced with 'subscriber'

Author: Pablo Parra

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/Handler.java

@@ -5,7 +5,6 @@
 
 import org.apache.activemq.broker.region.Queue;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 
 /**
@@ -73,27 +72,26 @@ public interface Integration {
    * Subscribes to the default simple channel. The subscriber will be listening to the channel and polling it for new
    * messages in an interval of time configured with the property "integration.one-direction.poller.rate"
    *
-   * @param messageHandler the {@link MessageHandler} in charge of managing each received {@link Message}
+   * @param messageHandler the {@link SubscriptionHandler} in charge of managing each received {@link Message}
    */
-  void subscribe(MessageHandler messageHandler);
+  void subscribe(SubscriptionHandler messageHandler);
 
   /**
    * Subscribes to the default asynchronous request-reply channel. The subscriber will be listening to the channel and
-   * will generate a response with the {@link IntegrationHandler}.
+   * will generate a response with the {@link RequestAsyncHandler}.
    *
-   * @param handler the {@link IntegrationHandler} in charge of managing each received {@link Message} and provide an
+   * @param handler the {@link RequestAsyncHandler} in charge of managing each received {@link Message} and provide an
    *        asynchronous response
    */
-  void subscribeAsync(IntegrationHandler handler);
+  void subscribeAsync(RequestAsyncHandler handler);
 
   /**
    * Subscribes to the default request-reply channel. The subscriber will be listening to the channel and will generate
-   * a response with the {@link IntegrationHandler}.
+   * a response with the {@link RequestHandler}.
    *
-   * @param handler the {@link IntegrationHandler} in charge of managing each received {@link Message} and return a
-   *        response
+   * @param handler the {@link RequestHandler} in charge of managing each received {@link Message} and return a response
    */
-  void subscribeAndReply(IntegrationHandler handler);
+  void subscribeAndReply(RequestHandler handler);
 
   // Implementation for new created Channels
 
@@ -103,42 +101,41 @@ public interface Integration {
    *
    * @param channelName the {@link IntegrationChannel} to be subscribed to
    * @param queueName the {@link Queue} to be listening to
-   * @param messageHandler the {@link MessageHandler} handler to manage each received message
+   * @param messageHandler the {@link SubscriptionHandler} handler to manage each received message
    */
-  void subscribeTo(String channelName, String queueName, MessageHandler messageHandler);
+  void subscribeTo(String channelName, String queueName, SubscriptionHandler messageHandler);
 
   /**
    * Subscribes to a new simple channel, no response is sent. The subscriber will be listening to the channel and
    * polling it for new messages.
    *
    * @param channelName the {@link IntegrationChannel} to be subscribed to
    * @param queueName the {@link Queue} to be listening to
-   * @param messageHandler the {@link MessageHandler} handler to manage each received message
+   * @param messageHandler the {@link SubscriptionHandler} handler to manage each received message
    * @param pollRate the time interval for making the poll to the message broker to check new messages
    */
-  void subscribeTo(String channelName, String queueName, MessageHandler messageHandler, long pollRate);
+  void subscribeTo(String channelName, String queueName, SubscriptionHandler messageHandler, long pollRate);
 
   /**
    * Subscribes to a new request-reply channel. The subscriber will be listening to the channel for new messages and
    * creating a response for each one with the {@link IntegrationHandler}.
    *
    * @param channelName channelName the {@link IntegrationChannel} to be subscribed to
    * @param queueName queueName the {@link Queue} to be listening to
-   * @param handler the {@link IntegrationHandler} in charge of managing each received {@link Message} and return a
-   *        response
+   * @param handler the {@link RequestHandler} in charge of managing each received {@link Message} and return a response
    */
-  void subscribeAndReplyTo(String channelName, String queueName, IntegrationHandler handler);
+  void subscribeAndReplyTo(String channelName, String queueName, RequestHandler handler);
 
   /**
    * Subscribes to a new asynchronous request-reply channel. The subscriber will be listening to the channel for new
    * messages and creating a response for each one with the {@link IntegrationHandler}.
    *
    * @param channelName channelName the {@link IntegrationChannel} to be subscribed to
    * @param queueName queueName the {@link Queue} to be listening to
-   * @param handler the {@link IntegrationHandler} in charge of managing each received {@link Message} and return a
+   * @param handler the {@link RequestAsyncHandler} in charge of managing each received {@link Message} and return a
    *        response
    */
-  void subscribeAndReplyAsyncTo(String channelName, String queueName, IntegrationHandler handler);
+  void subscribeAndReplyAsyncTo(String channelName, String queueName, RequestAsyncHandler handler);
 
   /**
    * Creates a new simple {@link IntegrationChannel}
@@ -155,46 +152,46 @@ public interface Integration {
    *
    * @param channelName name for the new {@link IntegrationChannel}
    * @param queueName name for the new {@link Queue}
-   * @param messageHandler the {@link MessageHandler} handler to manage each received response
+   * @param responseHandler the {@link ResponseHandler} handler to manage each received response
    * @return the new {@link IntegrationChannel}
    */
-  IntegrationChannel createRequestReplyChannel(String channelName, String queueName, MessageHandler messageHandler);
+  IntegrationChannel createRequestReplyChannel(String channelName, String queueName, ResponseHandler responseHandler);
 
   /**
    * Creates a new request-reply {@link IntegrationChannel}.
    *
    * @param channelName name for the new {@link IntegrationChannel}
    * @param queueName name for the new {@link Queue}
-   * @param messageHandler the {@link MessageHandler} handler to manage each received response
+   * @param responseHandler the {@link ResponseHandler} handler to manage each received response
    * @param receiveTimeout the waiting time for a response after sending a {@link Message}
    * @return the new {@link IntegrationChannel}
    */
-  IntegrationChannel createRequestReplyChannel(String channelName, String queueName, MessageHandler messageHandler,
+  IntegrationChannel createRequestReplyChannel(String channelName, String queueName, ResponseHandler responseHandler,
       long receiveTimeout);
 
   /**
    * Creates a new asynchronous request-reply {@link IntegrationChannel}. The core pool size for the asynchronous task
    * executor is configured with the property "integration.default.poolsize".
-   * 
+   *
    * @param channelName name for the new {@link IntegrationChannel}
    * @param queueName name for the new {@link Queue}
-   * @param messageHandler the {@link MessageHandler} handler to manage each received response
+   * @param responseHandler the {@link ResponseHandler} handler to manage each received response
    * @return the new {@link IntegrationChannel}
    */
   IntegrationChannel createAsyncRequestReplyChannel(String channelName, String queueName,
-      MessageHandler messageHandler);
+      ResponseHandler responseHandler);
 
   /**
    * Creates a new asynchronous request-reply {@link IntegrationChannel}.
    *
    * @param channelName name for the new {@link IntegrationChannel}
    * @param queueName name for the new {@link Queue}
-   * @param messageHandler the {@link MessageHandler} handler to manage each received response
+   * @param responseHandler the {@link ResponseHandler} handler to manage each received response
    * @param poolSize ThreadPoolExecutor's core pool size
    * @param receiveTimeout the waiting time for a response after sending a {@link Message}
    * @return the new {@link IntegrationChannel}
    */
-  IntegrationChannel createAsyncRequestReplyChannel(String channelName, String queueName, MessageHandler messageHandler,
-      int poolSize, long receiveTimeout);
+  IntegrationChannel createAsyncRequestReplyChannel(String channelName, String queueName,
+      ResponseHandler responseHandler, int poolSize, long receiveTimeout);
 
 }

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/Integration.java

@@ -0,0 +1,9 @@
+package com.capgemini.devonfw.module.integration.common.api;
+
+/**
+ * @author pparrado
+ *
+ */
+public interface RequestAsyncHandler extends IntegrationHandler {
+
+}

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/RequestAsyncHandler.java

@@ -0,0 +1,9 @@
+package com.capgemini.devonfw.module.integration.common.api;
+
+/**
+ * @author pparrado
+ *
+ */
+public interface RequestHandler extends IntegrationHandler {
+
+}

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/RequestHandler.java

@@ -0,0 +1,11 @@
+package com.capgemini.devonfw.module.integration.common.api;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @author pparrado
+ *
+ */
+public interface ResponseHandler extends MessageHandler {
+
+}

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/ResponseHandler.java

@@ -0,0 +1,11 @@
+package com.capgemini.devonfw.module.integration.common.api;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @author pparrado
+ *
+ */
+public interface SubscriptionHandler extends MessageHandler {
+
+}

modules/integration/src/main/java/com/capgemini/devonfw/module/integration/common/api/SubscriptionHandler.java

@@ -25,6 +25,9 @@
 import org.springframework.messaging.support.GenericMessage;
 
 import com.capgemini.devonfw.module.integration.common.api.IntegrationHandler;
+import com.capgemini.devonfw.module.integration.common.api.RequestAsyncHandler;
+import com.capgemini.devonfw.module.integration.common.api.RequestHandler;
+import com.capgemini.devonfw.module.integration.common.api.SubscriptionHandler;
 
 /**
  * @author pparrado
@@ -167,63 +170,63 @@ public IntegrationFlow asyncOutboundFlow() {
   // in
 
   /**
-   * If the property "integration.one-direction.listener" has value "true" this Bean will be loaded and will create a
+   * If the property "integration.one-direction.subscriber" has value "true" this Bean will be loaded and will create a
    * simple flow for receiving messages.
    *
    * @param handler the {@link MessageHandler} that will manage each message received.
    * @return the created {@link IntegrationFlow}
    */
   @Bean
-  @ConditionalOnProperty(prefix = "devonfw.integration.one-direction", name = "listener", havingValue = "true")
-  public IntegrationFlow inFlow(MessageHandler handler) {
+  @ConditionalOnProperty(prefix = "devonfw.integration.one-direction", name = "subscriber", havingValue = "true")
+  public IntegrationFlow inFlow(SubscriptionHandler handler) {
 
     return IntegrationFlows.from(Jms.inboundAdapter(this.connectionFactory).destination(this.queue_1d),
         c -> c.poller(Pollers.fixedRate(this.rate, TimeUnit.MILLISECONDS))).handle(m -> {
           try {
             handler.handleMessage(m);
           } catch (Exception e) {
-            LOG.error(String.format("MessageHandler threw an error: %s", e.getMessage()), e);
+            LOG.error(String.format("SubscriptionHandler threw an error: %s", e.getMessage()), e);
           }
         }).get();
   }
 
   /**
-   * If the property "integration.request-reply.listener" has value "true" this Bean will be loaded and will create a
+   * If the property "integration.request-reply.subscriber" has value "true" this Bean will be loaded and will create a
    * request-reply flow for receiving and replying to messages.
    *
    * @param handler the {@link IntegrationHandler} to manage the messages and send back the response
    * @return the created {@link IntegrationFlow}
    */
   @Bean
-  @ConditionalOnProperty(prefix = "devonfw.integration.request-reply", name = "listener", havingValue = "true")
-  public IntegrationFlow inAndOutFlow(IntegrationHandler handler) {
+  @ConditionalOnProperty(prefix = "devonfw.integration.request-reply", name = "subscriber", havingValue = "true")
+  public IntegrationFlow inAndOutFlow(RequestHandler handler) {
 
     return IntegrationFlows.from(Jms.inboundGateway(this.connectionFactory).destination(this.queue_rr))
-        .wireTap(flow -> flow.handle(System.out::println)).handle(new GenericHandler<String>() {
+        .wireTap(flow -> flow.handle(m -> LOG.info(m.getPayload().toString()))).handle(new GenericHandler<String>() {
 
           @Override
           public Object handle(String payload, Map<String, Object> headers) {
 
             try {
               return handler.handleMessage(new GenericMessage<>(payload, headers));
             } catch (Exception e) {
-              LOG.error(String.format("IntegrationHandler threw an error: %s", e.getMessage()), e);
+              LOG.error(String.format("RequestHandler threw an error: %s", e.getMessage()), e);
               return null;
             }
           }
         }).get();
   }
 
   /**
-   * If the property "integration.request-reply-async.listener" has value "true" this Bean will be loaded and will
+   * If the property "integration.request-reply-async.subscriber" has value "true" this Bean will be loaded and will
    * create a request-reply asynchronous flow for sending messages and receiving responses.
    *
    * @param handler the {@link IntegrationHandler} to manage the messages and send back the response
    * @return the created {@link IntegrationFlow}
    */
   @Bean
-  @ConditionalOnProperty(prefix = "devonfw.integration.request-reply-async", name = "listener", havingValue = "true")
-  public IntegrationFlow asyncInAndOutFlow(IntegrationHandler handler) {
+  @ConditionalOnProperty(prefix = "devonfw.integration.request-reply-async", name = "subscriber", havingValue = "true")
+  public IntegrationFlow asyncInAndOutFlow(RequestAsyncHandler handler) {
 
     return IntegrationFlows.from(Jms.inboundGateway(this.connectionFactory).destination(this.queue_async))
         .wireTap(flow -> flow.handle(System.out::println)).handle(new GenericHandler<String>() {
@@ -233,7 +236,7 @@ public Object handle(String payload, Map<String, Object> headers) {
             try {
               return handler.handleMessage(new GenericMessage<>(payload, headers));
             } catch (Exception e) {
-              LOG.error(String.format("IntegrationHandler threw an error: %s", e.getMessage()), e);
+              LOG.error(String.format("RequestAsyncHandler threw an error: %s", e.getMessage()), e);
               return null;
             }
           }