You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bifromq.apache.org by po...@apache.org on 2026/04/26 08:16:37 UTC
(bifromq) branch main updated: fix shared sub setting check. (#246)
This is an automated email from the ASF dual-hosted git repository.
popduke pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/bifromq.git
The following commit(s) were added to refs/heads/main by this push:
new f5dfaa1aa fix shared sub setting check. (#246)
f5dfaa1aa is described below
commit f5dfaa1aa0874283e4eb62501782d5a1929bb8b9
Author: Gu Jiawei <gu...@foxmail.com>
AuthorDate: Sun Apr 26 01:16:32 2026 -0700
fix shared sub setting check. (#246)
---
.../bifromq/mqtt/handler/MQTTSessionHandler.java | 2 +-
.../bifromq/mqtt/handler/v3/MQTTSubTest.java | 33 ++++++++++++++++++++++
2 files changed, 34 insertions(+), 1 deletion(-)
diff --git a/bifromq-mqtt/bifromq-mqtt-server/src/main/java/org/apache/bifromq/mqtt/handler/MQTTSessionHandler.java b/bifromq-mqtt/bifromq-mqtt-server/src/main/java/org/apache/bifromq/mqtt/handler/MQTTSessionHandler.java
index d15aee9cd..d118bd863 100644
--- a/bifromq-mqtt/bifromq-mqtt-server/src/main/java/org/apache/bifromq/mqtt/handler/MQTTSessionHandler.java
+++ b/bifromq-mqtt/bifromq-mqtt-server/src/main/java/org/apache/bifromq/mqtt/handler/MQTTSessionHandler.java
@@ -592,7 +592,7 @@ public abstract class MQTTSessionHandler extends MQTTMessageHandler implements I
return CompletableFuture.completedFuture(IMQTTProtocolHelper.SubResult.WILDCARD_NOT_SUPPORTED);
}
- if (isSharedSubscription(topicFilter) && !settings.subscriptionIdentifierEnabled) {
+ if (isSharedSubscription(topicFilter) && !settings.sharedSubscriptionEnabled) {
return CompletableFuture.completedFuture(IMQTTProtocolHelper.SubResult.SHARED_SUBSCRIPTION_NOT_SUPPORTED);
}
diff --git a/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/handler/v3/MQTTSubTest.java b/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/handler/v3/MQTTSubTest.java
index 5cd405f47..9cf809cbb 100644
--- a/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/handler/v3/MQTTSubTest.java
+++ b/bifromq-mqtt/bifromq-mqtt-server/src/test/java/org/apache/bifromq/mqtt/handler/v3/MQTTSubTest.java
@@ -29,10 +29,13 @@ import static org.apache.bifromq.plugin.eventcollector.EventType.RETAIN_MSG_MATC
import static org.apache.bifromq.plugin.eventcollector.EventType.SUB_ACKED;
import static org.apache.bifromq.plugin.eventcollector.EventType.SUB_ACTION_DISALLOW;
import static org.apache.bifromq.plugin.eventcollector.EventType.TOO_LARGE_SUBSCRIPTION;
+import static org.apache.bifromq.plugin.settingprovider.Setting.SharedSubscriptionEnabled;
+import static org.apache.bifromq.plugin.settingprovider.Setting.SubscriptionIdentifierEnabled;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -334,6 +337,36 @@ public class MQTTSubTest extends BaseMQTTTest {
shouldCleanSubs = true;
}
+ @Test
+ public void sharedSubRejectedWhenSharedSubscriptionDisabled() {
+ when(settingProvider.provide(eq(SharedSubscriptionEnabled), anyString())).thenReturn(false);
+ when(settingProvider.provide(eq(SubscriptionIdentifierEnabled), anyString())).thenReturn(true);
+ setupTransientSession();
+
+ mockAuthCheck(true);
+ MqttSubscribeMessage subMessage = MQTTMessageUtils.topicMqttSubMessages("$share/g/t1");
+ channel.writeInbound(subMessage);
+ MqttSubAckMessage subAckMessage = channel.readOutbound();
+ verifySubAck(subAckMessage, new int[] {128});
+ verifyEvent(MQTT_SESSION_START, CLIENT_CONNECTED, SUB_ACKED);
+ }
+
+ @Test
+ public void sharedSubAcceptedWhenOnlySubscriptionIdentifierDisabled() {
+ when(settingProvider.provide(eq(SharedSubscriptionEnabled), anyString())).thenReturn(true);
+ when(settingProvider.provide(eq(SubscriptionIdentifierEnabled), anyString())).thenReturn(false);
+ setupTransientSession();
+
+ mockAuthCheck(true);
+ mockDistMatch(QoS.AT_MOST_ONCE, true);
+ MqttSubscribeMessage subMessage = MQTTMessageUtils.topicMqttSubMessages("$share/g/t1");
+ channel.writeInbound(subMessage);
+ MqttSubAckMessage subAckMessage = channel.readOutbound();
+ verifySubAck(subAckMessage, new int[] {0});
+ verifyEvent(MQTT_SESSION_START, CLIENT_CONNECTED, SUB_ACKED);
+ shouldCleanSubs = true;
+ }
+
private void verifySubAck(MqttSubAckMessage subAckMessage, int[] expectedQos) {
assertEquals(subAckMessage.payload().grantedQoSLevels().size(), expectedQos.length);
for (int i = 0; i < expectedQos.length; i++) {