feat: add support for CoAP transport

distribution/conf/meta.conf

@@ -14,5 +14,5 @@
 #  limitations under the License.
 
 
-selfAddress=
-membersAddress=
+selfAddress=11.159.23.108:25000
+membersAddress=11.159.23.108:25000,11.159.23.111:25000,11.159.23.104:25000

distribution/conf/service.conf

@@ -14,11 +14,11 @@
 #  limitations under the License.
 
 
-username=
-secretKey=
+username=test
+secretKey=test
 
-NAMESRV_ADDR=
-eventNotifyRetryTopic=
-clientRetryTopic=
+NAMESRV_ADDR=11.159.23.108:9876
+eventNotifyRetryTopic=eventNotifyRetryTopic
+clientRetryTopic=clientRetryTopic
 
-metaAddr=
+metaAddr=11.159.23.108:25000,11.159.23.111:25000,11.159.23.104:25000

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

@@ -0,0 +1,163 @@
+/*
+ * 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.model;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class CoapMessage {
+    private int version;
+    private CoapMessageType type;
+    private int tokenLength;
+    private CoapMessageCode code;
+    private int messageId;
+    private byte[] token;
+    private List<CoapMessageOption> options = new ArrayList<>();
+    private byte[] payload;
+    private InetSocketAddress remoteAddress;
+
+    public CoapMessage(int version, CoapMessageType type, int tokenLength, CoapMessageCode code, int messageId, byte[] token, byte[] payload, InetSocketAddress remoteAddress) {
+        this.version = version;
+        this.type = type;
+        this.tokenLength = tokenLength;
+        this.code = code;
+        this.messageId = messageId;
+        this.token = token;
+        this.payload = payload;
+        this.remoteAddress = remoteAddress;
+    }
+
+    public CoapMessage(int version, CoapMessageType type, int tokenLength, CoapMessageCode code, int messageId, byte[] token, InetSocketAddress remoteAddress) {
+        this.version = version;
+        this.type = type;
+        this.tokenLength = tokenLength;
+        this.code = code;
+        this.messageId = messageId;
+        this.token = token;
+        this.remoteAddress = remoteAddress;
+    }
+
+    public int getVersion() {
+        return version;
+    }
+
+    public void setVersion(int version) {
+        this.version = version;
+    }
+
+    public CoapMessageType getType() {
+        return type;
+    }
+
+    public void setType(CoapMessageType type) {
+        this.type = type;
+    }
+
+    public int getTokenLength() {
+        return tokenLength;
+    }
+
+    public void setTokenLength(int tokenLength) {
+        this.tokenLength = tokenLength;
+    }
+
+    public CoapMessageCode getCode() {
+        return code;
+    }
+
+    public void setCode(CoapMessageCode code) {
+        this.code = code;
+    }
+
+    public int getMessageId() {
+        return messageId;
+    }
+
+    public void setMessageId(int messageId) {
+        this.messageId = messageId;
+    }
+
+    public byte[] getToken() {
+        return token;
+    }
+
+    public void setToken(byte[] token) {
+        this.token = token;
+    }
+
+    public List<CoapMessageOption> getOptions() {
+        return options;
+    }
+
+    public void clearOptions() {
+        this.options.clear();
+    }
+
+    public void setOptions(List<CoapMessageOption> options) {
+        this.options = options;
+    }
+
+    public void addOption(CoapMessageOption option) {
+        this.options.add(option);
+    }
+
+    public void addObserveOption(int value) {
+        this.options.add(new CoapMessageOption(CoapMessageOptionNumber.OBSERVE, intToByteArray(value)));
+    }
+
+    public byte[] getPayload() {
+        return payload;
+    }
+
+    public void setPayload(byte[] payload) {
+        this.payload = payload;
+    }
+
+    public InetSocketAddress getRemoteAddress() {
+        return remoteAddress;
+    }
+
+    public void setRemoteAddress(InetSocketAddress remoteAddress) {
+        this.remoteAddress = remoteAddress;
+    }
+
+    private byte[] intToByteArray(int value) {
+        byte[] byteArray = new byte[3];
+        byteArray[0] = (byte) (value >> 16);
+        byteArray[1] = (byte) (value >> 8);
+        byteArray[2] = (byte) value;
+        return byteArray;
+    }
+
+
+    @Override
+    public String toString() {
+        return "CoapMessage{" +
+                "version=" + version +
+                ", type=" + type +
+                ", tokenLength=" + tokenLength +
+                ", code=" + code +
+                ", messageId=" + messageId +
+                ", token=" + Arrays.toString(token) +
+                ", options=" + options +
+                ", payload=" + Arrays.toString(payload) +
+                ", remoteAddress=" + remoteAddress +
+                '}';
+    }
+}

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

@@ -0,0 +1,96 @@
+/*
+ * 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.model;
+
+public enum CoapMessageCode {
+    EMPTY(0),
+
+    // Request Code, 0.xx
+    GET(1),
+    POST(2),
+    PUT(3),
+    DELETE(4),
+
+    // Response Success Code, 2.xx
+    CREATED(65),
+    DELETED(66),
+    Valid(67),
+    CHANGED(68),
+    CONTENT(69),
+
+    // Response Client Error Code, 4.xx
+    BAD_REQUEST(128),
+    UNAUTHORIZED(129),
+    BAD_OPTION(130),
+    FORBIDDEN(131),
+    NOT_FOUND(132),
+    METHOD_NOT_ALLOWED(133),
+    NOT_ACCEPTABLE(134),
+    PRECONDITION_FAILED(140),
+    REQUEST_ENTITY_TOO_LARGE(141),
+    UNSUPPORTED_CONTENT_FORMAT(143),
+
+    // Response Server Error Code, 5.xx
+    INTERNAL_SERVER_ERROR(160),
+    NOT_IMPLEMENTED(161),
+    BAD_GATEWAY(162),
+    SERVICE_UNAVAILABLE(163),
+    GATEWAY_TIMEOUT(164),
+    PROXYING_NOT_SUPPORTED(165);
+
+    private static final CoapMessageCode[] VALUES;
+    private final int value;
+
+    private CoapMessageCode(int value) {
+        this.value = value;
+    }
+
+    public int value() {
+        return this.value;
+    }
+
+    public static CoapMessageCode valueOf(int code) {
+        if (code >= 0 && code < VALUES.length && VALUES[code] != null) {
+            return  VALUES[code];
+        } else {
+            throw new IllegalArgumentException("Unknown CoapMessageCode " + code);
+        }
+    }
+
+    public static boolean isRequestCode(CoapMessageCode code) {
+        return (code == GET) || (code == POST) || (code == PUT) || (code == DELETE);
+    }
+
+    public static boolean isEmptyCode(CoapMessageCode code) {
+        return code == EMPTY;
+    }
+
+    static {
+        CoapMessageCode[] values = values();
+        VALUES = new CoapMessageCode[192];  // Using 192 since the highest defined code is 192
+
+        for (CoapMessageCode code : values) {
+            int value = code.value;
+            if (VALUES[value] != null) {
+                throw new AssertionError("Value already in use: " + value + " by " + VALUES[value]);
+            }
+            VALUES[value] = code;
+        }
+    }
+
+
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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.model;
+
+public class CoapMessageOption {
+    private CoapMessageOptionNumber optionNumber;
+    private byte[] optionValue;
+
+    public CoapMessageOption(CoapMessageOptionNumber optionNumber, byte[] optionValue) {
+        this.optionNumber = optionNumber;
+        this.optionValue = optionValue;
+    }
+
+    public CoapMessageOption(int optionNumber, byte[] optionValue) {
+        this(CoapMessageOptionNumber.valueOf(optionNumber), optionValue);
+    }
+
+    public CoapMessageOptionNumber getOptionNumber() {
+        return optionNumber;
+    }
+
+    public void setOptionNumber(CoapMessageOptionNumber optionNumber) {
+        this.optionNumber = optionNumber;
+    }
+
+    public byte[] getOptionValue() {
+        return optionValue;
+    }
+
+    public void setOptionValue(byte[] optionValue) {
+        this.optionValue = optionValue;
+    }
+}