fix: grant requested QoS in SUBACK instead of hard-coded QoS 0

shenyu-protocol/shenyu-protocol-mqtt/src/main/java/org/apache/shenyu/protocol/mqtt/Subscribe.java

@@ -21,7 +21,6 @@
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.mqtt.MqttFixedHeader;
-import io.netty.handler.codec.mqtt.MqttQoS;
 import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import io.netty.handler.codec.mqtt.MqttMessageType;
@@ -35,7 +34,6 @@
 import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
 import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
 
-import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -61,9 +59,12 @@ public void subscribe(final ChannelHandlerContext ctx, final MqttSubscribeMessag
         int packetId = msg.variableHeader().messageId();
 
         //// todo Regular match
-        List<String> ackTopics = mqttTopicSubscriptions
+        List<MqttTopicSubscription> ackSubscriptions = mqttTopicSubscriptions
                 .stream()
                 .filter(topicSub -> topicSub.qualityOfService() != FAILURE)
+                .collect(Collectors.toList());
+
+        List<String> ackTopics = ackSubscriptions.stream()
                 .map(MqttTopicSubscription::topicName)
                 .collect(Collectors.toList());
 
@@ -76,22 +77,20 @@ public void subscribe(final ChannelHandlerContext ctx, final MqttSubscribeMessag
             }
         }
 
-        sendSubAckMessage(packetId, ackTopics, channel);
+        sendSubAckMessage(packetId, ackSubscriptions, channel);
     }
 
     /**
      * call back request of message.
      * @param packetId packetId
-     * @param ackTopics ackTopics
+     * @param ackSubscriptions ackSubscriptions
      * @param channel channel
      */
-    private void sendSubAckMessage(final int packetId, final List<String> ackTopics, final Channel channel) {
+    private void sendSubAckMessage(final int packetId, final List<MqttTopicSubscription> ackSubscriptions, final Channel channel) {
 
-        List<Integer> qos = new ArrayList<>();
-        for (int i = 0; i < ackTopics.size(); i++) {
-            // default qos 0
-            qos.add(MqttQoS.AT_MOST_ONCE.value());
-        }
+        List<Integer> qos = ackSubscriptions.stream()
+                .map(topicSub -> topicSub.qualityOfService().value())
+                .collect(Collectors.toList());
 
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, AT_MOST_ONCE,
                 false, 0);

shenyu-protocol/shenyu-protocol-mqtt/src/test/java/org/apache/shenyu/protocol/mqtt/SubscribeTest.java

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.protocol.mqtt;
+
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.codec.mqtt.MqttSubAckMessage;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import io.netty.handler.codec.mqtt.MqttSubscribePayload;
+import io.netty.handler.codec.mqtt.MqttTopicSubscription;
+import org.apache.shenyu.common.utils.Singleton;
+import org.apache.shenyu.protocol.mqtt.repositories.SubscribeRepository;
+import org.apache.shenyu.protocol.mqtt.repositories.TopicRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test case for {@link Subscribe}.
+ */
+@ExtendWith(MockitoExtension.class)
+class SubscribeTest {
+
+    @Mock
+    private SubscribeRepository subscribeRepository;
+
+    @Mock
+    private TopicRepository topicRepository;
+
+    private final Subscribe subscribe = new Subscribe();
+
+    @BeforeEach
+    void setUp() {
+        Singleton.INST.single(SubscribeRepository.class, subscribeRepository);
+        Singleton.INST.single(TopicRepository.class, topicRepository);
+    }
+
+    @Test
+    void testSubAckGrantsRequestedQoS() {
+        EmbeddedChannel channel = connectedChannel();
+        MqttSubscribeMessage msg = subscribeMessage(10,
+                new MqttTopicSubscription("topic/qos2", MqttQoS.EXACTLY_ONCE),
+                new MqttTopicSubscription("topic/qos0", MqttQoS.AT_MOST_ONCE),
+                new MqttTopicSubscription("topic/qos1", MqttQoS.AT_LEAST_ONCE));
+
+        subscribe.subscribe(channel.pipeline().lastContext(), msg);
+
+        MqttSubAckMessage subAck = channel.readOutbound();
+        assertEquals(10, subAck.variableHeader().messageId());
+        assertEquals(Arrays.asList(2, 0, 1), subAck.payload().grantedQoSLevels());
+    }
+
+    @Test
+    void testSubAckExcludesFailureSubscriptions() {
+        EmbeddedChannel channel = connectedChannel();
+        MqttSubscribeMessage msg = subscribeMessage(20,
+                new MqttTopicSubscription("topic/qos1", MqttQoS.AT_LEAST_ONCE),
+                new MqttTopicSubscription("topic/failure", MqttQoS.FAILURE),
+                new MqttTopicSubscription("topic/qos0", MqttQoS.AT_MOST_ONCE));
+
+        subscribe.subscribe(channel.pipeline().lastContext(), msg);
+
+        MqttSubAckMessage subAck = channel.readOutbound();
+        assertEquals(Arrays.asList(1, 0), subAck.payload().grantedQoSLevels());
+        verify(topicRepository, never()).get("topic/failure");
+    }
+
+    @Test
+    void testSubscribeBeforeConnectClosesChannel() {
+        EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter());
+        MqttSubscribeMessage msg = subscribeMessage(30,
+                new MqttTopicSubscription("topic/qos0", MqttQoS.AT_MOST_ONCE));
+
+        subscribe.subscribe(channel.pipeline().lastContext(), msg);
+
+        channel.runPendingTasks();
+        assertFalse(channel.isActive());
+        assertNull(channel.readOutbound());
+    }
+
+    private EmbeddedChannel connectedChannel() {
+        EmbeddedChannel channel = new EmbeddedChannel(new ChannelInboundHandlerAdapter());
+        new MessageType().setConnected(channel, true);
+        return channel;
+    }
+
+    private MqttSubscribeMessage subscribeMessage(final int packetId, final MqttTopicSubscription... subscriptions) {
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
+        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(packetId);
+        return new MqttSubscribeMessage(fixedHeader, variableHeader, new MqttSubscribePayload(Arrays.asList(subscriptions)));
+    }
+}