new

.gitignore

@@ -0,0 +1,33 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

common/.gitignore

@@ -0,0 +1,33 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

common/pom.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>commons</artifactId>
+        <groupId>com.shuhe</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>common</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>${lombok.version}</version>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.11.2</version>
+        </dependency>
+    </dependencies>
+
+
+</project>

common/src/main/java/com/shuhe/common/data/extensions/common/DataTypeMapping.java

@@ -0,0 +1,18 @@
+package com.shuhe.common.data.extensions.common;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.shuhe.common.data.kv.DataType;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class DataTypeMapping {
+
+    private DataType dataType;
+
+    @JsonCreator
+    public static DataTypeMapping forValue(String value) {
+        return new DataTypeMapping(DataType.valueOf(value.toUpperCase()));
+    }
+}

common/src/main/java/com/shuhe/common/data/extensions/common/ExtensionConfiguration.java

@@ -0,0 +1,12 @@
+package com.shuhe.common.data.extensions.common;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.Data;
+
+@Data
+public class ExtensionConfiguration {
+    private String id;
+    private String type;
+    private JsonNode configuration;
+    private String extensionConfiguration;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/Dlt645Configuration.java

@@ -0,0 +1,10 @@
+package com.shuhe.common.data.extensions.dlt645;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class Dlt645Configuration {
+    private List<Dlt645ServerConfiguration> servers;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/Dlt645ExtensionConstants.java

@@ -0,0 +1,22 @@
+package com.shuhe.common.data.extensions.dlt645;
+
+public class Dlt645ExtensionConstants {
+    public static final int DEFAULT_DLT645_TCP_PORT = 502;
+    public static final int DEFAULT_SOCKET_TIMEOUT = 3000; // in milliseconds
+
+    public static final int DEFAULT_POLL_PERIOD = 1000; // in milliseconds
+    public static final int NO_POLL_PERIOD_DEFINED = 0;
+    public static final int DEFAULT_INTERVAL_TIME = 40; // in milliseconds
+
+    public static final int DEFAULT_REGISTER_COUNT = 1;
+
+    public static final int DEFAULT_BIT_INDEX_FOR_BOOLEAN = 0;
+    public static final int NO_BIT_INDEX_DEFINED = -1;
+    public static final int DEFAULT_REGISTER_COUNT_FOR_BOOLEAN = 1;
+
+    public static final int MIN_BIT_INDEX_IN_REG = 0;
+    public static final int MAX_BIT_INDEX_IN_REG = 15;
+
+    public static final String LITTLE_ENDIAN_BYTE_ORDER = "LITTLE";
+    public static final String BIG_ENDIAN_BYTE_ORDER = "BIG";
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/Dlt645ServerConfiguration.java

@@ -0,0 +1,13 @@
+package com.shuhe.common.data.extensions.dlt645;
+
+import com.shuhe.common.data.extensions.dlt645.mapping.DeviceMapping;
+import com.shuhe.common.data.extensions.dlt645.transport.Dlt645TransportConfiguration;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class Dlt645ServerConfiguration {
+    private Dlt645TransportConfiguration transport;
+    private List<DeviceMapping> devices;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/mapping/DeviceMapping.java

@@ -0,0 +1,17 @@
+package com.shuhe.common.data.extensions.dlt645.mapping;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+import java.util.Collections;
+import java.util.List;
+
+@Data
+public class DeviceMapping {
+    private String unitId;
+    private String deviceCode;
+    private int attributesPollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int timeseriesPollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private List<PollingTagMapping> attributes = Collections.emptyList();
+    private List<PollingTagMapping> timeseries = Collections.emptyList();
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/mapping/PollingTagMapping.java

@@ -0,0 +1,13 @@
+package com.shuhe.common.data.extensions.dlt645.mapping;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true)
+public class PollingTagMapping extends TagMapping {
+    private int pollPeriod = ModbusExtensionConstants.NO_POLL_PERIOD_DEFINED;
+    //private DataTypeMapping type;
+    private String type;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/mapping/TagMapping.java

@@ -0,0 +1,13 @@
+package com.shuhe.common.data.extensions.dlt645.mapping;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+@Data
+public class TagMapping {
+    private String tag;
+    private int functionCode;
+    private int subType;
+    private String byteOrder = ModbusExtensionConstants.BIG_ENDIAN_BYTE_ORDER;
+    private int bit = ModbusExtensionConstants.NO_BIT_INDEX_DEFINED;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/mapping/TagValueMapping.java

@@ -0,0 +1,10 @@
+package com.shuhe.common.data.extensions.dlt645.mapping;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true)
+public class TagValueMapping extends TagMapping {
+    private Object value;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/transport/Dlt645IpTransportConfiguration.java

@@ -0,0 +1,11 @@
+package com.shuhe.common.data.extensions.dlt645.transport;
+
+import com.shuhe.common.data.extensions.dlt645.Dlt645ExtensionConstants;
+import lombok.Data;
+
+@Data
+public class Dlt645IpTransportConfiguration implements Dlt645TransportConfiguration {
+    private String host;
+    private int port = Dlt645ExtensionConstants.DEFAULT_DLT645_TCP_PORT;
+    private int timeout = Dlt645ExtensionConstants.DEFAULT_SOCKET_TIMEOUT;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/transport/Dlt645RtuTransportConfiguration.java

@@ -0,0 +1,17 @@
+package com.shuhe.common.data.extensions.dlt645.transport;
+
+import com.shuhe.common.data.extensions.dlt645.Dlt645ExtensionConstants;
+import lombok.Data;
+
+@Data
+public class Dlt645RtuTransportConfiguration implements Dlt645TransportConfiguration {
+    private String portName;
+    private int timeout = Dlt645ExtensionConstants.DEFAULT_SOCKET_TIMEOUT;
+    private String encoding;
+    private int baudRate;
+    private int dataBits;
+    private float stopBits;
+    private String parity;
+    private int pollPeriod = Dlt645ExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int intervalTime = Dlt645ExtensionConstants.DEFAULT_INTERVAL_TIME;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/transport/Dlt645TcpTransportConfiguration.java

@@ -0,0 +1,14 @@
+package com.shuhe.common.data.extensions.dlt645.transport;
+
+import com.shuhe.common.data.extensions.dlt645.Dlt645ExtensionConstants;
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true, includeFieldNames=true)
+public class Dlt645TcpTransportConfiguration extends Dlt645IpTransportConfiguration {
+    boolean rtuOverTcp;
+    boolean reconnect;
+    private int pollPeriod = Dlt645ExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int intervalTime = Dlt645ExtensionConstants.DEFAULT_INTERVAL_TIME;
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/transport/Dlt645TransportConfiguration.java

@@ -0,0 +1,16 @@
+package com.shuhe.common.data.extensions.dlt645.transport;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+@JsonTypeInfo(
+        use = JsonTypeInfo.Id.NAME,
+        include = JsonTypeInfo.As.PROPERTY,
+        property = "type")
+@JsonSubTypes({
+        @JsonSubTypes.Type(value = Dlt645TcpTransportConfiguration.class, name = "tcp"),
+        @JsonSubTypes.Type(value = Dlt645UdpTransportConfiguration.class, name = "udp"),
+        @JsonSubTypes.Type(value = Dlt645RtuTransportConfiguration.class, name = "rtu")})
+public interface Dlt645TransportConfiguration {
+
+}

common/src/main/java/com/shuhe/common/data/extensions/dlt645/transport/Dlt645UdpTransportConfiguration.java

@@ -0,0 +1,4 @@
+package com.shuhe.common.data.extensions.dlt645.transport;
+
+public class Dlt645UdpTransportConfiguration extends Dlt645IpTransportConfiguration {
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/ModbusConfiguration.java

@@ -0,0 +1,10 @@
+package com.shuhe.common.data.extensions.modbus;
+
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class ModbusConfiguration {
+    private List<ModbusServerConfiguration> servers;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/ModbusExtensionConstants.java

@@ -0,0 +1,29 @@
+package com.shuhe.common.data.extensions.modbus;
+
+public class ModbusExtensionConstants {
+    public static final int DEFAULT_MODBUS_TCP_PORT = 502;
+    public static final int DEFAULT_SOCKET_TIMEOUT = 3000; // in milliseconds
+
+    public static final int DEFAULT_POLL_PERIOD = 1000; // in milliseconds
+    public static final int NO_POLL_PERIOD_DEFINED = 0;
+    public static final int DEFAULT_INTERVAL_TIME = 40; // in milliseconds
+
+    public static final int DEFAULT_REGISTER_COUNT = 1;
+
+    public static final int DEFAULT_BIT_INDEX_FOR_BOOLEAN = 0;
+    public static final int NO_BIT_INDEX_DEFINED = -1;
+    public static final int DEFAULT_REGISTER_COUNT_FOR_BOOLEAN = 1;
+
+    public static final int MIN_BIT_INDEX_IN_REG = 0;
+    public static final int MAX_BIT_INDEX_IN_REG = 15;
+
+    public static final String LITTLE_ENDIAN_BYTE_ORDER = "LITTLE";
+    public static final String BIG_ENDIAN_BYTE_ORDER = "BIG";
+
+    public static final int WORD_REGISTER_COUNT = 1;
+    public static final int INTEGER_REGISTER_COUNT = 2;
+    public static final int LONG_REGISTER_COUNT = 4;
+
+    public static final int FLOAT_REGISTER_COUNT = 2;
+    public static final int DOUBLE_REGISTER_COUNT = 4;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/ModbusServerConfiguration.java

@@ -0,0 +1,13 @@
+package com.shuhe.common.data.extensions.modbus;
+
+import com.shuhe.common.data.extensions.modbus.mapping.DeviceMapping;
+import com.shuhe.common.data.extensions.modbus.transport.ModbusTransportConfiguration;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class ModbusServerConfiguration {
+    private ModbusTransportConfiguration transport;
+    private List<DeviceMapping> devices;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/mapping/DeviceMapping.java

@@ -0,0 +1,17 @@
+package com.shuhe.common.data.extensions.modbus.mapping;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+import java.util.Collections;
+import java.util.List;
+
+@Data
+public class DeviceMapping {
+    private int unitId;
+    private String deviceName;
+    private int attributesPollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int timeseriesPollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private List<PollingTagMapping> attributes = Collections.emptyList();
+    private List<PollingTagMapping> timeseries = Collections.emptyList();
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/mapping/PollingTagMapping.java

@@ -0,0 +1,14 @@
+package com.shuhe.common.data.extensions.modbus.mapping;
+
+import com.shuhe.common.data.extensions.common.DataTypeMapping;
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true)
+public class PollingTagMapping extends TagMapping {
+    private int pollPeriod = ModbusExtensionConstants.NO_POLL_PERIOD_DEFINED;
+    //private DataTypeMapping type;
+    private String type;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/mapping/TagMapping.java

@@ -0,0 +1,14 @@
+package com.shuhe.common.data.extensions.modbus.mapping;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+@Data
+public class TagMapping {
+    private String tag;
+    private int functionCode;
+    private int address;
+    private int registerCount = ModbusExtensionConstants.DEFAULT_REGISTER_COUNT;
+    private String byteOrder = ModbusExtensionConstants.BIG_ENDIAN_BYTE_ORDER;
+    private int bit = ModbusExtensionConstants.NO_BIT_INDEX_DEFINED;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/mapping/TagValueMapping.java

@@ -0,0 +1,10 @@
+package com.shuhe.common.data.extensions.modbus.mapping;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true)
+public class TagValueMapping extends TagMapping {
+    private Object value;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/transport/ModbusIpTransportConfiguration.java

@@ -0,0 +1,11 @@
+package com.shuhe.common.data.extensions.modbus.transport;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+@Data
+public class ModbusIpTransportConfiguration implements ModbusTransportConfiguration {
+    private String host;
+    private int port = ModbusExtensionConstants.DEFAULT_MODBUS_TCP_PORT;
+    private int timeout = ModbusExtensionConstants.DEFAULT_SOCKET_TIMEOUT;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/transport/ModbusRtuTransportConfiguration.java

@@ -0,0 +1,17 @@
+package com.shuhe.common.data.extensions.modbus.transport;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+
+@Data
+public class ModbusRtuTransportConfiguration implements ModbusTransportConfiguration {
+    private String portName;
+    private int timeout = ModbusExtensionConstants.DEFAULT_SOCKET_TIMEOUT;
+    private String encoding;
+    private int baudRate;
+    private int dataBits;
+    private float stopBits;
+    private String parity;
+    private int pollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int intervalTime = ModbusExtensionConstants.DEFAULT_INTERVAL_TIME;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/transport/ModbusTcpTransportConfiguration.java

@@ -0,0 +1,14 @@
+package com.shuhe.common.data.extensions.modbus.transport;
+
+import com.shuhe.common.data.extensions.modbus.ModbusExtensionConstants;
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString(callSuper=true, includeFieldNames=true)
+public class ModbusTcpTransportConfiguration extends ModbusIpTransportConfiguration {
+    boolean rtuOverTcp;
+    boolean reconnect;
+    private int pollPeriod = ModbusExtensionConstants.DEFAULT_POLL_PERIOD;
+    private int intervalTime = ModbusExtensionConstants.DEFAULT_INTERVAL_TIME;
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/transport/ModbusTransportConfiguration.java

@@ -0,0 +1,17 @@
+package com.shuhe.common.data.extensions.modbus.transport;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+
+
+@JsonTypeInfo(
+        use = JsonTypeInfo.Id.NAME,
+        include = JsonTypeInfo.As.PROPERTY,
+        property = "type")
+@JsonSubTypes({
+        @JsonSubTypes.Type(value = ModbusTcpTransportConfiguration.class, name = "tcp"),
+        @JsonSubTypes.Type(value = ModbusUdpTransportConfiguration.class, name = "udp"),
+        @JsonSubTypes.Type(value = ModbusRtuTransportConfiguration.class, name = "rtu")})
+public interface ModbusTransportConfiguration {
+
+}

common/src/main/java/com/shuhe/common/data/extensions/modbus/transport/ModbusUdpTransportConfiguration.java

@@ -0,0 +1,5 @@
+package com.shuhe.common.data.extensions.modbus.transport;
+
+public class ModbusUdpTransportConfiguration extends ModbusIpTransportConfiguration {
+
+}

common/src/main/java/com/shuhe/common/data/kv/Aggregation.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+/**
+ * Created by ashvayka on 20.02.17.
+ */
+public enum Aggregation {
+
+    MIN, MAX, AVG, SUM, COUNT, NONE;
+
+}

common/src/main/java/com/shuhe/common/data/kv/AttributeKey.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author Andrew Shvayka
+ */
+@Data
+public class AttributeKey implements Serializable {
+    private final String scope;
+    private final String attributeKey;
+}

common/src/main/java/com/shuhe/common/data/kv/AttributeKvEntry.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+/**
+ * @author Andrew Shvayka
+ */
+public interface AttributeKvEntry extends KvEntry {
+
+    long getLastUpdateTs();
+
+}

common/src/main/java/com/shuhe/common/data/kv/BaseAttributeKvEntry.java

@@ -0,0 +1,113 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Optional;
+
+/**
+ * @author Andrew Shvayka
+ */
+public class BaseAttributeKvEntry implements AttributeKvEntry {
+
+    private final long lastUpdateTs;
+    private final KvEntry kv;
+
+    public BaseAttributeKvEntry(KvEntry kv, long lastUpdateTs) {
+        this.kv = kv;
+        this.lastUpdateTs = lastUpdateTs;
+    }
+
+    public BaseAttributeKvEntry(long lastUpdateTs, KvEntry kv) {
+        this(kv, lastUpdateTs);
+    }
+
+    @Override
+    public long getLastUpdateTs() {
+        return lastUpdateTs;
+    }
+
+    @Override
+    public String getKey() {
+        return kv.getKey();
+    }
+
+    @Override
+    public DataType getDataType() {
+        return kv.getDataType();
+    }
+
+    @Override
+    public Optional<String> getStrValue() {
+        return kv.getStrValue();
+    }
+
+    @Override
+    public Optional<Long> getLongValue() {
+        return kv.getLongValue();
+    }
+
+    @Override
+    public Optional<Boolean> getBooleanValue() {
+        return kv.getBooleanValue();
+    }
+
+    @Override
+    public Optional<Double> getDoubleValue() {
+        return kv.getDoubleValue();
+    }
+
+    @Override
+    public Optional<String> getJsonValue() {
+        return kv.getJsonValue();
+    }
+
+    @Override
+    public String getValueAsString() {
+        return kv.getValueAsString();
+    }
+
+    @Override
+    public Object getValue() {
+        return kv.getValue();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        BaseAttributeKvEntry that = (BaseAttributeKvEntry) o;
+
+        if (lastUpdateTs != that.lastUpdateTs) return false;
+        return kv.equals(that.kv);
+
+    }
+
+    @Override
+    public int hashCode() {
+        int result = (int) (lastUpdateTs ^ (lastUpdateTs >>> 32));
+        result = 31 * result + kv.hashCode();
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return "BaseAttributeKvEntry{" +
+                "lastUpdateTs=" + lastUpdateTs +
+                ", kv=" + kv +
+                '}';
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/BaseDeleteTsKvQuery.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import lombok.Data;
+
+@Data
+public class BaseDeleteTsKvQuery extends BaseTsKvQuery implements DeleteTsKvQuery {
+
+    private final Boolean rewriteLatestIfDeleted;
+
+    public BaseDeleteTsKvQuery(String key, long startTs, long endTs, boolean rewriteLatestIfDeleted) {
+        super(key, startTs, endTs);
+        this.rewriteLatestIfDeleted = rewriteLatestIfDeleted;
+    }
+
+    public BaseDeleteTsKvQuery(String key, long startTs, long endTs) {
+        this(key, startTs, endTs, false);
+    }
+
+
+}

common/src/main/java/com/shuhe/common/data/kv/BaseReadTsKvQuery.java

@@ -0,0 +1,49 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import lombok.Data;
+
+@Data
+public class BaseReadTsKvQuery extends BaseTsKvQuery implements ReadTsKvQuery {
+
+    private final long interval;
+    private final int limit;
+    private final Aggregation aggregation;
+    private final String order;
+
+    public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation) {
+        this(key, startTs, endTs, interval, limit, aggregation, "DESC");
+    }
+
+    public BaseReadTsKvQuery(String key, long startTs, long endTs, long interval, int limit, Aggregation aggregation,
+                             String order) {
+        super(key, startTs, endTs);
+        this.interval = interval;
+        this.limit = limit;
+        this.aggregation = aggregation;
+        this.order = order;
+    }
+
+    public BaseReadTsKvQuery(String key, long startTs, long endTs) {
+        this(key, startTs, endTs, endTs - startTs, 1, Aggregation.AVG, "DESC");
+    }
+
+    public BaseReadTsKvQuery(String key, long startTs, long endTs, int limit, String order) {
+        this(key, startTs, endTs, endTs - startTs, limit, Aggregation.NONE, order);
+    }
+
+}

common/src/main/java/com/shuhe/common/data/kv/BaseTsKvQuery.java

@@ -0,0 +1,33 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import lombok.Data;
+
+@Data
+public class BaseTsKvQuery implements TsKvQuery {
+
+    private final String key;
+    private final long startTs;
+    private final long endTs;
+
+    public BaseTsKvQuery(String key, long startTs, long endTs) {
+        this.key = key;
+        this.startTs = startTs;
+        this.endTs = endTs;
+    }
+
+}

common/src/main/java/com/shuhe/common/data/kv/BasicKvEntry.java

@@ -0,0 +1,78 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public abstract class BasicKvEntry implements KvEntry {
+
+    private final String key;
+
+    protected BasicKvEntry(String key) {
+        this.key = key;
+    }
+
+    @Override
+    public String getKey() {
+        return key;
+    }
+
+    @Override
+    public Optional<String> getStrValue() {
+        return Optional.ofNullable(null);
+    }
+
+    @Override
+    public Optional<Long> getLongValue() {
+        return Optional.ofNullable(null);
+    }
+
+    @Override
+    public Optional<Boolean> getBooleanValue() {
+        return Optional.ofNullable(null);
+    }
+
+    @Override
+    public Optional<Double> getDoubleValue() {
+        return Optional.ofNullable(null);
+    }
+
+    @Override
+    public Optional<String> getJsonValue() {
+        return Optional.ofNullable(null);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof BasicKvEntry)) return false;
+        BasicKvEntry that = (BasicKvEntry) o;
+        return Objects.equals(key, that.key);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(key);
+    }
+
+    @Override
+    public String toString() {
+        return "BasicKvEntry{" +
+                "key='" + key + '\'' +
+                '}';
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/BasicTsKvEntry.java

@@ -0,0 +1,102 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class BasicTsKvEntry implements TsKvEntry {
+
+    private final long ts;
+    private final KvEntry kv;
+
+    public BasicTsKvEntry(long ts, KvEntry kv) {
+        this.ts = ts;
+        this.kv = kv;
+    }
+
+    @Override
+    public String getKey() {
+        return kv.getKey();
+    }
+
+    @Override
+    public DataType getDataType() {
+        return kv.getDataType();
+    }
+
+    @Override
+    public Optional<String> getStrValue() {
+        return kv.getStrValue();
+    }
+
+    @Override
+    public Optional<Long> getLongValue() {
+        return kv.getLongValue();
+    }
+
+    @Override
+    public Optional<Boolean> getBooleanValue() {
+        return kv.getBooleanValue();
+    }
+
+    @Override
+    public Optional<Double> getDoubleValue() {
+        return kv.getDoubleValue();
+    }
+
+    @Override
+    public Optional<String> getJsonValue() {
+        return kv.getJsonValue();
+    }
+
+    @Override
+    public Object getValue() {
+        return kv.getValue();
+    }
+
+    @Override
+    public long getTs() {
+        return ts;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof BasicTsKvEntry)) return false;
+        BasicTsKvEntry that = (BasicTsKvEntry) o;
+        return getTs() == that.getTs() &&
+                Objects.equals(kv, that.kv);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getTs(), kv);
+    }
+
+    @Override
+    public String toString() {
+        return "BasicTsKvEntry{" +
+                "ts=" + ts +
+                ", kv=" + kv +
+                '}';
+    }
+
+    @Override
+    public String getValueAsString() {
+        return kv.getValueAsString();
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/BooleanDataEntry.java

@@ -0,0 +1,69 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class BooleanDataEntry extends BasicKvEntry {
+    private final Boolean value;
+
+    public BooleanDataEntry(String key, Boolean value) {
+        super(key);
+        this.value = value;
+    }
+
+    @Override
+    public DataType getDataType() {
+        return DataType.BOOLEAN;
+    }
+
+    @Override
+    public Optional<Boolean> getBooleanValue() {
+        return Optional.ofNullable(value);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof BooleanDataEntry)) return false;
+        if (!super.equals(o)) return false;
+        BooleanDataEntry that = (BooleanDataEntry) o;
+        return Objects.equals(value, that.value);
+    }
+
+    @Override
+    public Object getValue() {
+        return value;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), value);
+    }
+
+    @Override
+    public String toString() {
+        return "BooleanDataEntry{" +
+                "value=" + value +
+                "} " + super.toString();
+    }
+
+    @Override
+    public String getValueAsString() {
+        return Boolean.toString(value);
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/DataType.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+public enum DataType {
+
+    STRING, LONG, BOOLEAN, DOUBLE, JSON;
+
+}

common/src/main/java/com/shuhe/common/data/kv/DeleteTsKvQuery.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+public interface DeleteTsKvQuery extends TsKvQuery {
+
+    Boolean getRewriteLatestIfDeleted();
+
+}

common/src/main/java/com/shuhe/common/data/kv/DoubleDataEntry.java

@@ -0,0 +1,70 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class DoubleDataEntry extends BasicKvEntry {
+
+    private final Double value;
+
+    public DoubleDataEntry(String key, Double value) {
+        super(key);
+        this.value = value;
+    }
+
+    @Override
+    public DataType getDataType() {
+        return DataType.DOUBLE;
+    }
+
+    @Override
+    public Optional<Double> getDoubleValue() {
+        return Optional.ofNullable(value);
+    }
+
+    @Override
+    public Object getValue() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof DoubleDataEntry)) return false;
+        if (!super.equals(o)) return false;
+        DoubleDataEntry that = (DoubleDataEntry) o;
+        return Objects.equals(value, that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), value);
+    }
+
+    @Override
+    public String toString() {
+        return "DoubleDataEntry{" +
+                "value=" + value +
+                "} " + super.toString();
+    }
+    
+    @Override
+    public String getValueAsString() {
+        return Double.toString(value);
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/JsonDataEntry.java

@@ -0,0 +1,69 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class JsonDataEntry extends BasicKvEntry {
+    private final String value;
+
+    public JsonDataEntry(String key, String value) {
+        super(key);
+        this.value = value;
+    }
+
+    @Override
+    public DataType getDataType() {
+        return DataType.JSON;
+    }
+
+    @Override
+    public Optional<String> getJsonValue() {
+        return Optional.ofNullable(value);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof JsonDataEntry)) return false;
+        if (!super.equals(o)) return false;
+        JsonDataEntry that = (JsonDataEntry) o;
+        return Objects.equals(value, that.value);
+    }
+
+    @Override
+    public Object getValue() {
+        return value;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), value);
+    }
+
+    @Override
+    public String toString() {
+        return "JsonDataEntry{" +
+                "value=" + value +
+                "} " + super.toString();
+    }
+
+    @Override
+    public String getValueAsString() {
+        return value;
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/KvEntry.java

@@ -0,0 +1,45 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.io.Serializable;
+import java.util.Optional;
+
+/**
+ * Represents attribute or any other KV data entry
+ *
+ * @author ashvayka
+ */
+public interface KvEntry extends Serializable {
+
+    String getKey();
+
+    DataType getDataType();
+
+    Optional<String> getStrValue();
+
+    Optional<Long> getLongValue();
+
+    Optional<Boolean> getBooleanValue();
+
+    Optional<Double> getDoubleValue();
+
+    Optional<String> getJsonValue();
+
+    String getValueAsString();
+
+    Object getValue();
+}

common/src/main/java/com/shuhe/common/data/kv/LongDataEntry.java

@@ -0,0 +1,70 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class LongDataEntry extends BasicKvEntry {
+
+    private final Long value;
+
+    public LongDataEntry(String key, Long value) {
+        super(key);
+        this.value = value;
+    }
+
+    @Override
+    public DataType getDataType() {
+        return DataType.LONG;
+    }
+
+    @Override
+    public Optional<Long> getLongValue() {
+        return Optional.ofNullable(value);
+    }
+
+    @Override
+    public Object getValue() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof LongDataEntry)) return false;
+        if (!super.equals(o)) return false;
+        LongDataEntry that = (LongDataEntry) o;
+        return Objects.equals(value, that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), value);
+    }
+
+    @Override
+    public String toString() {
+        return "LongDataEntry{" +
+                "value=" + value +
+                "} " + super.toString();
+    }
+    
+    @Override
+    public String getValueAsString() {
+        return Long.toString(value);
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/ReadTsKvQuery.java

@@ -0,0 +1,28 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+public interface ReadTsKvQuery extends TsKvQuery {
+
+    long getInterval();
+
+    int getLimit();
+
+    Aggregation getAggregation();
+
+    String getOrder();
+
+}

common/src/main/java/com/shuhe/common/data/kv/StringDataEntry.java

@@ -0,0 +1,74 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+import java.util.Objects;
+import java.util.Optional;
+
+public class StringDataEntry extends BasicKvEntry {
+
+    private static final long serialVersionUID = 1L;
+    private final String value;
+
+    public static final String StringByteCharsetName = "ISO-8859-1";
+
+    public StringDataEntry(String key, String value) {
+        super(key);
+        this.value = value;
+    }
+
+    @Override
+    public DataType getDataType() {
+        return DataType.STRING;
+    }
+
+    @Override
+    public Optional<String> getStrValue() {
+        return Optional.ofNullable(value);
+    }
+
+    @Override
+    public Object getValue() {
+        return value;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+        if (!(o instanceof StringDataEntry))
+            return false;
+        if (!super.equals(o))
+            return false;
+        StringDataEntry that = (StringDataEntry) o;
+        return Objects.equals(value, that.value);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), value);
+    }
+
+    @Override
+    public String toString() {
+        return "StringDataEntry{" + "value='" + value + '\'' + "} " + super.toString();
+    }
+    
+    @Override
+    public String getValueAsString() {
+        return value;
+    }
+}

common/src/main/java/com/shuhe/common/data/kv/TsKvEntry.java

@@ -0,0 +1,28 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+/**
+ * Represents time series KV data entry
+ * 
+ * @author ashvayka
+ *
+ */
+public interface TsKvEntry extends KvEntry {
+
+    long getTs();
+
+}

common/src/main/java/com/shuhe/common/data/kv/TsKvQuery.java

@@ -0,0 +1,26 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.common.data.kv;
+
+public interface TsKvQuery {
+
+    String getKey();
+
+    long getStartTs();
+
+    long getEndTs();
+
+}

netty-mqtt/.gitignore

@@ -0,0 +1,33 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/

netty-mqtt/pom.xml

@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>commons</artifactId>
+        <groupId>com.shuhe</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>netty-mqtt</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+
+    <name>Netty MQTT Client</name>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec-mqtt</artifactId>
+            <version>${netty.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-handler</artifactId>
+            <version>${netty.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.findbugs</groupId>
+            <artifactId>jsr305</artifactId>
+            <version>3.0.1</version>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <extensions>
+            <extension>
+                <groupId>org.apache.maven.wagon</groupId>
+                <artifactId>wagon-ssh</artifactId>
+                <version>2.6</version>
+            </extension>
+        </extensions>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.1</version>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>2.4</version>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
+                        </manifest>
+                    </archive>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

netty-mqtt/src/main/java/com/shuhe/mqtt/ChannelClosedException.java

@@ -0,0 +1,43 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+/**
+ * Created by Valerii Sosliuk on 12/26/2017.
+ */
+public class ChannelClosedException extends RuntimeException {
+
+    private static final long serialVersionUID = 6266638352424706909L;
+
+    public ChannelClosedException() {
+    }
+
+    public ChannelClosedException(String message) {
+        super(message);
+    }
+
+    public ChannelClosedException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public ChannelClosedException(Throwable cause) {
+        super(cause);
+    }
+
+    public ChannelClosedException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+        super(message, cause, enableSuppression, writableStackTrace);
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttChannelHandler.java

@@ -0,0 +1,265 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import com.google.common.collect.ImmutableSet;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.mqtt.*;
+import io.netty.util.CharsetUtil;
+import io.netty.util.concurrent.Promise;
+
+final class MqttChannelHandler extends SimpleChannelInboundHandler<MqttMessage> {
+
+    private final com.shuhe.mqtt.MqttClientImpl client;
+    private final Promise<com.shuhe.mqtt.MqttConnectResult> connectFuture;
+
+    MqttChannelHandler(com.shuhe.mqtt.MqttClientImpl client, Promise<com.shuhe.mqtt.MqttConnectResult> connectFuture) {
+        this.client = client;
+        this.connectFuture = connectFuture;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
+        switch (msg.fixedHeader().messageType()) {
+            case CONNACK:
+                handleConack(ctx.channel(), (MqttConnAckMessage) msg);
+                break;
+            case SUBACK:
+                handleSubAck((MqttSubAckMessage) msg);
+                break;
+            case PUBLISH:
+                handlePublish(ctx.channel(), (MqttPublishMessage) msg);
+                break;
+            case UNSUBACK:
+                handleUnsuback((MqttUnsubAckMessage) msg);
+                break;
+            case PUBACK:
+                handlePuback((MqttPubAckMessage) msg);
+                break;
+            case PUBREC:
+                handlePubrec(ctx.channel(), msg);
+                break;
+            case PUBREL:
+                handlePubrel(ctx.channel(), msg);
+                break;
+            case PUBCOMP:
+                handlePubcomp(msg);
+                break;
+        }
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        super.channelActive(ctx);
+
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
+        MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(
+                this.client.getClientConfig().getProtocolVersion().protocolName(),  // Protocol Name
+                this.client.getClientConfig().getProtocolVersion().protocolLevel(), // Protocol Level
+                this.client.getClientConfig().getUsername() != null,                // Has Username
+                this.client.getClientConfig().getPassword() != null,                // Has Password
+                this.client.getClientConfig().getLastWill() != null                 // Will Retain
+                        && this.client.getClientConfig().getLastWill().isRetain(),
+                this.client.getClientConfig().getLastWill() != null                 // Will QOS
+                        ? this.client.getClientConfig().getLastWill().getQos().value()
+                        : 0,
+                this.client.getClientConfig().getLastWill() != null,                // Has Will
+                this.client.getClientConfig().isCleanSession(),                     // Clean Session
+                this.client.getClientConfig().getTimeoutSeconds()                   // Timeout
+        );
+        MqttConnectPayload payload = new MqttConnectPayload(
+                this.client.getClientConfig().getClientId(),
+                this.client.getClientConfig().getLastWill() != null ? this.client.getClientConfig().getLastWill().getTopic() : null,
+                this.client.getClientConfig().getLastWill() != null ? this.client.getClientConfig().getLastWill().getMessage().getBytes(CharsetUtil.UTF_8) : null,
+                this.client.getClientConfig().getUsername(),
+                this.client.getClientConfig().getPassword() != null ? this.client.getClientConfig().getPassword().getBytes(CharsetUtil.UTF_8) : null
+        );
+        ctx.channel().writeAndFlush(new MqttConnectMessage(fixedHeader, variableHeader, payload));
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        super.channelInactive(ctx);
+    }
+
+    private void invokeHandlersForIncomingPublish(MqttPublishMessage message) {
+        boolean handlerInvoked = false;
+        for (com.shuhe.mqtt.MqttSubscription subscription : ImmutableSet.copyOf(this.client.getSubscriptions().values())) {
+            if (subscription.matches(message.variableHeader().topicName())) {
+                if (subscription.isOnce() && subscription.isCalled()) {
+                    continue;
+                }
+                message.payload().markReaderIndex();
+                subscription.setCalled(true);
+                subscription.getHandler().onMessage(message.variableHeader().topicName(), message.payload());
+                if (subscription.isOnce()) {
+                    this.client.off(subscription.getTopic(), subscription.getHandler());
+                }
+                message.payload().resetReaderIndex();
+                handlerInvoked = true;
+            }
+        }
+        if (!handlerInvoked && client.getDefaultHandler() != null) {
+            client.getDefaultHandler().onMessage(message.variableHeader().topicName(), message.payload());
+        }
+        message.payload().release();
+    }
+
+    private void handleConack(Channel channel, MqttConnAckMessage message) {
+        switch (message.variableHeader().connectReturnCode()) {
+            case CONNECTION_ACCEPTED:
+                this.connectFuture.setSuccess(new com.shuhe.mqtt.MqttConnectResult(true, MqttConnectReturnCode.CONNECTION_ACCEPTED, channel.closeFuture()));
+
+                this.client.getPendingSubscriptions().entrySet().stream().filter((e) -> !e.getValue().isSent()).forEach((e) -> {
+                    channel.write(e.getValue().getSubscribeMessage());
+                    e.getValue().setSent(true);
+                });
+
+                this.client.getPendingPublishes().forEach((id, publish) -> {
+                    if (publish.isSent()) return;
+                    channel.write(publish.getMessage());
+                    publish.setSent(true);
+                    if (publish.getQos() == MqttQoS.AT_MOST_ONCE) {
+                        publish.getFuture().setSuccess(null); //We don't get an ACK for QOS 0
+                        this.client.getPendingPublishes().remove(publish.getMessageId());
+                    }
+                });
+                channel.flush();
+                if (this.client.isReconnect()) {
+                    this.client.onSuccessfulReconnect();
+                }
+                break;
+
+            case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:
+            case CONNECTION_REFUSED_IDENTIFIER_REJECTED:
+            case CONNECTION_REFUSED_NOT_AUTHORIZED:
+            case CONNECTION_REFUSED_SERVER_UNAVAILABLE:
+            case CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION:
+                this.connectFuture.setSuccess(new com.shuhe.mqtt.MqttConnectResult(false, message.variableHeader().connectReturnCode(), channel.closeFuture()));
+                channel.close();
+                // Don't start reconnect logic here
+                break;
+        }
+    }
+
+    private void handleSubAck(MqttSubAckMessage message) {
+        com.shuhe.mqtt.MqttPendingSubscription pendingSubscription = this.client.getPendingSubscriptions().remove(message.variableHeader().messageId());
+        if (pendingSubscription == null) {
+            return;
+        }
+        pendingSubscription.onSubackReceived();
+        for (com.shuhe.mqtt.MqttPendingSubscription.MqttPendingHandler handler : pendingSubscription.getHandlers()) {
+            com.shuhe.mqtt.MqttSubscription subscription = new com.shuhe.mqtt.MqttSubscription(pendingSubscription.getTopic(), handler.getHandler(), handler.isOnce());
+            this.client.getSubscriptions().put(pendingSubscription.getTopic(), subscription);
+            this.client.getHandlerToSubscribtion().put(handler.getHandler(), subscription);
+        }
+        this.client.getPendingSubscribeTopics().remove(pendingSubscription.getTopic());
+
+        this.client.getServerSubscriptions().add(pendingSubscription.getTopic());
+
+        if (!pendingSubscription.getFuture().isDone()) {
+            pendingSubscription.getFuture().setSuccess(null);
+        }
+    }
+
+    private void handlePublish(Channel channel, MqttPublishMessage message) {
+        switch (message.fixedHeader().qosLevel()) {
+            case AT_MOST_ONCE:
+                invokeHandlersForIncomingPublish(message);
+                break;
+
+            case AT_LEAST_ONCE:
+                invokeHandlersForIncomingPublish(message);
+                if (message.variableHeader().packetId() != -1) {
+                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
+                    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
+                    channel.writeAndFlush(new MqttPubAckMessage(fixedHeader, variableHeader));
+                }
+                break;
+
+            case EXACTLY_ONCE:
+                if (message.variableHeader().packetId() != -1) {
+                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
+                    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
+                    MqttMessage pubrecMessage = new MqttMessage(fixedHeader, variableHeader);
+
+                    com.shuhe.mqtt.MqttIncomingQos2Publish incomingQos2Publish = new com.shuhe.mqtt.MqttIncomingQos2Publish(message);
+                    this.client.getQos2PendingIncomingPublishes().put(message.variableHeader().packetId(), incomingQos2Publish);
+                    message.payload().retain();
+
+                    channel.writeAndFlush(pubrecMessage);
+                }
+                break;
+        }
+    }
+
+    private void handleUnsuback(MqttUnsubAckMessage message) {
+        com.shuhe.mqtt.MqttPendingUnsubscription unsubscription = this.client.getPendingServerUnsubscribes().get(message.variableHeader().messageId());
+        if (unsubscription == null) {
+            return;
+        }
+        unsubscription.onUnsubackReceived();
+        this.client.getServerSubscriptions().remove(unsubscription.getTopic());
+        unsubscription.getFuture().setSuccess(null);
+        this.client.getPendingServerUnsubscribes().remove(message.variableHeader().messageId());
+    }
+
+    private void handlePuback(MqttPubAckMessage message) {
+        com.shuhe.mqtt.MqttPendingPublish pendingPublish = this.client.getPendingPublishes().get(message.variableHeader().messageId());
+        if (pendingPublish == null) {
+            return;
+        }
+        pendingPublish.getFuture().setSuccess(null);
+        pendingPublish.onPubackReceived();
+        this.client.getPendingPublishes().remove(message.variableHeader().messageId());
+        pendingPublish.getPayload().release();
+    }
+
+    private void handlePubrec(Channel channel, MqttMessage message) {
+        com.shuhe.mqtt.MqttPendingPublish pendingPublish = this.client.getPendingPublishes().get(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
+        pendingPublish.onPubackReceived();
+
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
+        MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
+        MqttMessage pubrelMessage = new MqttMessage(fixedHeader, variableHeader);
+        channel.writeAndFlush(pubrelMessage);
+
+        pendingPublish.setPubrelMessage(pubrelMessage);
+        pendingPublish.startPubrelRetransmissionTimer(this.client.getEventLoop().next(), this.client::sendAndFlushPacket);
+    }
+
+    private void handlePubrel(Channel channel, MqttMessage message) {
+        if (this.client.getQos2PendingIncomingPublishes().containsKey(((MqttMessageIdVariableHeader) message.variableHeader()).messageId())) {
+            com.shuhe.mqtt.MqttIncomingQos2Publish incomingQos2Publish = this.client.getQos2PendingIncomingPublishes().get(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
+            this.invokeHandlersForIncomingPublish(incomingQos2Publish.getIncomingPublish());
+            this.client.getQos2PendingIncomingPublishes().remove(incomingQos2Publish.getIncomingPublish().variableHeader().packetId());
+        }
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
+        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
+        channel.writeAndFlush(new MqttMessage(fixedHeader, variableHeader));
+    }
+
+    private void handlePubcomp(MqttMessage message) {
+        MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
+        com.shuhe.mqtt.MqttPendingPublish pendingPublish = this.client.getPendingPublishes().get(variableHeader.messageId());
+        pendingPublish.getFuture().setSuccess(null);
+        this.client.getPendingPublishes().remove(variableHeader.messageId());
+        pendingPublish.getPayload().release();
+        pendingPublish.onPubcompReceived();
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttClient.java

@@ -0,0 +1,199 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.util.concurrent.Future;
+
+public interface MqttClient {
+
+    /**
+     * Connect to the specified hostname/ip. By default uses port 1883.
+     * If you want to change the port number, see {@link #connect(String, int)}
+     *
+     * @param host The ip address or host to connect to
+     * @return A future which will be completed when the connection is opened and we received an CONNACK
+     */
+    Future<com.shuhe.mqtt.MqttConnectResult> connect(String host);
+
+    /**
+     * Connect to the specified hostname/ip using the specified port
+     *
+     * @param host The ip address or host to connect to
+     * @param port The tcp port to connect to
+     * @return A future which will be completed when the connection is opened and we received an CONNACK
+     */
+    Future<com.shuhe.mqtt.MqttConnectResult> connect(String host, int port);
+
+    /**
+     *
+     * @return boolean value indicating if channel is active
+     */
+    boolean isConnected();
+
+    /**
+     * Attempt reconnect to the host that was attempted with {@link #connect(String, int)} method before
+     *
+     * @return A future which will be completed when the connection is opened and we received an CONNACK
+     * @throws IllegalStateException if no previous {@link #connect(String, int)} calls were attempted
+     */
+    Future<com.shuhe.mqtt.MqttConnectResult> reconnect();
+
+    /**
+     * Retrieve the netty {@link EventLoopGroup} we are using
+     * @return The netty {@link EventLoopGroup} we use for the connection
+     */
+    EventLoopGroup getEventLoop();
+
+    /**
+     * By default we use the netty {@link NioEventLoopGroup}.
+     * If you change the EventLoopGroup to another type, make sure to change the {@link Channel} class using {@link MqttClientConfig#setChannelClass(Class)}
+     * If you want to force the MqttClient to use another {@link EventLoopGroup}, call this function before calling {@link #connect(String, int)}
+     *
+     * @param eventLoop The new eventloop to use
+     */
+    void setEventLoop(EventLoopGroup eventLoop);
+
+    /**
+     * Subscribe on the given topic. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     *
+     * @param topic The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    Future<Void> on(String topic, MqttHandler handler);
+
+    /**
+     * Subscribe on the given topic, with the given qos. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     *
+     * @param topic The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @param qos The qos to request to the server
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    Future<Void> on(String topic, MqttHandler handler, MqttQoS qos);
+
+    /**
+     * Subscribe on the given topic. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     * This subscription is only once. If the MqttClient has received 1 message, the subscription will be removed
+     *
+     * @param topic The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    Future<Void> once(String topic, MqttHandler handler);
+
+    /**
+     * Subscribe on the given topic, with the given qos. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     * This subscription is only once. If the MqttClient has received 1 message, the subscription will be removed
+     *
+     * @param topic The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @param qos The qos to request to the server
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    Future<Void> once(String topic, MqttHandler handler, MqttQoS qos);
+
+    /**
+     * Remove the subscription for the given topic and handler
+     * If you want to unsubscribe from all handlers known for this topic, use {@link #off(String)}
+     *
+     * @param topic The topic to unsubscribe for
+     * @param handler The handler to unsubscribe
+     * @return A future which will be completed when the server acknowledges our unsubscribe request
+     */
+    Future<Void> off(String topic, MqttHandler handler);
+
+    /**
+     * Remove all subscriptions for the given topic.
+     * If you want to specify which handler to unsubscribe, use {@link #off(String, MqttHandler)}
+     *
+     * @param topic The topic to unsubscribe for
+     * @return A future which will be completed when the server acknowledges our unsubscribe request
+     */
+    Future<Void> off(String topic);
+
+    /**
+     * Publish a message to the given payload
+     * @param topic The topic to publish to
+     * @param payload The payload to send
+     * @return A future which will be completed when the message is sent out of the MqttClient
+     */
+    Future<Void> publish(String topic, ByteBuf payload);
+
+    /**
+     * Publish a message to the given payload, using the given qos
+     * @param topic The topic to publish to
+     * @param payload The payload to send
+     * @param qos The qos to use while publishing
+     * @return A future which will be completed when the message is delivered to the server
+     */
+    Future<Void> publish(String topic, ByteBuf payload, MqttQoS qos);
+
+    /**
+     * Publish a message to the given payload, using optional retain
+     * @param topic The topic to publish to
+     * @param payload The payload to send
+     * @param retain true if you want to retain the message on the server, false otherwise
+     * @return A future which will be completed when the message is sent out of the MqttClient
+     */
+    Future<Void> publish(String topic, ByteBuf payload, boolean retain);
+
+    /**
+     * Publish a message to the given payload, using the given qos and optional retain
+     * @param topic The topic to publish to
+     * @param payload The payload to send
+     * @param qos The qos to use while publishing
+     * @param retain true if you want to retain the message on the server, false otherwise
+     * @return A future which will be completed when the message is delivered to the server
+     */
+    Future<Void> publish(String topic, ByteBuf payload, MqttQoS qos, boolean retain);
+
+    /**
+     * Retrieve the MqttClient configuration
+     * @return The {@link MqttClientConfig} instance we use
+     */
+    MqttClientConfig getClientConfig();
+
+
+    /**
+     * Construct the MqttClientImpl with additional config.
+     * This config can also be changed using the {@link #getClientConfig()} function
+     *
+     * @param config The config object to use while looking for settings
+     * @param defaultHandler The handler for incoming messages that do not match any topic subscriptions
+     */
+    static MqttClient create(MqttClientConfig config, MqttHandler defaultHandler){
+        return new MqttClientImpl(config, defaultHandler);
+    }
+
+    /**
+     * Send disconnect and close channel
+     *
+     */
+    void disconnect();
+
+    /**
+     * Sets the {@see #MqttClientCallback} object for this MqttClient
+     * @param callback The callback to be set
+     */
+    void setCallback(com.shuhe.mqtt.MqttClientCallback callback);
+
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttClientCallback.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+/**
+ * Created by Valerii Sosliuk on 12/30/2017.
+ */
+public interface MqttClientCallback {
+
+    /**
+     * This method is called when the connection to the server is lost.
+     *
+     * @param cause the reason behind the loss of connection.
+     */
+    void connectionLost(Throwable cause);
+
+    /**
+     * This method is called when the connection to the server is recovered.
+     *
+     */
+    void onSuccessfulReconnect();
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttClientConfig.java

@@ -0,0 +1,185 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.Channel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.mqtt.MqttVersion;
+import io.netty.handler.ssl.SslContext;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.Random;
+
+@SuppressWarnings({"WeakerAccess", "unused"})
+public final class MqttClientConfig {
+
+    private final SslContext sslContext;
+    private final String randomClientId;
+
+    private String clientId;
+    private int timeoutSeconds = 60;
+    private MqttVersion protocolVersion = MqttVersion.MQTT_3_1;
+    @Nullable private String username = null;
+    @Nullable private String password = null;
+    private boolean cleanSession = true;
+    @Nullable private com.shuhe.mqtt.MqttLastWill lastWill;
+    private Class<? extends Channel> channelClass = NioSocketChannel.class;
+
+    private boolean reconnect = true;
+    private long reconnectDelay = 1L;
+    private int maxBytesInMessage = 8092;
+
+    public MqttClientConfig() {
+        this(null);
+    }
+
+    public MqttClientConfig(SslContext sslContext) {
+        this.sslContext = sslContext;
+        Random random = new Random();
+        String id = "netty-mqtt/";
+        String[] options = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789".split("");
+        for(int i = 0; i < 8; i++){
+            id += options[random.nextInt(options.length)];
+        }
+        this.clientId = id;
+        this.randomClientId = id;
+    }
+
+    @Nonnull
+    public String getClientId() {
+        return clientId;
+    }
+
+    public void setClientId(@Nullable String clientId) {
+        if(clientId == null){
+            this.clientId = randomClientId;
+        }else{
+            this.clientId = clientId;
+        }
+    }
+
+    public int getTimeoutSeconds() {
+        return timeoutSeconds;
+    }
+
+    public void setTimeoutSeconds(int timeoutSeconds) {
+        if(timeoutSeconds != -1 && timeoutSeconds <= 0){
+            throw new IllegalArgumentException("timeoutSeconds must be > 0 or -1");
+        }
+        this.timeoutSeconds = timeoutSeconds;
+    }
+
+    public MqttVersion getProtocolVersion() {
+        return protocolVersion;
+    }
+
+    public void setProtocolVersion(MqttVersion protocolVersion) {
+        if(protocolVersion == null){
+            throw new NullPointerException("protocolVersion");
+        }
+        this.protocolVersion = protocolVersion;
+    }
+
+    @Nullable
+    public String getUsername() {
+        return username;
+    }
+
+    public void setUsername(@Nullable String username) {
+        this.username = username;
+    }
+
+    @Nullable
+    public String getPassword() {
+        return password;
+    }
+
+    public void setPassword(@Nullable String password) {
+        this.password = password;
+    }
+
+    public boolean isCleanSession() {
+        return cleanSession;
+    }
+
+    public void setCleanSession(boolean cleanSession) {
+        this.cleanSession = cleanSession;
+    }
+
+    @Nullable
+    public com.shuhe.mqtt.MqttLastWill getLastWill() {
+        return lastWill;
+    }
+
+    public void setLastWill(@Nullable com.shuhe.mqtt.MqttLastWill lastWill) {
+        this.lastWill = lastWill;
+    }
+
+    public Class<? extends Channel> getChannelClass() {
+        return channelClass;
+    }
+
+    public void setChannelClass(Class<? extends Channel> channelClass) {
+        this.channelClass = channelClass;
+    }
+
+    public SslContext getSslContext() {
+        return sslContext;
+    }
+
+    public boolean isReconnect() {
+        return reconnect;
+    }
+
+    public void setReconnect(boolean reconnect) {
+        this.reconnect = reconnect;
+    }
+
+    public long getReconnectDelay() {
+        return reconnectDelay;
+    }
+
+    /**
+     * Sets the reconnect delay in seconds. Defaults to 1 second.
+     * @param reconnectDelay
+     * @throws IllegalArgumentException if reconnectDelay is smaller than 1.
+     */
+    public void setReconnectDelay(long reconnectDelay) {
+        if (reconnectDelay <= 0) {
+            throw new IllegalArgumentException("reconnectDelay must be > 0");
+        }
+        this.reconnectDelay = reconnectDelay;
+    }
+
+    public int getMaxBytesInMessage() {
+        return maxBytesInMessage;
+    }
+
+    /**
+     * Sets the maximum number of bytes in the message for the {@link io.netty.handler.codec.mqtt.MqttDecoder}.
+     * Default value is 8092 as specified by Netty. The absolute maximum size is 256MB as set by the MQTT spec.
+     *
+     * @param maxBytesInMessage
+     * @throws IllegalArgumentException if maxBytesInMessage is smaller than 1 or greater than 256_000_000.
+     */
+    public void setMaxBytesInMessage(int maxBytesInMessage) {
+        if (maxBytesInMessage <= 0 || maxBytesInMessage > 256_000_000) {
+            throw new IllegalArgumentException("maxBytesInMessage must be > 0 or < 256_000_000");
+        }
+        this.maxBytesInMessage = maxBytesInMessage;
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttClientImpl.java

@@ -0,0 +1,533 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.mqtt.*;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.collection.IntObjectHashMap;
+import io.netty.util.concurrent.DefaultPromise;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.Promise;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Represents an MqttClientImpl connected to a single MQTT  Will try to keep the connection going at all times
+ */
+@SuppressWarnings({"WeakerAccess", "unused"})
+final class MqttClientImpl implements com.shuhe.mqtt.MqttClient {
+
+    private final Set<String> serverSubscriptions = new HashSet<>();
+    private final IntObjectHashMap<MqttPendingUnsubscription> pendingServerUnsubscribes = new IntObjectHashMap<>();
+    private final IntObjectHashMap<com.shuhe.mqtt.MqttIncomingQos2Publish> qos2PendingIncomingPublishes = new IntObjectHashMap<>();
+    private final IntObjectHashMap<MqttPendingPublish> pendingPublishes = new IntObjectHashMap<>();
+    private final HashMultimap<String, MqttSubscription> subscriptions = HashMultimap.create();
+    private final IntObjectHashMap<com.shuhe.mqtt.MqttPendingSubscription> pendingSubscriptions = new IntObjectHashMap<>();
+    private final Set<String> pendingSubscribeTopics = new HashSet<>();
+    private final HashMultimap<MqttHandler, MqttSubscription> handlerToSubscribtion = HashMultimap.create();
+    private final AtomicInteger nextMessageId = new AtomicInteger(1);
+
+    private final MqttClientConfig clientConfig;
+
+    private final MqttHandler defaultHandler;
+
+    private EventLoopGroup eventLoop;
+
+    private volatile Channel channel;
+
+    private volatile boolean disconnected = false;
+    private volatile boolean reconnect = false;
+    private String host;
+    private int port;
+    private com.shuhe.mqtt.MqttClientCallback callback;
+
+
+    /**
+     * Construct the MqttClientImpl with default config
+     */
+    public MqttClientImpl(MqttHandler defaultHandler) {
+        this.clientConfig = new MqttClientConfig();
+        this.defaultHandler = defaultHandler;
+    }
+
+    /**
+     * Construct the MqttClientImpl with additional config.
+     * This config can also be changed using the {@link #getClientConfig()} function
+     *
+     * @param clientConfig The config object to use while looking for settings
+     */
+    public MqttClientImpl(MqttClientConfig clientConfig, MqttHandler defaultHandler) {
+        this.clientConfig = clientConfig;
+        this.defaultHandler = defaultHandler;
+    }
+
+    /**
+     * Connect to the specified hostname/ip. By default uses port 1883.
+     * If you want to change the port number, see {@link #connect(String, int)}
+     *
+     * @param host The ip address or host to connect to
+     * @return A future which will be completed when the connection is opened and we received an CONNACK
+     */
+    @Override
+    public Future<com.shuhe.mqtt.MqttConnectResult> connect(String host) {
+        return connect(host, 1883);
+    }
+
+    /**
+     * Connect to the specified hostname/ip using the specified port
+     *
+     * @param host The ip address or host to connect to
+     * @param port The tcp port to connect to
+     * @return A future which will be completed when the connection is opened and we received an CONNACK
+     */
+    @Override
+    public Future<com.shuhe.mqtt.MqttConnectResult> connect(String host, int port) {
+        return connect(host, port, false);
+    }
+
+    private Future<com.shuhe.mqtt.MqttConnectResult> connect(String host, int port, boolean reconnect) {
+        if (this.eventLoop == null) {
+            this.eventLoop = new NioEventLoopGroup();
+        }
+        this.host = host;
+        this.port = port;
+        Promise<com.shuhe.mqtt.MqttConnectResult> connectFuture = new DefaultPromise<>(this.eventLoop.next());
+        Bootstrap bootstrap = new Bootstrap();
+        bootstrap.group(this.eventLoop);
+        bootstrap.channel(clientConfig.getChannelClass());
+        bootstrap.remoteAddress(host, port);
+        bootstrap.handler(new MqttChannelInitializer(connectFuture, host, port, clientConfig.getSslContext()));
+        ChannelFuture future = bootstrap.connect();
+
+        future.addListener((ChannelFutureListener) f -> {
+            if (f.isSuccess()) {
+                MqttClientImpl.this.channel = f.channel();
+                MqttClientImpl.this.channel.closeFuture().addListener((ChannelFutureListener) channelFuture -> {
+                    if (isConnected()) {
+                        return;
+                    }
+                    ChannelClosedException e = new ChannelClosedException("Channel is closed!");
+                    if (callback != null) {
+                        callback.connectionLost(e);
+                    }
+                    pendingSubscriptions.clear();
+                    serverSubscriptions.clear();
+                    subscriptions.clear();
+                    pendingServerUnsubscribes.clear();
+                    qos2PendingIncomingPublishes.clear();
+                    pendingPublishes.clear();
+                    pendingSubscribeTopics.clear();
+                    handlerToSubscribtion.clear();
+                    scheduleConnectIfRequired(host, port, true);
+                });
+            } else {
+                scheduleConnectIfRequired(host, port, reconnect);
+            }
+        });
+        return connectFuture;
+    }
+
+    private void scheduleConnectIfRequired(String host, int port, boolean reconnect) {
+        if (clientConfig.isReconnect() && !disconnected) {
+            if (reconnect) {
+                this.reconnect = true;
+            }
+            eventLoop.schedule((Runnable) () -> connect(host, port, reconnect), clientConfig.getReconnectDelay(), TimeUnit.SECONDS);
+        }
+    }
+
+    @Override
+    public boolean isConnected() {
+        return !disconnected && channel != null && channel.isActive();
+    }
+
+    @Override
+    public Future<com.shuhe.mqtt.MqttConnectResult> reconnect() {
+        if (host == null) {
+            throw new IllegalStateException("Cannot reconnect. Call connect() first");
+        }
+        return connect(host, port);
+    }
+
+    /**
+     * Retrieve the netty {@link EventLoopGroup} we are using
+     *
+     * @return The netty {@link EventLoopGroup} we use for the connection
+     */
+    @Override
+    public EventLoopGroup getEventLoop() {
+        return eventLoop;
+    }
+
+    /**
+     * By default we use the netty {@link NioEventLoopGroup}.
+     * If you change the EventLoopGroup to another type, make sure to change the {@link Channel} class using {@link MqttClientConfig#setChannelClass(Class)}
+     * If you want to force the MqttClient to use another {@link EventLoopGroup}, call this function before calling {@link #connect(String, int)}
+     *
+     * @param eventLoop The new eventloop to use
+     */
+    @Override
+    public void setEventLoop(EventLoopGroup eventLoop) {
+        this.eventLoop = eventLoop;
+    }
+
+    /**
+     * Subscribe on the given topic. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     *
+     * @param topic   The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    @Override
+    public Future<Void> on(String topic, MqttHandler handler) {
+        return on(topic, handler, MqttQoS.AT_MOST_ONCE);
+    }
+
+    /**
+     * Subscribe on the given topic, with the given qos. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     *
+     * @param topic   The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @param qos     The qos to request to the server
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    @Override
+    public Future<Void> on(String topic, MqttHandler handler, MqttQoS qos) {
+        return createSubscription(topic, handler, false, qos);
+    }
+
+    /**
+     * Subscribe on the given topic. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     * This subscription is only once. If the MqttClient has received 1 message, the subscription will be removed
+     *
+     * @param topic   The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    @Override
+    public Future<Void> once(String topic, MqttHandler handler) {
+        return once(topic, handler, MqttQoS.AT_MOST_ONCE);
+    }
+
+    /**
+     * Subscribe on the given topic, with the given qos. When a message is received, MqttClient will invoke the {@link MqttHandler#onMessage(String, ByteBuf)} function of the given handler
+     * This subscription is only once. If the MqttClient has received 1 message, the subscription will be removed
+     *
+     * @param topic   The topic filter to subscribe to
+     * @param handler The handler to invoke when we receive a message
+     * @param qos     The qos to request to the server
+     * @return A future which will be completed when the server acknowledges our subscribe request
+     */
+    @Override
+    public Future<Void> once(String topic, MqttHandler handler, MqttQoS qos) {
+        return createSubscription(topic, handler, true, qos);
+    }
+
+    /**
+     * Remove the subscription for the given topic and handler
+     * If you want to unsubscribe from all handlers known for this topic, use {@link #off(String)}
+     *
+     * @param topic   The topic to unsubscribe for
+     * @param handler The handler to unsubscribe
+     * @return A future which will be completed when the server acknowledges our unsubscribe request
+     */
+    @Override
+    public Future<Void> off(String topic, MqttHandler handler) {
+        Promise<Void> future = new DefaultPromise<>(this.eventLoop.next());
+        for (MqttSubscription subscription : this.handlerToSubscribtion.get(handler)) {
+            this.subscriptions.remove(topic, subscription);
+        }
+        this.handlerToSubscribtion.removeAll(handler);
+        this.checkSubscribtions(topic, future);
+        return future;
+    }
+
+    /**
+     * Remove all subscriptions for the given topic.
+     * If you want to specify which handler to unsubscribe, use {@link #off(String, MqttHandler)}
+     *
+     * @param topic The topic to unsubscribe for
+     * @return A future which will be completed when the server acknowledges our unsubscribe request
+     */
+    @Override
+    public Future<Void> off(String topic) {
+        Promise<Void> future = new DefaultPromise<>(this.eventLoop.next());
+        ImmutableSet<MqttSubscription> subscriptions = ImmutableSet.copyOf(this.subscriptions.get(topic));
+        for (MqttSubscription subscription : subscriptions) {
+            for (MqttSubscription handSub : this.handlerToSubscribtion.get(subscription.getHandler())) {
+                this.subscriptions.remove(topic, handSub);
+            }
+            this.handlerToSubscribtion.remove(subscription.getHandler(), subscription);
+        }
+        this.checkSubscribtions(topic, future);
+        return future;
+    }
+
+    /**
+     * Publish a message to the given payload
+     *
+     * @param topic   The topic to publish to
+     * @param payload The payload to send
+     * @return A future which will be completed when the message is sent out of the MqttClient
+     */
+    @Override
+    public Future<Void> publish(String topic, ByteBuf payload) {
+        return publish(topic, payload, MqttQoS.AT_MOST_ONCE, false);
+    }
+
+    /**
+     * Publish a message to the given payload, using the given qos
+     *
+     * @param topic   The topic to publish to
+     * @param payload The payload to send
+     * @param qos     The qos to use while publishing
+     * @return A future which will be completed when the message is delivered to the server
+     */
+    @Override
+    public Future<Void> publish(String topic, ByteBuf payload, MqttQoS qos) {
+        return publish(topic, payload, qos, false);
+    }
+
+    /**
+     * Publish a message to the given payload, using optional retain
+     *
+     * @param topic   The topic to publish to
+     * @param payload The payload to send
+     * @param retain  true if you want to retain the message on the server, false otherwise
+     * @return A future which will be completed when the message is sent out of the MqttClient
+     */
+    @Override
+    public Future<Void> publish(String topic, ByteBuf payload, boolean retain) {
+        return publish(topic, payload, MqttQoS.AT_MOST_ONCE, retain);
+    }
+
+    /**
+     * Publish a message to the given payload, using the given qos and optional retain
+     *
+     * @param topic   The topic to publish to
+     * @param payload The payload to send
+     * @param qos     The qos to use while publishing
+     * @param retain  true if you want to retain the message on the server, false otherwise
+     * @return A future which will be completed when the message is delivered to the server
+     */
+    @Override
+    public Future<Void> publish(String topic, ByteBuf payload, MqttQoS qos, boolean retain) {
+        Promise<Void> future = new DefaultPromise<>(this.eventLoop.next());
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retain, 0);
+        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topic, getNewMessageId().messageId());
+        MqttPublishMessage message = new MqttPublishMessage(fixedHeader, variableHeader, payload);
+        MqttPendingPublish pendingPublish = new MqttPendingPublish(variableHeader.packetId(), future, payload.retain(), message, qos);
+        ChannelFuture channelFuture = this.sendAndFlushPacket(message);
+
+        if (channelFuture != null) {
+            pendingPublish.setSent(true);
+            if (channelFuture.cause() != null) {
+                future.setFailure(channelFuture.cause());
+                return future;
+            }
+        }
+        if (pendingPublish.isSent() && pendingPublish.getQos() == MqttQoS.AT_MOST_ONCE) {
+            pendingPublish.getFuture().setSuccess(null); //We don't get an ACK for QOS 0
+        } else if (pendingPublish.isSent()) {
+            this.pendingPublishes.put(pendingPublish.getMessageId(), pendingPublish);
+            pendingPublish.startPublishRetransmissionTimer(this.eventLoop.next(), this::sendAndFlushPacket);
+        }
+        return future;
+    }
+
+    /**
+     * Retrieve the MqttClient configuration
+     *
+     * @return The {@link MqttClientConfig} instance we use
+     */
+    @Override
+    public MqttClientConfig getClientConfig() {
+        return clientConfig;
+    }
+
+    @Override
+    public void disconnect() {
+        disconnected = true;
+        if (this.channel != null) {
+            MqttMessage message = new MqttMessage(new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0));
+            this.sendAndFlushPacket(message).addListener(future1 -> channel.close());
+        }
+    }
+
+    @Override
+    public void setCallback(com.shuhe.mqtt.MqttClientCallback callback) {
+        this.callback = callback;
+    }
+
+
+    ///////////////////////////////////////////// PRIVATE API /////////////////////////////////////////////
+
+    public boolean isReconnect() {
+        return reconnect;
+    }
+
+    public void onSuccessfulReconnect() {
+        if (callback != null) {
+            callback.onSuccessfulReconnect();
+        }
+    }
+
+
+    ChannelFuture sendAndFlushPacket(Object message) {
+        if (this.channel == null) {
+            return null;
+        }
+        if (this.channel.isActive()) {
+            return this.channel.writeAndFlush(message);
+        }
+        return this.channel.newFailedFuture(new ChannelClosedException("Channel is closed!"));
+    }
+
+    private MqttMessageIdVariableHeader getNewMessageId() {
+        int messageId;
+        synchronized (this.nextMessageId) {
+            this.nextMessageId.compareAndSet(0xffff, 1);
+            messageId = this.nextMessageId.getAndIncrement();
+        }
+        return MqttMessageIdVariableHeader.from(messageId);
+    }
+
+    private Future<Void> createSubscription(String topic, MqttHandler handler, boolean once, MqttQoS qos) {
+        if (this.pendingSubscribeTopics.contains(topic)) {
+            Optional<Map.Entry<Integer, com.shuhe.mqtt.MqttPendingSubscription>> subscriptionEntry = this.pendingSubscriptions.entrySet().stream().filter((e) -> e.getValue().getTopic().equals(topic)).findAny();
+            if (subscriptionEntry.isPresent()) {
+                subscriptionEntry.get().getValue().addHandler(handler, once);
+                return subscriptionEntry.get().getValue().getFuture();
+            }
+        }
+        if (this.serverSubscriptions.contains(topic)) {
+            MqttSubscription subscription = new MqttSubscription(topic, handler, once);
+            this.subscriptions.put(topic, subscription);
+            this.handlerToSubscribtion.put(handler, subscription);
+            return this.channel.newSucceededFuture();
+        }
+
+        Promise<Void> future = new DefaultPromise<>(this.eventLoop.next());
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
+        MqttTopicSubscription subscription = new MqttTopicSubscription(topic, qos);
+        MqttMessageIdVariableHeader variableHeader = getNewMessageId();
+        MqttSubscribePayload payload = new MqttSubscribePayload(Collections.singletonList(subscription));
+        MqttSubscribeMessage message = new MqttSubscribeMessage(fixedHeader, variableHeader, payload);
+
+        final com.shuhe.mqtt.MqttPendingSubscription pendingSubscription = new com.shuhe.mqtt.MqttPendingSubscription(future, topic, message);
+        pendingSubscription.addHandler(handler, once);
+        this.pendingSubscriptions.put(variableHeader.messageId(), pendingSubscription);
+        this.pendingSubscribeTopics.add(topic);
+        pendingSubscription.setSent(this.sendAndFlushPacket(message) != null); //If not sent, we will send it when the connection is opened
+
+        pendingSubscription.startRetransmitTimer(this.eventLoop.next(), this::sendAndFlushPacket);
+
+        return future;
+    }
+
+    private void checkSubscribtions(String topic, Promise<Void> promise) {
+        if (!(this.subscriptions.containsKey(topic) && this.subscriptions.get(topic).size() != 0) && this.serverSubscriptions.contains(topic)) {
+            MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
+            MqttMessageIdVariableHeader variableHeader = getNewMessageId();
+            MqttUnsubscribePayload payload = new MqttUnsubscribePayload(Collections.singletonList(topic));
+            MqttUnsubscribeMessage message = new MqttUnsubscribeMessage(fixedHeader, variableHeader, payload);
+
+            MqttPendingUnsubscription pendingUnsubscription = new MqttPendingUnsubscription(promise, topic, message);
+            this.pendingServerUnsubscribes.put(variableHeader.messageId(), pendingUnsubscription);
+            pendingUnsubscription.startRetransmissionTimer(this.eventLoop.next(), this::sendAndFlushPacket);
+
+            this.sendAndFlushPacket(message);
+        } else {
+            promise.setSuccess(null);
+        }
+    }
+
+    IntObjectHashMap<com.shuhe.mqtt.MqttPendingSubscription> getPendingSubscriptions() {
+        return pendingSubscriptions;
+    }
+
+    HashMultimap<String, MqttSubscription> getSubscriptions() {
+        return subscriptions;
+    }
+
+    Set<String> getPendingSubscribeTopics() {
+        return pendingSubscribeTopics;
+    }
+
+    HashMultimap<MqttHandler, MqttSubscription> getHandlerToSubscribtion() {
+        return handlerToSubscribtion;
+    }
+
+    Set<String> getServerSubscriptions() {
+        return serverSubscriptions;
+    }
+
+    IntObjectHashMap<MqttPendingUnsubscription> getPendingServerUnsubscribes() {
+        return pendingServerUnsubscribes;
+    }
+
+    IntObjectHashMap<MqttPendingPublish> getPendingPublishes() {
+        return pendingPublishes;
+    }
+
+    IntObjectHashMap<com.shuhe.mqtt.MqttIncomingQos2Publish> getQos2PendingIncomingPublishes() {
+        return qos2PendingIncomingPublishes;
+    }
+
+    private class MqttChannelInitializer extends ChannelInitializer<SocketChannel> {
+
+        private final Promise<com.shuhe.mqtt.MqttConnectResult> connectFuture;
+        private final String host;
+        private final int port;
+        private final SslContext sslContext;
+
+
+        public MqttChannelInitializer(Promise<com.shuhe.mqtt.MqttConnectResult> connectFuture, String host, int port, SslContext sslContext) {
+            this.connectFuture = connectFuture;
+            this.host = host;
+            this.port = port;
+            this.sslContext = sslContext;
+        }
+
+        @Override
+        protected void initChannel(SocketChannel ch) throws Exception {
+            if (sslContext != null) {
+                ch.pipeline().addLast(sslContext.newHandler(ch.alloc(), host, port));
+            }
+
+            ch.pipeline().addLast("mqttDecoder", new MqttDecoder(clientConfig.getMaxBytesInMessage()));
+            ch.pipeline().addLast("mqttEncoder", MqttEncoder.INSTANCE);
+            ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(MqttClientImpl.this.clientConfig.getTimeoutSeconds(), MqttClientImpl.this.clientConfig.getTimeoutSeconds(), 0));
+            ch.pipeline().addLast("mqttPingHandler", new MqttPingHandler(MqttClientImpl.this.clientConfig.getTimeoutSeconds()));
+            ch.pipeline().addLast("mqttHandler", new MqttChannelHandler(MqttClientImpl.this, connectFuture));
+        }
+    }
+
+    MqttHandler getDefaultHandler() {
+        return defaultHandler;
+    }
+
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttConnectResult.java

@@ -0,0 +1,45 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
+
+@SuppressWarnings({"WeakerAccess", "unused"})
+public final class MqttConnectResult {
+
+    private final boolean success;
+    private final MqttConnectReturnCode returnCode;
+    private final ChannelFuture closeFuture;
+
+    MqttConnectResult(boolean success, MqttConnectReturnCode returnCode, ChannelFuture closeFuture) {
+        this.success = success;
+        this.returnCode = returnCode;
+        this.closeFuture = closeFuture;
+    }
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+    public MqttConnectReturnCode getReturnCode() {
+        return returnCode;
+    }
+
+    public ChannelFuture getCloseFuture() {
+        return closeFuture;
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttHandler.java

@@ -0,0 +1,23 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.buffer.ByteBuf;
+
+public interface MqttHandler {
+
+    void onMessage(String topic, ByteBuf payload);
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttIncomingQos2Publish.java

@@ -0,0 +1,31 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+
+final class MqttIncomingQos2Publish {
+
+    private final MqttPublishMessage incomingPublish;
+
+    MqttIncomingQos2Publish(MqttPublishMessage incomingPublish) {
+        this.incomingPublish = incomingPublish;
+    }
+
+    MqttPublishMessage getIncomingPublish() {
+        return incomingPublish;
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttLastWill.java

@@ -0,0 +1,154 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.handler.codec.mqtt.MqttQoS;
+
+@SuppressWarnings({"WeakerAccess", "unused", "SimplifiableIfStatement", "StringBufferReplaceableByString"})
+public final class MqttLastWill {
+
+    private final String topic;
+    private final String message;
+    private final boolean retain;
+    private final MqttQoS qos;
+
+    public MqttLastWill(String topic, String message, boolean retain, MqttQoS qos) {
+        if(topic == null){
+            throw new NullPointerException("topic");
+        }
+        if(message == null){
+            throw new NullPointerException("message");
+        }
+        if(qos == null){
+            throw new NullPointerException("qos");
+        }
+        this.topic = topic;
+        this.message = message;
+        this.retain = retain;
+        this.qos = qos;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public boolean isRetain() {
+        return retain;
+    }
+
+    public MqttQoS getQos() {
+        return qos;
+    }
+
+    public static Builder builder(){
+        return new Builder();
+    }
+
+    public static final class Builder {
+
+        private String topic;
+        private String message;
+        private boolean retain;
+        private MqttQoS qos;
+
+        public String getTopic() {
+            return topic;
+        }
+
+        public Builder setTopic(String topic) {
+            if(topic == null){
+                throw new NullPointerException("topic");
+            }
+            this.topic = topic;
+            return this;
+        }
+
+        public String getMessage() {
+            return message;
+        }
+
+        public Builder setMessage(String message) {
+            if(message == null){
+                throw new NullPointerException("message");
+            }
+            this.message = message;
+            return this;
+        }
+
+        public boolean isRetain() {
+            return retain;
+        }
+
+        public Builder setRetain(boolean retain) {
+            this.retain = retain;
+            return this;
+        }
+
+        public MqttQoS getQos() {
+            return qos;
+        }
+
+        public Builder setQos(MqttQoS qos) {
+            if(qos == null){
+                throw new NullPointerException("qos");
+            }
+            this.qos = qos;
+            return this;
+        }
+
+        public MqttLastWill build(){
+            return new MqttLastWill(topic, message, retain, qos);
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        MqttLastWill that = (MqttLastWill) o;
+
+        if (retain != that.retain) return false;
+        if (!topic.equals(that.topic)) return false;
+        if (!message.equals(that.message)) return false;
+        return qos == that.qos;
+
+    }
+
+    @Override
+    public int hashCode() {
+        int result = topic.hashCode();
+        result = 31 * result + message.hashCode();
+        result = 31 * result + (retain ? 1 : 0);
+        result = 31 * result + qos.hashCode();
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        final StringBuilder sb = new StringBuilder("MqttLastWill{");
+        sb.append("topic='").append(topic).append('\'');
+        sb.append(", message='").append(message).append('\'');
+        sb.append(", retain=").append(retain);
+        sb.append(", qos=").append(qos.name());
+        sb.append('}');
+        return sb.toString();
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttPendingPublish.java

@@ -0,0 +1,101 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttPublishMessage;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.util.concurrent.Promise;
+
+import java.util.function.Consumer;
+
+final class MqttPendingPublish {
+
+    private final int messageId;
+    private final Promise<Void> future;
+    private final ByteBuf payload;
+    private final MqttPublishMessage message;
+    private final MqttQoS qos;
+
+    private final RetransmissionHandler<MqttPublishMessage> publishRetransmissionHandler = new RetransmissionHandler<>();
+    private final RetransmissionHandler<MqttMessage> pubrelRetransmissionHandler = new RetransmissionHandler<>();
+
+    private boolean sent = false;
+
+    MqttPendingPublish(int messageId, Promise<Void> future, ByteBuf payload, MqttPublishMessage message, MqttQoS qos) {
+        this.messageId = messageId;
+        this.future = future;
+        this.payload = payload;
+        this.message = message;
+        this.qos = qos;
+
+        this.publishRetransmissionHandler.setOriginalMessage(message);
+    }
+
+    int getMessageId() {
+        return messageId;
+    }
+
+    Promise<Void> getFuture() {
+        return future;
+    }
+
+    ByteBuf getPayload() {
+        return payload;
+    }
+
+    boolean isSent() {
+        return sent;
+    }
+
+    void setSent(boolean sent) {
+        this.sent = sent;
+    }
+
+    MqttPublishMessage getMessage() {
+        return message;
+    }
+
+    MqttQoS getQos() {
+        return qos;
+    }
+
+    void startPublishRetransmissionTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
+        this.publishRetransmissionHandler.setHandle(((fixedHeader, originalMessage) ->
+                sendPacket.accept(new MqttPublishMessage(fixedHeader, originalMessage.variableHeader(), this.payload.retain()))));
+        this.publishRetransmissionHandler.start(eventLoop);
+    }
+
+    void onPubackReceived() {
+        this.publishRetransmissionHandler.stop();
+    }
+
+    void setPubrelMessage(MqttMessage pubrelMessage) {
+        this.pubrelRetransmissionHandler.setOriginalMessage(pubrelMessage);
+    }
+
+    void startPubrelRetransmissionTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
+        this.pubrelRetransmissionHandler.setHandle((fixedHeader, originalMessage) ->
+                sendPacket.accept(new MqttMessage(fixedHeader, originalMessage.variableHeader())));
+        this.pubrelRetransmissionHandler.start(eventLoop);
+    }
+
+    void onPubcompReceived() {
+        this.pubrelRetransmissionHandler.stop();
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttPendingSubscription.java

@@ -0,0 +1,102 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
+import io.netty.util.concurrent.Promise;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Consumer;
+
+final class MqttPendingSubscription {
+
+    private final Promise<Void> future;
+    private final String topic;
+    private final Set<MqttPendingHandler> handlers = new HashSet<>();
+    private final MqttSubscribeMessage subscribeMessage;
+
+    private final RetransmissionHandler<MqttSubscribeMessage> retransmissionHandler = new RetransmissionHandler<>();
+
+    private boolean sent = false;
+
+    MqttPendingSubscription(Promise<Void> future, String topic, MqttSubscribeMessage message) {
+        this.future = future;
+        this.topic = topic;
+        this.subscribeMessage = message;
+
+        this.retransmissionHandler.setOriginalMessage(message);
+    }
+
+    Promise<Void> getFuture() {
+        return future;
+    }
+
+    String getTopic() {
+        return topic;
+    }
+
+    boolean isSent() {
+        return sent;
+    }
+
+    void setSent(boolean sent) {
+        this.sent = sent;
+    }
+
+    MqttSubscribeMessage getSubscribeMessage() {
+        return subscribeMessage;
+    }
+
+    void addHandler(MqttHandler handler, boolean once){
+        this.handlers.add(new MqttPendingHandler(handler, once));
+    }
+
+    Set<MqttPendingHandler> getHandlers() {
+        return handlers;
+    }
+
+    void startRetransmitTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
+        if(this.sent){ //If the packet is sent, we can start the retransmit timer
+            this.retransmissionHandler.setHandle((fixedHeader, originalMessage) ->
+                    sendPacket.accept(new MqttSubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload())));
+            this.retransmissionHandler.start(eventLoop);
+        }
+    }
+
+    void onSubackReceived(){
+        this.retransmissionHandler.stop();
+    }
+
+    final class MqttPendingHandler {
+        private final MqttHandler handler;
+        private final boolean once;
+
+        MqttPendingHandler(MqttHandler handler, boolean once) {
+            this.handler = handler;
+            this.once = once;
+        }
+
+        MqttHandler getHandler() {
+            return handler;
+        }
+
+        boolean isOnce() {
+            return once;
+        }
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttPendingUnsubscription.java

@@ -0,0 +1,55 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
+import io.netty.util.concurrent.Promise;
+
+import java.util.function.Consumer;
+
+final class MqttPendingUnsubscription {
+
+    private final Promise<Void> future;
+    private final String topic;
+
+    private final com.shuhe.mqtt.RetransmissionHandler<MqttUnsubscribeMessage> retransmissionHandler = new com.shuhe.mqtt.RetransmissionHandler<>();
+
+    MqttPendingUnsubscription(Promise<Void> future, String topic, MqttUnsubscribeMessage unsubscribeMessage) {
+        this.future = future;
+        this.topic = topic;
+
+        this.retransmissionHandler.setOriginalMessage(unsubscribeMessage);
+    }
+
+    Promise<Void> getFuture() {
+        return future;
+    }
+
+    String getTopic() {
+        return topic;
+    }
+
+    void startRetransmissionTimer(EventLoop eventLoop, Consumer<Object> sendPacket) {
+        this.retransmissionHandler.setHandle((fixedHeader, originalMessage) ->
+                sendPacket.accept(new MqttUnsubscribeMessage(fixedHeader, originalMessage.variableHeader(), originalMessage.payload())));
+        this.retransmissionHandler.start(eventLoop);
+    }
+
+    void onUnsubackReceived(){
+        this.retransmissionHandler.stop();
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttPingHandler.java

@@ -0,0 +1,98 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.handler.timeout.IdleStateEvent;
+import io.netty.util.ReferenceCountUtil;
+import io.netty.util.concurrent.ScheduledFuture;
+
+import java.util.concurrent.TimeUnit;
+
+final class MqttPingHandler extends ChannelInboundHandlerAdapter {
+
+    private final int keepaliveSeconds;
+
+    private ScheduledFuture<?> pingRespTimeout;
+
+    MqttPingHandler(int keepaliveSeconds) {
+        this.keepaliveSeconds = keepaliveSeconds;
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (!(msg instanceof MqttMessage)) {
+            ctx.fireChannelRead(msg);
+            return;
+        }
+        MqttMessage message = (MqttMessage) msg;
+        if(message.fixedHeader().messageType() == MqttMessageType.PINGREQ){
+            this.handlePingReq(ctx.channel());
+        } else if(message.fixedHeader().messageType() == MqttMessageType.PINGRESP){
+            this.handlePingResp();
+        }else{
+            ctx.fireChannelRead(ReferenceCountUtil.retain(msg));
+        }
+    }
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        super.userEventTriggered(ctx, evt);
+
+        if(evt instanceof IdleStateEvent){
+            IdleStateEvent event = (IdleStateEvent) evt;
+            switch(event.state()){
+                case READER_IDLE:
+                    break;
+                case WRITER_IDLE:
+                    this.sendPingReq(ctx.channel());
+                    break;
+            }
+        }
+    }
+
+    private void sendPingReq(Channel channel){
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
+        channel.writeAndFlush(new MqttMessage(fixedHeader));
+
+        if(this.pingRespTimeout != null){
+            this.pingRespTimeout = channel.eventLoop().schedule(() -> {
+                MqttFixedHeader fixedHeader2 = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
+                channel.writeAndFlush(new MqttMessage(fixedHeader2)).addListener(ChannelFutureListener.CLOSE);
+                //TODO: what do when the connection is closed ?
+            }, this.keepaliveSeconds, TimeUnit.SECONDS);
+        }
+    }
+
+    private void handlePingReq(Channel channel){
+        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
+        channel.writeAndFlush(new MqttMessage(fixedHeader));
+    }
+
+    private void handlePingResp(){
+        if(this.pingRespTimeout != null && !this.pingRespTimeout.isCancelled() && !this.pingRespTimeout.isDone()){
+            this.pingRespTimeout.cancel(true);
+            this.pingRespTimeout = null;
+        }
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/MqttSubscription.java

@@ -0,0 +1,84 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import java.util.regex.Pattern;
+
+final class MqttSubscription {
+
+    private final String topic;
+    private final Pattern topicRegex;
+    private final com.shuhe.mqtt.MqttHandler handler;
+
+    private final boolean once;
+
+    private boolean called;
+
+    MqttSubscription(String topic, com.shuhe.mqtt.MqttHandler handler, boolean once) {
+        if(topic == null){
+            throw new NullPointerException("topic");
+        }
+        if(handler == null){
+            throw new NullPointerException("handler");
+        }
+        this.topic = topic;
+        this.handler = handler;
+        this.once = once;
+        this.topicRegex = Pattern.compile(topic.replace("+", "[^/]+").replace("#", ".+") + "$");
+    }
+
+    String getTopic() {
+        return topic;
+    }
+
+    public com.shuhe.mqtt.MqttHandler getHandler() {
+        return handler;
+    }
+
+    boolean isOnce() {
+        return once;
+    }
+
+    boolean isCalled() {
+        return called;
+    }
+
+    boolean matches(String topic){
+        return this.topicRegex.matcher(topic).matches();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        MqttSubscription that = (MqttSubscription) o;
+
+        return once == that.once && topic.equals(that.topic) && handler.equals(that.handler);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = topic.hashCode();
+        result = 31 * result + handler.hashCode();
+        result = 31 * result + (once ? 1 : 0);
+        return result;
+    }
+
+    void setCalled(boolean called) {
+        this.called = called;
+    }
+}

netty-mqtt/src/main/java/com/shuhe/mqtt/RetransmissionHandler.java

@@ -0,0 +1,72 @@
+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed 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 com.shuhe.mqtt;
+
+import io.netty.channel.EventLoop;
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import io.netty.handler.codec.mqtt.MqttMessageType;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.netty.util.concurrent.ScheduledFuture;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+
+final class RetransmissionHandler<T extends MqttMessage> {
+
+    private ScheduledFuture<?> timer;
+    private int timeout = 10;
+    private BiConsumer<MqttFixedHeader, T> handler;
+    private T originalMessage;
+
+    void start(EventLoop eventLoop){
+        if(eventLoop == null){
+            throw new NullPointerException("eventLoop");
+        }
+        if(this.handler == null){
+            throw new NullPointerException("handler");
+        }
+        this.timeout = 10;
+        this.startTimer(eventLoop);
+    }
+
+    private void startTimer(EventLoop eventLoop){
+        this.timer = eventLoop.schedule(() -> {
+            this.timeout += 5;
+            boolean isDup = this.originalMessage.fixedHeader().isDup();
+            if(this.originalMessage.fixedHeader().messageType() == MqttMessageType.PUBLISH && this.originalMessage.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE){
+                isDup = true;
+            }
+            MqttFixedHeader fixedHeader = new MqttFixedHeader(this.originalMessage.fixedHeader().messageType(), isDup, this.originalMessage.fixedHeader().qosLevel(), this.originalMessage.fixedHeader().isRetain(), this.originalMessage.fixedHeader().remainingLength());
+            handler.accept(fixedHeader, originalMessage);
+            startTimer(eventLoop);
+        }, timeout, TimeUnit.SECONDS);
+    }
+
+    void stop(){
+        if(this.timer != null){
+            this.timer.cancel(true);
+        }
+    }
+
+    void setHandle(BiConsumer<MqttFixedHeader, T> runnable) {
+        this.handler = runnable;
+    }
+
+    void setOriginalMessage(T originalMessage) {
+        this.originalMessage = originalMessage;
+    }
+}

pom.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.shuhe</groupId>
+    <artifactId>commons</artifactId>
+    <packaging>pom</packaging>
+    <version>1.0-SNAPSHOT</version>
+    <modules>
+        <module>netty-mqtt</module>
+        <module>common</module>
+    </modules>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+        <netty.version>4.1.49.Final</netty.version>
+        <guava.version>28.2-jre</guava.version>
+        <lombok.version>1.16.18</lombok.version>
+        <slf4j.version>1.7.7</slf4j.version>
+        <logback.version>1.1.7</logback.version>
+        <j2mod.version>LATEST</j2mod.version>
+        <paho.client.version>1.2.0</paho.client.version>
+        <json-path.version>2.2.0</json-path.version>
+        <commons-lang3.version>3.4</commons-lang3.version>
+        <logback.version>1.1.7</logback.version>
+    </properties>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.1</version>
+                <configuration>
+                    <source>${java.version}</source>
+                    <target>${java.version}</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <repositories>
+        <repository>
+            <id>central</id>
+            <name>aliyun maven</name>
+            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
+            <layout>default</layout>
+            <!-- 是否开启发布版构件下载 -->
+            <releases>
+                <enabled>true</enabled>
+            </releases>
+            <!-- 是否开启快照版构件下载 -->
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+        </repository>
+    </repositories>
+
+</project>