opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #4841] Fix the reply message is the original request msg. (#4896)

Browse Source
This commit is contained in:
echooymxq
2022-08-27 11:00:20 +08:00
committed by GitHub
parent 1cc7afd6af
commit f7c2072a27
2 changed files with 13 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
-1
View File
@@ -1374,7 +1374,6 @@ public class DefaultMQProducerImpl implements MQProducerInner {
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
requestResponseFuture.putResponseMessage(msg);
}
@Override
client/src/test/java/org/apache/rocketmq/client/producer/DefaultMQProducerTest.java
+13 -3
View File
@@ -42,6 +42,7 @@ import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.route.BrokerData;
@@ -377,15 +378,19 @@ public class DefaultMQProducerTest {
Thread.sleep(10);
} catch (InterruptedException e) {
}
MessageExt responseMsg = new MessageExt();
responseMsg.setTopic(message.getTopic());
responseMsg.setBody(message.getBody());
for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
future.putResponseMessage(message);
future.putResponseMessage(responseMsg);
}
}
}
}).start();
Message result = producer.request(message, 3 * 1000L);
finish.getAndSet(true);
assertThat(result).isExactlyInstanceOf(MessageExt.class);
assertThat(result.getTopic()).isEqualTo("FooBar");
assertThat(result.getBody()).isEqualTo(new byte[] {'a'});
}
@@ -403,6 +408,7 @@ public class DefaultMQProducerTest {
RequestCallback requestCallback = new RequestCallback() {
@Override
public void onSuccess(Message message) {
assertThat(message).isExactlyInstanceOf(MessageExt.class);
assertThat(message.getTopic()).isEqualTo("FooBar");
assertThat(message.getBody()).isEqualTo(new byte[] {'a'});
assertThat(message.getFlag()).isEqualTo(1);
@@ -416,11 +422,15 @@ public class DefaultMQProducerTest {
producer.request(message, requestCallback, 3 * 1000L);
ConcurrentHashMap<String, RequestResponseFuture> responseMap = RequestFutureHolder.getInstance().getRequestFutureTable();
assertThat(responseMap).isNotNull();
MessageExt responseMsg = new MessageExt();
responseMsg.setTopic(message.getTopic());
responseMsg.setBody(message.getBody());
responseMsg.setFlag(1);
for (Map.Entry<String, RequestResponseFuture> entry : responseMap.entrySet()) {
RequestResponseFuture future = entry.getValue();
future.setSendRequestOk(true);
message.setFlag(1);
future.getRequestCallback().onSuccess(message);
future.getRequestCallback().onSuccess(responseMsg);
}
countDownLatch.await(3000L, TimeUnit.MILLISECONDS);
}