Skip to content

Commit

Permalink
mqtt离线缓存、通网续传
Browse files Browse the repository at this point in the history
  • Loading branch information
@iioter
iioter committed Jun 1, 2024
1 parent e717ef6 commit b88b2a5
Show file tree
Hide file tree
Showing 6 changed files with 50 additions and 45 deletions.
Binary file removed Export_DeviceSettings_本地.xlsx
View file
Binary file not shown.
3 changes: 1 addition & 2 deletions IoTGateway/Areas/BasicData/Controllers/DeviceController.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -259,8 +259,7 @@ public IActionResult ExportExcel(DeviceListVM vm)
var data = myExporter.Export();

string ContentType = "application/vnd.ms-excel";
string exportName = "DeviceSettings";
exportName = $"Export_{exportName}_{DateTime.Now.ToString("yyyyMMddHHmmssffff")}.xlsx";
string exportName = $"iotgateway.net_bakup_{DateTime.Now.ToString("yyyyMMddHHmmssffff")}.xlsx";
FileContentResult Result = new FileContentResult(data, ContentType);
Result.FileDownloadName = exportName;
return Result;
Expand Down
10 changes: 5 additions & 5 deletions IoTGateway/IoTGateway.csproj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@

<ItemGroup>
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.17.0" />
<PackageReference Include="MQTTnet" Version="4.3.1.873" />
<PackageReference Include="MQTTnet.AspNetCore" Version="4.3.1.873" />
<PackageReference Include="MQTTnet.Extensions.Rpc" Version="4.3.1.873" />
<PackageReference Include="NLog" Version="5.2.5" />
<PackageReference Include="NLog.Web.AspNetCore" Version="5.3.5" />
<PackageReference Include="MQTTnet" Version="4.3.6.1152" />
<PackageReference Include="MQTTnet.AspNetCore" Version="4.3.6.1152" />
<PackageReference Include="MQTTnet.Extensions.Rpc" Version="4.3.6.1152" />
<PackageReference Include="NLog" Version="5.3.2" />
<PackageReference Include="NLog.Web.AspNetCore" Version="5.3.11" />
<PackageReference Include="System.IO.Ports" Version="6.0.0" />
</ItemGroup>

Expand Down
79 changes: 42 additions & 37 deletions Plugins/Plugin/MyMqttClient.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
using PluginInterface.HuaWeiRoma;
using PluginInterface.ThingsBoard;
using Microsoft.Extensions.Logging;
using MQTTnet.Extensions.ManagedClient;

