✨ 简化客户端使用。

README.md

@@ -11,10 +11,12 @@
 - 继续抽象，方便使用。
 - 实现 `mqtt-broker` 功能。
 
-## 参考
+## 参考vs借鉴
 - [netty codec-mqtt](https://github.com/netty/netty/tree/4.1/codec-mqtt)
+- [jmqtt](https://github.com/Cicizz/jmqtt)
 - [iot-mqtt-server](https://gitee.com/recallcode/iot-mqtt-server)
 - [moquette](https://github.com/moquette-io/moquette)
+- [netty-mqtt-client](https://github.com/jetlinks/netty-mqtt-client)
 
 ## 工具
 - [mqttx 优雅的跨平台 MQTT 5.0 客户端工具](https://mqttx.app/cn/)

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClient.java

@@ -16,27 +16,116 @@
 
 package net.dreamlu.iot.mqtt.core.client;
 
-import net.dreamlu.iot.mqtt.codec.MqttMessage;
+import net.dreamlu.iot.mqtt.codec.*;
+import net.dreamlu.iot.mqtt.core.common.MqttMessageListener;
 import org.tio.client.ClientChannelContext;
 import org.tio.client.TioClient;
 import org.tio.core.Tio;
 
+import java.nio.ByteBuffer;
+
 /**
  * mqtt 客户端
  *
  * @author L.cm
  */
-public class MqttClient {
+public final class MqttClient {
+	private final TioClient tioClient;
+	private final MqttClientCreator config;
+	private final ClientChannelContext context;
 
-	private MqttClientConfig clientConfig;
-	private MqttClientProcessor processor;
-	private TioClient tioClient;
-	private ClientChannelContext clientContext;
+	public static MqttClientCreator create() {
+		return new MqttClientCreator();
+	}
 
-	// TODO add subscribe
+	MqttClient(TioClient tioClient, MqttClientCreator config, ClientChannelContext context) {
+		this.tioClient = tioClient;
+		this.config = config;
+		this.context = context;
+	}
 
-	public void connect() {
+	/**
+	 * 订阅
+	 *
+	 * @param topicFilter topicFilter
+	 * @param listener    MqttMessageListener
+	 * @return MqttClient
+	 */
+	public MqttClient subscribe(String topicFilter, MqttMessageListener listener) {
+		return this;
+	}
 
+	/**
+	 * 订阅
+	 *
+	 * @param topicFilter topicFilter
+	 * @param qos         MqttQoS
+	 * @param listener    MqttMessageListener
+	 * @return MqttClient
+	 */
+	public MqttClient subscribe(String topicFilter, MqttQoS qos, MqttMessageListener listener) {
+		return this;
+	}
+
+	/**
+	 * 取消订阅
+	 *
+	 * @param topicFilter topicFilter
+	 * @return MqttClient
+	 */
+	public MqttClient unSubscribe(String topicFilter) {
+		return this;
+	}
+
+	/**
+	 * 发布消息
+	 *
+	 * @param topic   topic
+	 * @param payload 消息体
+	 * @return 是否发送成功
+	 */
+	public Boolean publish(String topic, ByteBuffer payload) {
+		return publish(topic, payload, MqttQoS.AT_MOST_ONCE);
+	}
+
+	/**
+	 * 发布消息
+	 *
+	 * @param topic   topic
+	 * @param payload 消息体
+	 * @param qos     MqttQoS
+	 * @return 是否发送成功
+	 */
+	public Boolean publish(String topic, ByteBuffer payload, MqttQoS qos) {
+		return publish(topic, payload, qos, false);
+	}
+
+	/**
+	 * 发布消息
+	 *
+	 * @param topic   topic
+	 * @param payload 消息体
+	 * @param retain  是否在服务器上保留消息
+	 * @return 是否发送成功
+	 */
+	public Boolean publish(String topic, ByteBuffer payload, boolean retain) {
+		return publish(topic, payload, MqttQoS.AT_MOST_ONCE, retain);
+	}
+
+	/**
+	 * 发布消息
+	 *
+	 * @param topic   topic
+	 * @param payload 消息体
+	 * @param qos     MqttQoS
+	 * @param retain  是否在服务器上保留消息
+	 * @return 是否发送成功
+	 */
+	public Boolean publish(String topic, ByteBuffer payload, MqttQoS qos, boolean retain) {
+		MqttPublishMessage message = (MqttPublishMessage) MqttMessageFactory.newMessage(
+			new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retain, 0),
+			new MqttPublishVariableHeader(topic, 0), payload);
+		return Tio.send(context, message);
 	}
 
 	/**
@@ -44,17 +133,16 @@ public class MqttClient {
 	 *
 	 * @throws Exception 异常
 	 */
-	public void reconnect() throws Exception {
-		checkState();
-		tioClient.reconnect(clientContext, clientConfig.getTimeout());
+	public MqttClient reconnect() throws Exception {
+		tioClient.reconnect(context, config.getTimeout());
+		return this;
 	}
 
 	/**
 	 * 断开 mqtt 连接
 	 */
 	public void disconnect() {
-		checkState();
-		Tio.send(clientContext, MqttMessage.DISCONNECT);
+		Tio.send(context, MqttMessage.DISCONNECT);
 	}
 
 	/**
@@ -63,19 +151,11 @@ public class MqttClient {
 	 * @return 是否停止成功
 	 */
 	public boolean stop() {
-		checkState();
 		return tioClient.stop();
 	}
 
-	public ClientChannelContext getClientContext() {
-		checkState();
-		return clientContext;
-	}
-
-	private void checkState() {
-		if (clientContext == null) {
-			throw new IllegalStateException("您需要先 connect。");
-		}
+	public ClientChannelContext getContext() {
+		return context;
 	}
 
 }

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientAioListener.java

@@ -31,10 +31,10 @@ import java.util.Objects;
  * @author L.cm
  */
 public class MqttClientAioListener extends DefaultClientAioListener {
-	private final MqttClientConfig clientConfig;
+	private final MqttClientCreator clientConfig;
 	private final MqttWillMessage willMessage;
 
-	public MqttClientAioListener(MqttClientConfig clientConfig) {
+	public MqttClientAioListener(MqttClientCreator clientConfig) {
 		this.clientConfig = Objects.requireNonNull(clientConfig);
 		this.willMessage = clientConfig.getWillMessage();
 	}

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientConfig.java → mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/client/MqttClientCreator.java

@@ -17,23 +17,33 @@
 package net.dreamlu.iot.mqtt.core.client;
 
 import net.dreamlu.iot.mqtt.codec.MqttVersion;
+import org.tio.client.ClientChannelContext;
+import org.tio.client.ClientTioConfig;
+import org.tio.client.ReconnConf;
+import org.tio.client.TioClient;
+import org.tio.client.intf.ClientAioHandler;
+import org.tio.client.intf.ClientAioListener;
+import org.tio.core.Node;
 import org.tio.core.ssl.SslConfig;
 
+import java.util.Objects;
+import java.util.function.Consumer;
+
 /**
- * MqttClient 配置
+ * mqtt 客户端构造器
  *
  * @author L.cm
  */
-public class MqttClientConfig {
+public final class MqttClientCreator {
 
 	/**
 	 * ip，可为空，为空 t-io 默认为 127.0.0.1
 	 */
 	private String ip;
 	/**
-	 * 端口
+	 * 端口，默认：1883
 	 */
-	private int port;
+	private int port = 1883;
 	/**
 	 * 超时时间，t-io 配置，可为 null
 	 */
@@ -82,108 +92,160 @@ public class MqttClientConfig {
 	 * 遗嘱消息
 	 */
 	private MqttWillMessage willMessage;
+	/**
+	 * 客户端处理器
+	 */
+	private MqttClientProcessor processor;
 
-	public String getIp() {
+	protected String getIp() {
 		return ip;
 	}
 
-	public void setIp(String ip) {
-		this.ip = ip;
+	protected int getPort() {
+		return port;
 	}
 
-	public int getPort() {
-		return port;
+	protected Integer getTimeout() {
+		return timeout;
 	}
 
-	public void setPort(int port) {
-		this.port = port;
+	protected int getKeepAliveSecs() {
+		return keepAliveSecs;
 	}
 
-	public Integer getTimeout() {
-		return timeout;
+	protected SslConfig getSslConfig() {
+		return sslConfig;
 	}
 
-	public void setTimeout(Integer timeout) {
-		this.timeout = timeout;
+	protected boolean isReconnect() {
+		return reconnect;
 	}
 
-	public Integer getKeepAliveSecs() {
-		return keepAliveSecs;
+	protected Long getReInterval() {
+		return reInterval;
 	}
 
-	public void setKeepAliveSecs(Integer keepAliveSecs) {
-		this.keepAliveSecs = keepAliveSecs;
+	public String getClientId() {
+		return clientId;
 	}
 
-	public SslConfig getSslConfig() {
-		return sslConfig;
+	protected MqttVersion getProtocolVersion() {
+		return protocolVersion;
 	}
 
-	public void setSslConfig(SslConfig sslConfig) {
-		this.sslConfig = sslConfig;
+	protected String getUsername() {
+		return username;
 	}
 
-	public boolean isReconnect() {
-		return reconnect;
+	protected String getPassword() {
+		return password;
 	}
 
-	public void setReconnect(boolean reconnect) {
-		this.reconnect = reconnect;
+	protected boolean isCleanSession() {
+		return cleanSession;
 	}
 
-	public Long getReInterval() {
-		return reInterval;
+	protected MqttWillMessage getWillMessage() {
+		return willMessage;
 	}
 
-	public void setReInterval(Long reInterval) {
-		this.reInterval = reInterval;
+	protected MqttClientProcessor getProcessor() {
+		return processor;
 	}
 
-	public String getClientId() {
-		return clientId;
+	public MqttClientCreator ip(String ip) {
+		this.ip = ip;
+		return this;
 	}
 
-	public void setClientId(String clientId) {
-		this.clientId = clientId;
+	public MqttClientCreator port(int port) {
+		this.port = port;
+		return this;
 	}
 
-	public MqttVersion getProtocolVersion() {
-		return protocolVersion;
+	public MqttClientCreator timeout(int timeout) {
+		this.timeout = timeout;
+		return this;
 	}
 
-	public void setProtocolVersion(MqttVersion protocolVersion) {
-		this.protocolVersion = protocolVersion;
+	public MqttClientCreator keepAliveSecs(int keepAliveSecs) {
+		this.keepAliveSecs = keepAliveSecs;
+		return this;
 	}
 
-	public String getUsername() {
-		return username;
+	public MqttClientCreator sslConfig(SslConfig sslConfig) {
+		this.sslConfig = sslConfig;
+		return this;
 	}
 
-	public void setUsername(String username) {
-		this.username = username;
+	public MqttClientCreator reconnect(boolean reconnect) {
+		this.reconnect = reconnect;
+		return this;
 	}
 
-	public String getPassword() {
-		return password;
+	public MqttClientCreator reInterval(long reInterval) {
+		this.reInterval = reInterval;
+		return this;
 	}
 
-	public void setPassword(String password) {
-		this.password = password;
+	public MqttClientCreator clientId(String clientId) {
+		this.clientId = clientId;
+		return this;
 	}
 
-	public boolean isCleanSession() {
-		return cleanSession;
+	public MqttClientCreator protocolVersion(MqttVersion protocolVersion) {
+		this.protocolVersion = protocolVersion;
+		return this;
 	}
 
-	public void setCleanSession(boolean cleanSession) {
-		this.cleanSession = cleanSession;
+	public MqttClientCreator username(String username) {
+		this.username = username;
+		return this;
 	}
 
-	public MqttWillMessage getWillMessage() {
-		return willMessage;
+	public MqttClientCreator password(String password) {
+		this.password = password;
+		return this;
+	}
+
+	public MqttClientCreator cleanSession(boolean cleanSession) {
+		this.cleanSession = cleanSession;
+		return this;
 	}
 
-	public void setWillMessage(MqttWillMessage willMessage) {
+	public MqttClientCreator willMessage(MqttWillMessage willMessage) {
 		this.willMessage = willMessage;
+		return this;
 	}
+
+	public MqttClientCreator willMessage(Consumer<MqttWillMessage.Builder> consumer) {
+		MqttWillMessage.Builder builder = MqttWillMessage.builder();
+		consumer.accept(builder);
+		return willMessage(builder.build());
+	}
+
+	public MqttClientCreator processor(MqttClientProcessor processor) {
+		this.processor = Objects.requireNonNull(processor);
+		return this;
+	}
+
+	public MqttClient connect() throws Exception {
+		// 1. 初始化 mqtt 处理器
+		ClientAioHandler clientAioHandler = new MqttClientAioHandler(Objects.requireNonNull(this.processor));
+		ClientAioListener clientAioListener = new MqttClientAioListener(this);
+		// 2. 重连
+		ReconnConf reconnConf = null;
+		if (this.reconnect) {
+			if (reInterval != null && reInterval > 0) {
+				reconnConf = new ReconnConf(reInterval);
+			} else {
+				reconnConf = new ReconnConf();
+			}
+		}
+		// tioClient
+		TioClient tioClient = new TioClient(new ClientTioConfig(clientAioHandler, clientAioListener, reconnConf));
+		ClientChannelContext context = tioClient.connect(new Node(this.ip, this.port), this.timeout);
+		return new MqttClient(tioClient, this, context);
+	}
+
 }

mica-mqtt-core/src/main/java/net/dreamlu/iot/mqtt/core/common/MqttMessageListener.java

+/*
+ * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & www.net.dreamlu.net).
+ *
+ * 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 net.dreamlu.iot.mqtt.core.common;
+
+import java.nio.ByteBuffer;
+
+/**
+ * mqtt 消息处理
+ *
+ * @author L.cm
+ */
+@FunctionalInterface
+public interface MqttMessageListener {
+
+	/**
+	 * 监听到消息
+	 *
+	 * @param topic   topic
+	 * @param payload payload
+	 */
+	void onMessage(String topic, ByteBuffer payload);
+
+}

mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/client/MqttClientTest.java

 package net.dreamlu.iot.mqtt.client;
 
-import net.dreamlu.iot.mqtt.codec.*;
-import net.dreamlu.iot.mqtt.core.client.MqttClientAioHandler;
-import net.dreamlu.iot.mqtt.core.client.MqttClientAioListener;
-import net.dreamlu.iot.mqtt.core.client.MqttClientConfig;
-import net.dreamlu.iot.mqtt.core.client.MqttClientProcessor;
-import org.tio.client.ClientChannelContext;
-import org.tio.client.ClientTioConfig;
-import org.tio.client.ReconnConf;
-import org.tio.client.TioClient;
-import org.tio.client.intf.ClientAioHandler;
-import org.tio.client.intf.ClientAioListener;
-import org.tio.core.Node;
-import org.tio.core.Tio;
+import net.dreamlu.iot.mqtt.core.client.MqttClient;
 
 import java.nio.ByteBuffer;
 import java.util.Timer;
@@ -26,31 +14,19 @@ import java.util.TimerTask;
 public class MqttClientTest {
 
 	public static void main(String[] args) throws Exception {
-		MqttClientProcessor processor = new MqttClientProcessorImpl();
-		ClientAioHandler clientAioHandler = new MqttClientAioHandler(processor);
-		// 暂时用 set 后期抽成 builder
-		MqttClientConfig clientConfig = new MqttClientConfig();
-		clientConfig.setClientId("MqttClientTest");
-		clientConfig.setUsername("admin");
-		clientConfig.setPassword("123456");
-		//
-		ClientAioListener clientAioListener = new MqttClientAioListener(clientConfig);
-		ReconnConf reconnConf = new ReconnConf();
-		ClientTioConfig tioConfig = new ClientTioConfig(clientAioHandler, clientAioListener, reconnConf);
-
-		TioClient tioClient = new TioClient(tioConfig);
-		ClientChannelContext context = tioClient.connect(new Node("127.0.0.1", 1883), 1000);
+		// 初始化 mqtt 客户端
+		MqttClient client = MqttClient.create()
+			.clientId("MqttClientTest")
+			.username("admin")
+			.password("123456")
+			.processor(new MqttClientProcessorImpl())
+			.connect();
 
 		Timer timer = new Timer();
 		timer.schedule(new TimerTask() {
 			@Override
 			public void run() {
-				if (!context.isClosed) {
-					MqttPublishMessage message = (MqttPublishMessage) MqttMessageFactory.newMessage(
-						new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
-						new MqttPublishVariableHeader("testtopicxx", 0), ByteBuffer.wrap("mica最牛皮".getBytes()));
-					Tio.send(context, message);
-				}
+				client.publish("testtopicxx", ByteBuffer.wrap("mica最牛皮".getBytes()));
 			}
 		}, 1000, 2000);
 

mica-mqtt-example/src/main/java/net/dreamlu/iot/mqtt/server/MqttServerTest.java

@@ -54,6 +54,6 @@ public class MqttServerTest {
 		}, 1000, 2000);
 
 		// 启动
-		tioServer.start("127.0.0.1", socketPort);
+		tioServer.start("0.0.0.0", socketPort);
 	}
 }