64 lines
2.6 KiB
C#
64 lines
2.6 KiB
C#
using MQTTnet;
|
|
using MQTTnet.Client;
|
|
using MQTTnet.Formatter;
|
|
using MQTTnet.Protocol;
|
|
|
|
namespace Evergreen.Health.Application.Health.Integrations.Ib43;
|
|
|
|
public interface IIb43MqttTransport : IDisposable
|
|
{
|
|
bool IsConnected { get; }
|
|
event Func<string, string, bool, Task> Message;
|
|
event Action<string> OversizedMessage;
|
|
Task ConnectAndSubscribeAsync(Ib43Options options, CancellationToken cancellationToken);
|
|
}
|
|
|
|
public interface IIb43MqttTransportFactory
|
|
{
|
|
IIb43MqttTransport Create();
|
|
}
|
|
|
|
public sealed class Ib43MqttTransportFactory : IIb43MqttTransportFactory
|
|
{
|
|
public IIb43MqttTransport Create() => new Ib43MqttTransport();
|
|
}
|
|
|
|
public sealed class Ib43MqttTransport : IIb43MqttTransport
|
|
{
|
|
private readonly IMqttClient _client = new MqttFactory().CreateMqttClient();
|
|
public bool IsConnected => _client.IsConnected;
|
|
public event Func<string, string, bool, Task> Message;
|
|
public event Action<string> OversizedMessage;
|
|
|
|
public Ib43MqttTransport()
|
|
{
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
var m = e.ApplicationMessage;
|
|
// 在解码前限制长度；不要将大包全部复制进诊断缓存。
|
|
if (m.PayloadSegment.Count > Ib43StateStore.MaxPayloadBytes)
|
|
OversizedMessage?.Invoke(m.Topic);
|
|
else if (Message != null) await Message.Invoke(m.Topic, m.ConvertPayloadToString(), m.Retain);
|
|
};
|
|
}
|
|
|
|
public async Task ConnectAndSubscribeAsync(Ib43Options options, CancellationToken cancellationToken)
|
|
{
|
|
var builder = new MqttClientOptionsBuilder().WithTcpServer(options.Host, options.Port)
|
|
.WithClientId(options.ClientId).WithCredentials(options.Username, options.Password)
|
|
.WithProtocolVersion(MqttProtocolVersion.V311).WithCleanSession()
|
|
.WithKeepAlivePeriod(TimeSpan.FromSeconds(30)).WithTimeout(TimeSpan.FromSeconds(10));
|
|
if (options.UseTls) builder.WithTlsOptions(t => t.UseTls());
|
|
await _client.ConnectAsync(builder.Build(), cancellationToken);
|
|
var subscription = new MqttClientSubscribeOptionsBuilder();
|
|
foreach (var device in options.Devices)
|
|
subscription.WithTopicFilter(f => f.WithTopic(device.PostTopic)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce));
|
|
var result = await _client.SubscribeAsync(subscription.Build(), cancellationToken);
|
|
if (result.Items.Count != options.Devices.Count || result.Items.Any(i => (int)i.ResultCode >= 128))
|
|
throw new InvalidOperationException("IB43 subscription rejected");
|
|
}
|
|
|
|
public void Dispose() => _client.Dispose();
|
|
}
|