namespace Plugin
{
Expand All @@ -19,9 +20,9 @@ public class MyMqttClient
//private readonly ReferenceNodeManager? _uaNodeManager;

private SystemConfig _systemConfig;
private MqttClientOptions _options;
private ManagedMqttClientOptions _options;
public bool IsConnected => (Client.IsConnected);
private IMqttClient? Client { get; set; }
private IManagedMqttClient? Client { get; set; }
public event EventHandler<RpcRequest> OnExcRpc;
public event EventHandler<ISAttributeResponse> OnReceiveAttributes;
private readonly string _tbRpcTopic = "v1/gateway/rpc";
Expand All @@ -43,33 +44,37 @@ public async Task StartClientAsync()
{
Client.Dispose();
}
Client = new MqttFactory().CreateMqttClient();
Client = new MqttFactory().CreateManagedMqttClient();
await using var dc = new DataContext(IoTBackgroundService.connnectSetting, IoTBackgroundService.DbType);
_systemConfig = dc.Set<SystemConfig>().First();

#region ClientOptions
_options = new MqttClientOptionsBuilder()
.WithClientId(string.IsNullOrEmpty(_systemConfig.ClientId)
? Guid.NewGuid().ToString()
: _systemConfig.ClientId)
.WithTcpServer(_systemConfig.MqttIp, _systemConfig.MqttPort)
.WithCredentials(_systemConfig.MqttUName, _systemConfig.MqttUPwd)
.WithTimeout(TimeSpan.FromSeconds(30))
.WithKeepAlivePeriod(TimeSpan.FromSeconds(60))
.WithProtocolVersion(MqttProtocolVersion.V311)
.WithCleanSession(true)

_options = new ManagedMqttClientOptionsBuilder()
.WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
.WithMaxPendingMessages(100000)
.WithClientOptions(new MqttClientOptionsBuilder()
.WithClientId(string.IsNullOrEmpty(_systemConfig.ClientId)
? Guid.NewGuid().ToString()
: _systemConfig.ClientId)
.WithTcpServer(_systemConfig.MqttIp, _systemConfig.MqttPort)
.WithCredentials(_systemConfig.MqttUName, _systemConfig.MqttUPwd)
.WithTimeout(TimeSpan.FromSeconds(30))
.WithKeepAlivePeriod(TimeSpan.FromSeconds(60))
.WithProtocolVersion(MqttProtocolVersion.V311)
.WithCleanSession(true)
.Build())
.Build();
#endregion

Client.ConnectedAsync += Client_ConnectedAsync;
Client.DisconnectedAsync += Client_DisconnectedAsync;
Client.ApplicationMessageReceivedAsync += Client_ApplicationMessageReceivedAsync;

if(Client.ConnectAsync(_options).IsCompletedSuccessfully) ;
{
_logger.LogInformation("MQTT WAITING FOR APPLICATION MESSAGES");
}
await Client.StartAsync(_options);

_logger.LogInformation("MQTT WAITING FOR APPLICATION MESSAGES");


}
catch (Exception ex)
{
Expand Down Expand Up @@ -129,7 +134,7 @@ private async Task Client_DisconnectedAsync(MqttClientDisconnectedEventArgs arg)
try
{
_logger.LogError($"MQTT DISCONNECTED WITH SERVER ");
await Client.ConnectAsync(_options);
//await Client.ConnectAsync(_options);
}
catch (Exception ex)
{
Expand Down Expand Up @@ -259,7 +264,7 @@ private void ReceiveIsRpc(MqttApplicationMessageReceivedEventArgs e)

private async Task ResponseTbRpcAsync(TBRpcResponse tBRpcResponse)
{
await Client.PublishAsync(new MqttApplicationMessageBuilder()
await Client.EnqueueAsync(new MqttApplicationMessageBuilder()
.WithTopic(_tbRpcTopic)
.WithPayload(JsonConvert.SerializeObject(tBRpcResponse))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build());
Expand All @@ -268,7 +273,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder()
private async Task ResponseTcRpcAsync(TCRpcRequest tCRpcResponse)
{
var topic = $"command/reply/{tCRpcResponse.RequestData.RequestId}";
await Client.PublishAsync(new MqttApplicationMessageBuilder()
await Client.EnqueueAsync(new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(tCRpcResponse))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
Expand All @@ -278,7 +283,7 @@ private async Task ResponseIsRpcAsync(ISRpcResponse rpcResult)
{
//var responseTopic = $"/devices/{deviceid}/rpc/response/{methodName}/{rpcid}";
var topic = $"devices/{rpcResult.DeviceId}/rpc/response/{rpcResult.Method}/{rpcResult.ResponseId}";
await Client.PublishAsync(new MqttApplicationMessageBuilder()
await Client.EnqueueAsync(new MqttApplicationMessageBuilder()
.WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(rpcResult))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
Expand Down Expand Up @@ -316,7 +321,7 @@ public Task UploadAttributeAsync(string deviceName, object obj)
//Message: {"Device A":{"attribute1":"value1", "attribute2": 42}, "Device B":{"attribute1":"value1", "attribute2": 42}
try
{
return Client.PublishAsync(new MqttApplicationMessageBuilder()
return Client.EnqueueAsync(new MqttApplicationMessageBuilder()
.WithTopic($"devices/{deviceName}/attributes").WithPayload(JsonConvert.SerializeObject(obj))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
.Build());
Expand All @@ -331,14 +336,14 @@ public Task UploadAttributeAsync(string deviceName, object obj)

public async Task UploadIsTelemetryDataAsync(string deviceName, object obj)
{
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"devices/{deviceName}/telemetry")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"devices/{deviceName}/telemetry")
.WithPayload(JsonConvert.SerializeObject(obj)).Build());
}

public async Task UploadTcTelemetryDataAsync(string deviceName, object obj)
{
var toSend = new Dictionary<string, object> { { deviceName, obj } };
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"gateway/attributes")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"gateway/attributes")
.WithPayload(JsonConvert.SerializeObject(toSend)).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
}

Expand All @@ -365,7 +370,7 @@ public async Task UploadHwTelemetryDataAsync(Device device, object obj)
Devices = hwTelemetry
};

await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/datas")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/datas")
.WithPayload(JsonConvert.SerializeObject(hwTelemetrys)).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
}

Expand Down Expand Up @@ -433,7 +438,7 @@ public async Task RequestAttributes(string deviceName, bool anySide, params stri
{ "client", true },
{ "key", args[0] }
};
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gateway/attributes/request")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gateway/attributes/request")
.WithPayload(JsonConvert.SerializeObject(tbRequestData)).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
break;
case IoTPlatformType.IoTSharp:
Expand All @@ -443,7 +448,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gat
keys.Add(anySide ? "anySide" : "server", string.Join(",", args));
await Client.SubscribeAsync($"devices/{deviceName}/attributes/response/{id}",
MqttQualityOfServiceLevel.ExactlyOnce);
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(keys))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
break;
Expand Down Expand Up @@ -520,7 +525,7 @@ public async Task PublishTelemetryAsync(string deviceName, Device device, Dictio
switch (_systemConfig.IoTPlatformType)
{
case IoTPlatformType.ThingsBoard:
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("v1/gateway/telemetry")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("v1/gateway/telemetry")
.WithPayload(JsonConvert.SerializeObject(sendModel))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build());
break;
Expand Down Expand Up @@ -591,7 +596,7 @@ public async Task DeviceConnected(string DeviceName, Device device)
case IoTPlatformType.ThingsBoard:
case IoTPlatformType.IoTSharp:
case IoTPlatformType.IoTGateway:
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("v1/gateway/connect")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("v1/gateway/connect")
.WithPayload(JsonConvert.SerializeObject(new Dictionary<string, string>
{ { "device", DeviceName } }))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build());
Expand All @@ -605,7 +610,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("v1/gate
case IoTPlatformType.OneNET:
break;
case IoTPlatformType.ThingsCloud:
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("gateway/connect")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic("gateway/connect")
.WithPayload(JsonConvert.SerializeObject(new Dictionary<string, string>
{ { "device", DeviceName } }))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
Expand All @@ -625,7 +630,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic("gateway
}
}
};
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/topo/update")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/topo/update")
.WithPayload(JsonConvert.SerializeObject(deviceOnLine))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
break;
Expand All @@ -646,7 +651,7 @@ public async Task DeviceDisconnected(string DeviceName, Device device)
case IoTPlatformType.ThingsBoard:
case IoTPlatformType.IoTSharp:
case IoTPlatformType.IoTGateway:
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gateway/disconnect")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gateway/disconnect")
.WithPayload(JsonConvert.SerializeObject(new Dictionary<string, string>
{ { "device", DeviceName } }))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce).Build());
Expand All @@ -660,7 +665,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"v1/gat
case IoTPlatformType.OneNET:
break;
case IoTPlatformType.ThingsCloud:
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"gateway/disconnect")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"gateway/disconnect")
.WithPayload(JsonConvert.SerializeObject(new Dictionary<string, string>
{ { "device", DeviceName } }))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
Expand All @@ -680,7 +685,7 @@ await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"gatewa
}
}
};
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/topo/update")
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic($"/v1/devices/{_systemConfig.GatewayName}/topo/update")
.WithPayload(JsonConvert.SerializeObject(deviceOnLine))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
break;
Expand Down Expand Up @@ -715,7 +720,7 @@ public async Task DeviceAdded(Device device)
ProductType = "A_n"
}
);
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(addDeviceDto))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
break;
Expand Down Expand Up @@ -753,7 +758,7 @@ public async Task DeviceDeleted(Device device)
ProductType = "A_n"
}
};
await Client.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
await Client.EnqueueAsync(new MqttApplicationMessageBuilder().WithTopic(topic)
.WithPayload(JsonConvert.SerializeObject(deleteDeviceDto))
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce).Build());
}
Expand Down
3 changes: 2 additions & 1 deletion Plugins/Plugin/Plugin.csproj
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,8 @@
<ItemGroup>
<PackageReference Include="DynamicExpresso.Core" Version="2.16.1" />
<PackageReference Include="Mono.Options" Version="6.12.0.148" />
<PackageReference Include="MQTTnet" Version="4.3.1.873" />
<PackageReference Include="MQTTnet" Version="4.3.6.1152" />
<PackageReference Include="MQTTnet.Extensions.ManagedClient" Version="4.3.6.1152" />
<PackageReference Include="OPCFoundation.NetStandard.Opc.Ua.Server" Version="1.4.370.12" />
</ItemGroup>

Expand Down
Binary file added iotgateway.net_bakup_202406010912327657.xlsx
View file
Binary file not shown.

0 comments on commit b88b2a5

Please sign in to comment.