You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2021/05/08 05:44:34 UTC
[GitHub] [rocketmq-spring] weblqb opened a new issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
weblqb opened a new issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364
Hi,
I am using the version of 2.2.
I found out that the function `convertToRocketMessage` in class `org.apache.rocketmq.spring.support.RocketMQUtil` did not implement converting header of "DELAY" to `rocketMsg.setDelayTimeLevel()`. This bug makes spring-cloud-starter-stream-rocketmq cannot send delayed message when producer working in asynchronous mode.
I hope you can do the change like this in the futher version:
```java
public static Message convertToRocketMessage(ObjectMapper objectMapper, String charset, String destination, org.springframework.messaging.Message<?> message) {
Object payloadObj = message.getPayload();
byte[] payloads;
if (payloadObj instanceof String) {
payloads = ((String)payloadObj).getBytes(Charset.forName(charset));
} else if (payloadObj instanceof byte[]) {
payloads = (byte[])((byte[])message.getPayload());
} else {
try {
String jsonObj = objectMapper.writeValueAsString(payloadObj);
payloads = jsonObj.getBytes(Charset.forName(charset));
} catch (Exception var17) {
throw new RuntimeException("convert to RocketMQ message failed.", var17);
}
}
String[] tempArr = destination.split(":", 2);
String topic = tempArr[0];
String tags = "";
if (tempArr.length > 1) {
tags = tempArr[1];
}
Message rocketMsg = new Message(topic, tags, payloads);
MessageHeaders headers = message.getHeaders();
if (Objects.nonNull(headers) && !headers.isEmpty()) {
// hope you can add this part 👍
Object delayLevelObj = headers.getOrDefault("DELAY", 0);
if (delayLevelObj instanceof Number) {
delayLevel = ((Number)delayLevelObj).intValue();
} else if (delayLevelObj instanceof String) {
delayLevel = Integer.parseInt((String)delayLevelObj);
}
rocketMsg.setDelayTimeLevel(delayLevel );
// thanks
Object keys = headers.get("KEYS");
if (!StringUtils.isEmpty(keys)) {
rocketMsg.setKeys(keys.toString());
}
Object flagObj = headers.getOrDefault("FLAG", "0");
int flag = 0;
try {
flag = Integer.parseInt(flagObj.toString());
} catch (NumberFormatException var16) {
log.info("flag must be integer, flagObj:{}", flagObj);
}
rocketMsg.setFlag(flag);
Object waitStoreMsgOkObj = headers.getOrDefault("WAIT_STORE_MSG_OK", "true");
boolean waitStoreMsgOK = Boolean.TRUE.equals(waitStoreMsgOkObj);
rocketMsg.setWaitStoreMsgOK(waitStoreMsgOK);
headers.entrySet().stream().filter((entry) -> {
return !Objects.equals(entry.getKey(), "FLAG") && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK");
}).forEach((entry) -> {
if (!MessageConst.STRING_HASH_SET.contains(entry.getKey())) {
rocketMsg.putUserProperty((String)entry.getKey(), String.valueOf(entry.getValue()));
}
});
}
return rocketMsg;
}
```
Good day!
Billy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] weblqb commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
weblqb commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835254144
> Hi Billy, RocketMQTemplate has an asyncSend(RocketMQTemplate#asyncSend(java.lang.String, org.springframework.messaging.Message<?>, org.apache.rocketmq.client.producer.SendCallback, long, int)) API to send delay messages asynchronously.
And btw, I found that the source code in github dose have it. But the maven package (v2.2) dose not, which is weird.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] weblqb closed issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
weblqb closed issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] RongtongJin commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835173921
Hi Billy, RocketMQTemplate has an asyncSend(RocketMQTemplate#asyncSend(java.lang.String, org.springframework.messaging.Message<?>, org.apache.rocketmq.client.producer.SendCallback, long, int)) API to send delay messages asynchronously.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] vongosling commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
vongosling commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-836033276
@fangjian0423 Would you like to move the spring-cloud-starter-stream-rocketmq to rocketmq spring community? I have noticed many guys are asking for help in this project. We could do more support in this all-in-all project with spring.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] weblqb removed a comment on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
weblqb removed a comment on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835254144
> Hi Billy, RocketMQTemplate has an asyncSend(RocketMQTemplate#asyncSend(java.lang.String, org.springframework.messaging.Message<?>, org.apache.rocketmq.client.producer.SendCallback, long, int)) API to send delay messages asynchronously.
And btw, I found that the source code in github dose have it. But the maven package (v2.2) dose not, which is weird.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] weblqb commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
weblqb commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835256340
> Here
>
> https://github.com/apache/rocketmq-spring/blob/66a6d8b51c6202b14195b1ac1a4311a7ce3e81dc/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java#L668-L684
Thanks! spring-cloud-starter-stream-rocketmq depends on an older version 2.0.2, which is the reason why the problem appears.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] weblqb commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
weblqb commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835250983
> Hi Billy, RocketMQTemplate has an asyncSend(RocketMQTemplate#asyncSend(java.lang.String, org.springframework.messaging.Message<?>, org.apache.rocketmq.client.producer.SendCallback, long, int)) API to send delay messages asynchronously.
But actually, there is not in v2.2 RocketMQTemplate. Here are all the asyncSend implementations:
```java
public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout) {
if (!Objects.isNull(message) && !Objects.isNull(message.getPayload())) {
try {
org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(this.objectMapper, this.charset, destination, message);
this.producer.send(rocketMsg, sendCallback, timeout);
} catch (Exception var7) {
log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(var7.getMessage(), var7);
}
} else {
log.error("asyncSend failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
}
public void asyncSend(String destination, Message<?> message, SendCallback sendCallback) {
this.asyncSend(destination, message, sendCallback, (long)this.producer.getSendMsgTimeout());
}
public void asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout) {
Message<?> message = this.doConvert(payload, (Map)null, (MessagePostProcessor)null);
this.asyncSend(destination, message, sendCallback, timeout);
}
public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
this.asyncSend(destination, payload, sendCallback, (long)this.producer.getSendMsgTimeout());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-spring] RongtongJin commented on issue #364: No Implementation for DelayTimeLevel in org.apache.rocketmq.spring.support.RocketMQUtil.convertToRocketMessage
Posted by GitBox <gi...@apache.org>.
RongtongJin commented on issue #364:
URL: https://github.com/apache/rocketmq-spring/issues/364#issuecomment-835254114
Here https://github.com/apache/rocketmq-spring/blob/66a6d8b51c6202b14195b1ac1a4311a7ce3e81dc/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java#L668-L684
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org