[ISSUE #22] Improving codeCov for mqtt.common

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Message.java

@@ -189,10 +189,7 @@ public void clearUserProperty(String key) {
     }
 
     public boolean isP2P() {
-        if (TopicUtils.isP2P(TopicUtils.decode(firstTopic).getSecondTopic())) {
-            return true;
-        }
-        return false;
+        return TopicUtils.isP2P(TopicUtils.decode(firstTopic).getSecondTopic());
     }
 
     @Override

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/model/Subscription.java

@@ -36,11 +36,6 @@ public Subscription(String topicFilter, int qos) {
         this.qos = qos;
     }
 
-    public boolean isWildCard() {
-        return topicFilter != null &&
-                (topicFilter.contains(Constants.NUMBER_SIGN) || topicFilter.contains(Constants.PLUS_SIGN));
-    }
-
     public String toFirstTopic() {
         return TopicUtils.decode(topicFilter).getFirstTopic();
     }

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/NamespaceUtil.java

@@ -24,45 +24,35 @@ public class NamespaceUtil {
     private static final int RESOURCE_LENGTH = 2;
     public static final String MQ_DEFAULT_NAMESPACE_NAME = "DEFAULT_INSTANCE";
 
-    public NamespaceUtil() {
-    }
-
     public static String encodeToNamespaceResource(String namespace, String resource) {
-        return resource != null && namespace != null ? StringUtils.join(new String[]{namespace, "%", resource}) : resource;
+        return resource != null && namespace != null ? StringUtils.join(namespace, NAMESPACE_SPLITER, resource) : resource;
     }
 
     public static String decodeOriginResource(String resource) {
-        if (resource != null && resource.contains("%")) {
-            int firstIndex = resource.indexOf("%");
+        if (resource != null && resource.contains(NAMESPACE_SPLITER)) {
+            int firstIndex = resource.indexOf(NAMESPACE_SPLITER);
             return resource.substring(firstIndex + 1);
-        } else {
-            return resource;
         }
+        return resource;
     }
 
     public static String decodeMqttNamespaceIdFromKey(String key) {
         return decodeMqttNamespaceIdFromClientId(key);
     }
 
     public static String decodeMqttNamespaceIdFromClientId(String clientId) {
-        if (clientId != null && clientId.contains("%")) {
-            String mqttNamespaceId = clientId.split("%")[0];
-            return mqttNamespaceId;
-        } else {
-            return null;
-        }
+        return splitNamespaceStr(clientId);
     }
 
     public static String decodeStoreNamespaceIdFromTopic(String topic) {
-        if (topic != null && topic.contains("%")) {
-            String storeNamespaceId = topic.split("%")[0];
-            return storeNamespaceId;
-        } else {
-            return null;
-        }
+        return splitNamespaceStr(topic);
     }
 
     public static String decodeNamespaceId(String resource) {
-        return resource != null && resource.contains("%") ? resource.split("%")[0] : null;
+        return splitNamespaceStr(resource);
+    }
+
+    private static String splitNamespaceStr(String namespaceStr) {
+        return namespaceStr != null && namespaceStr.contains(NAMESPACE_SPLITER) ? namespaceStr.split(NAMESPACE_SPLITER)[0] : null;
     }
 }

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/StatUtil.java

@@ -227,40 +227,7 @@ public static void addInvoke(String key, int num, long rt, boolean success) {
     }
 
     public static void addInvoke(String key, long rt, boolean success) {
-        if (invokeCache.size() > MAX_KEY_NUM || secondInvokeCache.size() > MAX_KEY_NUM) {
-            return;
-        }
-        Invoke invoke = getAndSetInvoke(key);
-        if (invoke == null) {
-            return;
-        }
-
-        invoke.totalPv.getAndIncrement();
-        if (!success) {
-            invoke.failPv.getAndIncrement();
-        }
-        long now = nowSecond();
-        AtomicLong oldSecond = invoke.second;
-        if (oldSecond.get() == now) {
-            invoke.secondPv.getAndIncrement();
-        } else {
-            if (oldSecond.compareAndSet(oldSecond.get(), now)) {
-                if (invoke.secondPv.get() > invoke.topSecondPv.get()) {
-                    invoke.topSecondPv.set(invoke.secondPv.get());
-                }
-                invoke.secondPv.set(1);
-            } else {
-                invoke.secondPv.getAndIncrement();
-            }
-        }
-
-        invoke.sumRt.addAndGet(rt);
-        if (invoke.maxRt.get() < rt) {
-            invoke.maxRt.set(rt);
-        }
-        if (invoke.minRt.get() > rt) {
-            invoke.minRt.set(rt);
-        }
+        addInvoke(key, 1, rt, success);
     }
 
     public static SecondInvoke getAndSetSecondInvoke(String key) {
@@ -431,7 +398,6 @@ public static int avgRtInWindow(String key, int windowSeconds) {
     public static int maxRtInWindow(String key, int windowSeconds) {
         List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
         long maxRt = 0;
-        long totalPv = 0;
         for (int i = 0; i < windowSeconds && i < list.size(); i++) {
             if (maxRt < list.get(i).maxRt.get()) {
                 maxRt = list.get(i).maxRt.get();
@@ -443,7 +409,6 @@ public static int maxRtInWindow(String key, int windowSeconds) {
     public static int minRtInWindow(String key, int windowSeconds) {
         List<SecondInvoke> list = secondInvokeList(key, windowSeconds);
         long minRt = 0;
-        long totalPv = 0;
         for (int i = 0; i < windowSeconds && i < list.size(); i++) {
             if (minRt < list.get(i).minRt.get()) {
                 minRt = list.get(i).minRt.get();

mqtt-common/src/main/java/org/apache/rocketmq/mqtt/common/util/TopicUtils.java

@@ -31,7 +31,7 @@ public class TopicUtils {
      * @return
      */
     public static String normalizeTopic(String topic) {
-        if (topic == null) {
+        if (StringUtils.isBlank(topic)) {
             return null;
         }
         if (!topic.contains(Constants.MQTT_TOPIC_DELIMITER)) {
@@ -46,20 +46,20 @@ public static String normalizeTopic(String topic) {
     /**
      * /t2/t3/t4/
      *
-     * @param secondtopic
+     * @param secondTopic
      * @return
      */
-    public static String normalizeSecondTopic(String secondtopic) {
-        if (secondtopic == null || secondtopic.isEmpty()) {
+    public static String normalizeSecondTopic(String secondTopic) {
+        if (StringUtils.isBlank(secondTopic)) {
             return null;
         }
-        if (!secondtopic.startsWith(Constants.MQTT_TOPIC_DELIMITER)) {
-            secondtopic = Constants.MQTT_TOPIC_DELIMITER + secondtopic;
+        if (!secondTopic.startsWith(Constants.MQTT_TOPIC_DELIMITER)) {
+            secondTopic = Constants.MQTT_TOPIC_DELIMITER + secondTopic;
         }
-        if (!secondtopic.endsWith(Constants.MQTT_TOPIC_DELIMITER)) {
-            return secondtopic + Constants.MQTT_TOPIC_DELIMITER;
+        if (!secondTopic.endsWith(Constants.MQTT_TOPIC_DELIMITER)) {
+            return secondTopic + Constants.MQTT_TOPIC_DELIMITER;
         }
-        return secondtopic;
+        return secondTopic;
     }
 
     public static boolean isP2P(String secondTopic) {
@@ -93,10 +93,10 @@ public static boolean isP2pTopic(String topic) {
     }
 
     public static String getP2Peer(MqttTopic mqttTopic, String namespace) {
-        if (!isP2P(mqttTopic.getSecondTopic())) {
+        if (mqttTopic.getSecondTopic() == null || mqttTopic.getFirstTopic() == null) {
             return null;
         }
-        if (mqttTopic.getSecondTopic() == null || mqttTopic.getFirstTopic() == null) {
+        if (!isP2P(mqttTopic.getSecondTopic())) {
             return null;
         }
         if (mqttTopic.getFirstTopic().contains(Constants.NAMESPACE_SPLITER) && StringUtils.isNotBlank(namespace)) {

mqtt-common/src/test/java/org/apache/rocketmq/mqtt/common/test/hook/TestAbstractUpstreamHook.java

@@ -0,0 +1,64 @@
+/*
+ *
+ *  * 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.rocketmq.mqtt.common.test.hook;
+
+import io.netty.handler.codec.mqtt.MqttMessage;
+import org.apache.rocketmq.mqtt.common.hook.AbstractUpstreamHook;
+import org.apache.rocketmq.mqtt.common.hook.HookResult;
+import org.apache.rocketmq.mqtt.common.model.MqttMessageUpContext;
+import org.apache.rocketmq.mqtt.common.model.Remark;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.Assert.assertEquals;
+
+@RunWith(MockitoJUnitRunner.class)
+public class TestAbstractUpstreamHook {
+    private final int code = -200;
+    private final String remark = Remark.FAIL;
+
+    @Mock
+    private MqttMessageUpContext context;
+
+    @Mock
+    private MqttMessage mqttMessage;
+
+    private AbstractUpstreamHook upstreamHook = new AbstractUpstreamHook() {
+        @Override
+        public void register() {}
+
+        @Override
+        public CompletableFuture<HookResult> processMqttMessage(MqttMessageUpContext context, MqttMessage message) {
+            return HookResult.newHookResult(code, remark, null);
+        }
+    };
+
+    @Test
+    public void test() throws ExecutionException, InterruptedException {
+        CompletableFuture<HookResult> future = upstreamHook.doHook(context, mqttMessage);
+        assertEquals(code, future.get().getCode());
+    }
+
+}

mqtt-common/src/test/java/org/apache/rocketmq/mqtt/common/test/hook/TestHookResult.java

@@ -0,0 +1,48 @@
+/*
+ *
+ *  * 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.rocketmq.mqtt.common.test.hook;
+
+import org.apache.rocketmq.mqtt.common.hook.HookResult;
+import org.junit.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestHookResult {
+    private final int code = 200;
+    private final int subCode = 200;
+    private final String remark = "test";
+
+    @Test
+    public void test() throws ExecutionException, InterruptedException {
+        HookResult hookResult = new HookResult(code, remark, null);
+        CompletableFuture<HookResult> future = HookResult.newHookResult(code, remark, null);
+        assertEquals(hookResult, future.get());
+        assertTrue(future.isDone());
+
+        hookResult = new HookResult(code, subCode, remark, null);
+        future = HookResult.newHookResult(code, subCode, remark, null);
+        assertEquals(hookResult, future.get());
+        assertTrue(future.isDone());
+    }
+}