148 lines
4.7 KiB
C#
148 lines
4.7 KiB
C#
using Logger;
|
||
using MQTTnet;
|
||
using MQTTnet.Client;
|
||
using MQTTnet.Extensions.ManagedClient;
|
||
using MQTTnet.Server;
|
||
using System;
|
||
using System.Net.Sockets;
|
||
using System.Reflection;
|
||
using System.Text;
|
||
using System.Threading.Tasks;
|
||
|
||
namespace WebApi.Helpers
|
||
{
|
||
|
||
public class Mqtt
|
||
{
|
||
public static IMqttClient mqttClient = null;
|
||
|
||
public static event Action<bool> ConnectionStatusChanged; // 连接状态改变事件
|
||
/// <summary>
|
||
/// Param1：ClientId
|
||
/// Param2：Topic
|
||
/// Param3：Message
|
||
/// </summary>
|
||
public static event Action<string,string,string> MessageReceived; // 接收到消息事件
|
||
|
||
public static async Task RunMqttStart(string ip,int port,string username, string password)
|
||
{
|
||
try
|
||
{
|
||
string clientId = Guid.NewGuid().ToString();
|
||
// Create a MQTT client factory
|
||
var factory = new MqttFactory();
|
||
|
||
// Create a MQTT client instance
|
||
mqttClient = factory.CreateMqttClient();
|
||
|
||
// 连接成功
|
||
mqttClient.ConnectedAsync += (e =>
|
||
{
|
||
ConnectionStatusChanged?.Invoke(true);
|
||
return Task.CompletedTask;
|
||
});
|
||
// 连接断开
|
||
mqttClient.DisconnectedAsync += (e =>
|
||
{
|
||
ConnectionStatusChanged?.Invoke(false);
|
||
return Task.CompletedTask;
|
||
});
|
||
// 收到消息
|
||
mqttClient.ApplicationMessageReceivedAsync += (e =>
|
||
{
|
||
string messagePayload = null;
|
||
if (e.ApplicationMessage.PayloadSegment.Count > 0)
|
||
{
|
||
byte[] payload = e.ApplicationMessage.PayloadSegment.Array;
|
||
int payloadOffset = e.ApplicationMessage.PayloadSegment.Offset;
|
||
int payloadLength = e.ApplicationMessage.PayloadSegment.Count;
|
||
messagePayload = Encoding.UTF8.GetString(payload, payloadOffset, payloadLength);
|
||
}
|
||
else
|
||
{
|
||
messagePayload = "";
|
||
}
|
||
|
||
MessageReceived?.Invoke(e.ClientId, e.ApplicationMessage.Topic, messagePayload);
|
||
return Task.CompletedTask;
|
||
});
|
||
|
||
// Create MQTT client options
|
||
var options = new MqttClientOptionsBuilder()
|
||
.WithTcpServer(ip, port) // MQTT broker address and port
|
||
.WithCredentials(username, password) // Set username and password
|
||
.WithClientId(clientId)
|
||
.WithCleanSession()
|
||
.Build();
|
||
|
||
// Connect to MQTT broker
|
||
var connectResult = await mqttClient.ConnectAsync(options);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
Log.FunError(ex, MethodBase.GetCurrentMethod().Name);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 断开连接
|
||
/// </summary>
|
||
/// <returns></returns>
|
||
public static async Task RunMqttStop()
|
||
{
|
||
if (mqttClient != null)
|
||
{
|
||
await mqttClient.DisconnectAsync();
|
||
mqttClient.Dispose();
|
||
mqttClient = null;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 推送消息
|
||
/// </summary>
|
||
/// <param name="topic"></param>
|
||
/// <param name="message"></param>
|
||
/// <returns></returns>
|
||
public static async Task RunMqttPublish(string topic, string message)
|
||
{
|
||
if (mqttClient != null)
|
||
{
|
||
var publishOptions = new MqttApplicationMessageBuilder()
|
||
.WithTopic(topic)
|
||
.WithPayload(message)
|
||
.WithRetainFlag(false)
|
||
.Build();
|
||
|
||
await mqttClient.PublishAsync(publishOptions);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 添加订阅
|
||
/// </summary>
|
||
/// <param name="topic"></param>
|
||
/// <returns></returns>
|
||
public static async Task RunMqttSubscribe(string topic)
|
||
{
|
||
if (mqttClient != null)
|
||
{
|
||
await mqttClient.SubscribeAsync(topic);
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 取消订阅
|
||
/// </summary>
|
||
/// <param name="topic"></param>
|
||
/// <returns></returns>
|
||
public static async Task RunMqttUnSubscribe(string topic)
|
||
{
|
||
if (mqttClient != null)
|
||
{
|
||
await mqttClient.UnsubscribeAsync(topic);
|
||
}
|
||
}
|
||
}
|
||
}
|