You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "YxAc (via GitHub)" <gi...@apache.org> on 2024/05/21 13:00:07 UTC
[PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
YxAc opened a new pull request, #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279
### Which Issue(s) This PR Fixes
Close #246
### Brief Description
implement flow control for MQTT 5.0
### How Did You Test This Change?
UT
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "YxAc (via GitHub)" <gi...@apache.org>.
YxAc commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1630789596
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/ChannelManager.java:
##########
@@ -63,4 +63,35 @@ public interface ChannelManager {
*/
int totalConn();
+ /**
+ * Refills the publish receive quota for the given channel.
+ * This method is typically called after a publish acknowledgement or other scenarios that refill the quota.
+ * @param channel The channel for which the quota needs to be refilled.
+ * @return true if the quota was successfully refilled, false otherwise.
+ */
+ boolean publishReceiveRefill(Channel channel);
Review Comment:
OK，i will put them to Session.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "YxAc (via GitHub)" <gi...@apache.org>.
YxAc commented on PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#issuecomment-2154840175
hi @DongyuanPan , i have updated this PR according to the review comments, pls take a look when you are free. Thx.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "YxAc (via GitHub)" <gi...@apache.org>.
YxAc commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1630793092
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/infly/PushAction.java:
##########
@@ -181,6 +185,12 @@ public void write(Session session, Message message, int mqttId, int qos, Subscri
data = MqttMessageFactory.buildPublishMessage(topicName, message.getPayload(), qos, retained, mqttId);
break;
case MQTT_5:
+ // Server send PublishMessage flow control
+ if (qos > 0 && !channelManager.publishSendTryAcquire(channel)) {
+ logger.error("publishSendTryAcquire failed, client:{}, message:{}", clientId, message);
+ return;
+ }
Review Comment:
Yes, the retry logic will also step into the `write` function and require the quota by flow control.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "ChangingFond (via GitHub)" <gi...@apache.org>.
ChangingFond commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1626836891
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java:
##########
@@ -61,6 +61,7 @@ public class ConnectConf {
private boolean enableRetain = false;
private boolean enableSubscriptionIdentifier = false;
private int topicAliasMaximum = 10;
+ private int serverReceiveMaximum = 65535;
Review Comment:
why set to 65535？
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/ChannelManager.java:
##########
@@ -63,4 +63,35 @@ public interface ChannelManager {
*/
int totalConn();
+ /**
+ * Refills the publish receive quota for the given channel.
+ * This method is typically called after a publish acknowledgement or other scenarios that refill the quota.
+ * @param channel The channel for which the quota needs to be refilled.
+ * @return true if the quota was successfully refilled, false otherwise.
+ */
+ boolean publishReceiveRefill(Channel channel);
Review Comment:
extract these flow releated method into another class？
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "DongyuanPan (via GitHub)" <gi...@apache.org>.
DongyuanPan commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1628718513
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java:
##########
@@ -177,4 +188,71 @@ public int totalConn() {
return channelMap.size();
}
+ public boolean publishReceiveRefill(Channel channel) {
Review Comment:
These processing functions can be placed in the Session
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/session/infly/PushAction.java:
##########
@@ -181,6 +185,12 @@ public void write(Session session, Message message, int mqttId, int qos, Subscri
data = MqttMessageFactory.buildPublishMessage(topicName, message.getPayload(), qos, retained, mqttId);
break;
case MQTT_5:
+ // Server send PublishMessage flow control
+ if (qos > 0 && !channelManager.publishSendTryAcquire(channel)) {
+ logger.error("publishSendTryAcquire failed, client:{}, message:{}", clientId, message);
+ return;
+ }
Review Comment:
If the push fails, there will be a retry, and the retry also requires flow control.
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java:
##########
@@ -47,6 +48,12 @@ public class DefaultChannelManager implements ChannelManager {
private HashedWheelTimer hashedWheelTimer;
private static int minBlankChannelSeconds = 10;
private ScheduledThreadPoolExecutor scheduler;
+ // using to control publish-client's PUBLISH rate by MQTT-Server Receive Maximum
Review Comment:
Quota is related to the session and can be placed directly in the session without using map.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "YxAc (via GitHub)" <gi...@apache.org>.
YxAc commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1630791168
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java:
##########
@@ -47,6 +48,12 @@ public class DefaultChannelManager implements ChannelManager {
private HashedWheelTimer hashedWheelTimer;
private static int minBlankChannelSeconds = 10;
private ScheduledThreadPoolExecutor scheduler;
+ // using to control publish-client's PUBLISH rate by MQTT-Server Receive Maximum
Review Comment:
good catch, thx.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #246] feat: implement flow control [rocketmq-mqtt]
Posted by "YxAc (via GitHub)" <gi...@apache.org>.
YxAc commented on code in PR #279:
URL: https://github.com/apache/rocketmq-mqtt/pull/279#discussion_r1630790862
##########
mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/config/ConnectConf.java:
##########
@@ -61,6 +61,7 @@ public class ConnectConf {
private boolean enableRetain = false;
private boolean enableSubscriptionIdentifier = false;
private int topicAliasMaximum = 10;
+ private int serverReceiveMaximum = 65535;
Review Comment:
According to MQTT protocol, the initial Receive Maximum is 65535.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org