[ISSUE #377] Add the replyTimeout configuration parameter (#384)

* [ISSUE #377]Add the replyTimeout configuration parameter

* add unit test

Co-authored-by: zhangjidi2016 <zhangjidi@cmss.chinamobile.com>

Author: zhangjidi2016

rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumerWithReplyString.java

@@ -25,7 +25,8 @@
  * The consumer that replying String
  */
 @Service
-@RocketMQMessageListener(topic = "${demo.rocketmq.stringRequestTopic}", consumerGroup = "${demo.rocketmq.stringRequestConsumer}", selectorExpression = "${demo.rocketmq.tag}")
+@RocketMQMessageListener(topic = "${demo.rocketmq.stringRequestTopic}", consumerGroup = "${demo.rocketmq.stringRequestConsumer}",
+    selectorExpression = "${demo.rocketmq.tag}", replyTimeout = 10000)
 public class StringConsumerWithReplyString implements RocketMQReplyListener<String, String> {
 
     @Override

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java

@@ -88,6 +88,11 @@
      */
     long consumeTimeout() default 15L;
 
+    /**
+     * Timeout for sending reply messages.
+     */
+    int replyTimeout() default 3000;
+
     /**
      * The property of "access-key".
      */

rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java

@@ -34,6 +34,7 @@
 import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
 import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
 import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
@@ -121,6 +122,7 @@ public class DefaultRocketMQListenerContainer implements InitializingBean,
     private MessageModel messageModel;
     private long consumeTimeout;
     private int maxReconsumeTimes;
+    private int replyTimeout;
 
     public long getSuspendCurrentQueueTimeMillis() {
         return suspendCurrentQueueTimeMillis;
@@ -221,6 +223,7 @@ public void setRocketMQMessageListener(RocketMQMessageListener anno) {
         this.selectorExpression = anno.selectorExpression();
         this.consumeTimeout = anno.consumeTimeout();
         this.maxReconsumeTimes = anno.maxReconsumeTimes();
+        this.replyTimeout = anno.replyTimeout();
     }
 
     public ConsumeMode getConsumeMode() {
@@ -399,7 +402,9 @@ private void handleMessage(
             Message<?> message = MessageBuilder.withPayload(replyContent).build();
 
             org.apache.rocketmq.common.message.Message replyMessage = MessageUtil.createReplyMessage(messageExt, convertToBytes(message));
-            consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(replyMessage, new SendCallback() {
+            DefaultMQProducer producer = consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer();
+            producer.setSendMsgTimeout(replyTimeout);
+            producer.send(replyMessage, new SendCallback() {
                 @Override public void onSuccess(SendResult sendResult) {
                     if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                         log.error("Consumer replies message failed. SendStatus: {}", sendResult.getSendStatus());

rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainerTest.java

@@ -20,9 +20,15 @@
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendCallback;
 import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.apache.rocketmq.spring.core.RocketMQReplyListener;
 import org.junit.Test;
@@ -34,9 +40,12 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Date;
-import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class DefaultRocketMQListenerContainerTest {
     @Test
@@ -184,6 +193,45 @@ public String onMessage(ArrayList<Date> message) {
         assertThat(methodParameter.getParameterType() == ArrayList.class);
     }
 
+    @Test
+    public void testHandleMessage() throws Exception {
+        DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer();
+        Method handleMessage = DefaultRocketMQListenerContainer.class.getDeclaredMethod("handleMessage", MessageExt.class);
+        handleMessage.setAccessible(true);
+        listenerContainer.setRocketMQListener(new RocketMQListener<String>() {
+            @Override
+            public void onMessage(String message) {
+            }
+        });
+        Field messageType = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
+        messageType.setAccessible(true);
+        messageType.set(listenerContainer, String.class);
+        MessageExt messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null);
+        MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "defaultCluster");
+        messageExt.setBody("hello".getBytes());
+        handleMessage.invoke(listenerContainer, messageExt);
+
+        // reply message
+        listenerContainer.setRocketMQListener(null);
+        DefaultMQPushConsumer consumer = mock(DefaultMQPushConsumer.class);
+        DefaultMQPushConsumerImpl pushConsumer = mock(DefaultMQPushConsumerImpl.class);
+        MQClientInstance mqClientInstance = mock(MQClientInstance.class);
+        DefaultMQProducer producer = mock(DefaultMQProducer.class);
+        when(consumer.getDefaultMQPushConsumerImpl()).thenReturn(pushConsumer);
+        when(pushConsumer.getmQClientFactory()).thenReturn(mqClientInstance);
+        when(mqClientInstance.getDefaultMQProducer()).thenReturn(producer);
+        listenerContainer.setConsumer(consumer);
+        listenerContainer.setMessageConverter(new CompositeMessageConverter(Arrays.asList(new StringMessageConverter(), new MappingJackson2MessageConverter())));
+        doNothing().when(producer).send(any(MessageExt.class), any(SendCallback.class));
+        listenerContainer.setRocketMQReplyListener(new RocketMQReplyListener<String, String>() {
+            @Override
+            public String onMessage(String message) {
+                return "test";
+            }
+        });
+        handleMessage.invoke(listenerContainer, messageExt);
+    }
+
     class User {
         private String userName;
         private int userAge;