117 lines
2.9 KiB
Dart
117 lines
2.9 KiB
Dart
import 'dart:convert';
|
|
import 'dart:typed_data';
|
|
|
|
import 'package:mqtt5_client/mqtt5_client.dart';
|
|
import 'package:mqtt5_client/mqtt5_server_client.dart';
|
|
|
|
class Mqtt {
|
|
final MqttConfig _config;
|
|
MqttClient? _client;
|
|
|
|
Mqtt(this._config);
|
|
|
|
static late Mqtt _mqtt;
|
|
|
|
static Mqtt getInstance() {
|
|
return _mqtt;
|
|
}
|
|
|
|
static void setInstance(Mqtt server) {
|
|
_mqtt = server;
|
|
}
|
|
|
|
Future<bool> connect() async {
|
|
if (_client != null) {
|
|
return true;
|
|
}
|
|
_client = MqttServerClient.withPort(
|
|
_config.host,
|
|
"${_config.clientId}_${DateTime.now().millisecondsSinceEpoch}",
|
|
_config.port);
|
|
_client?.autoReconnect = true;
|
|
|
|
MqttConnectionStatus? state;
|
|
try {
|
|
state = await _client?.connect(_config.username, _config.password);
|
|
} catch (e) {
|
|
//连接错误
|
|
print("mqtt连接失败: 地址:${_config.host} 原因:$e");
|
|
return false;
|
|
}
|
|
if (state!.state != MqttConnectionState.connected) {
|
|
//连接失败
|
|
print("mqtt连接失败: 地址:${_config.host} 原因:$state");
|
|
return false;
|
|
}
|
|
print("mqtt服务器: ${_config.host} 连接成功");
|
|
|
|
_client?.updates.listen((List<MqttReceivedMessage<MqttMessage>> message) {
|
|
final recMess = message[0].payload as MqttPublishMessage;
|
|
try {
|
|
if (_config.messgae != null) {
|
|
final payload = Utf8Decoder().convert(recMess.payload.message!);
|
|
// final payload =
|
|
// MqttUtilities.bytesToStringAsString(recMess.payload.message!);
|
|
_config.messgae!(message[0].topic!, payload);
|
|
} else if (_config.buffMessage != null) {
|
|
_config.buffMessage!(
|
|
message[0].topic!, recMess.payload.message!.buffer.asUint8List());
|
|
}
|
|
} catch (e) {
|
|
//转字符串失败
|
|
print(e);
|
|
}
|
|
});
|
|
_config.topic?.forEach((topic) {
|
|
subscribe(topic, _config.qos);
|
|
});
|
|
|
|
return true;
|
|
}
|
|
|
|
void disconnect() {
|
|
_client?.disconnect();
|
|
_client = null;
|
|
}
|
|
|
|
void subscribe(String topic, int qos) {
|
|
_client?.subscribe(topic, MqttUtilities.getQosLevel(qos));
|
|
}
|
|
|
|
void unSubscribe(String topic) {
|
|
_client?.unsubscribeStringTopic(topic);
|
|
}
|
|
|
|
void publish(String topic, String msg, {int qos = 0}) {
|
|
var payload = MqttPayloadBuilder();
|
|
payload.addUTF8String(msg);
|
|
_client?.publishMessage(
|
|
topic, MqttUtilities.getQosLevel(qos), payload.payload!);
|
|
}
|
|
}
|
|
|
|
class MqttConfig {
|
|
final String host;
|
|
final int port;
|
|
final String clientId;
|
|
String? username;
|
|
String? password;
|
|
List<String>? topic;
|
|
int qos;
|
|
//文本消息
|
|
Function(String topic, String message)? messgae;
|
|
//二进制消息
|
|
Function(String topic, Uint8List buff)? buffMessage;
|
|
|
|
MqttConfig(
|
|
{required this.host,
|
|
this.port = 1883,
|
|
required this.clientId,
|
|
this.messgae,
|
|
this.buffMessage,
|
|
this.topic,
|
|
this.qos = 0,
|
|
this.username,
|
|
this.password});
|
|
}
|