YangWX/MQTTnet
1
0
Fork 0
mirror of https://github.com/dotnet/MQTTnet.git synced 2026-10-18 10:26:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor code.

Browse Source
This commit is contained in:
Christian Kratky
2020-11-11 21:11:49 +01:00
parent da5a172504
commit 38d3531db6
1 changed files with 21 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/MQTTnet/Server/MqttClientConnection.cs
+21 -24
View File
@@ -186,40 +186,24 @@ namespace MQTTnet.Server
if (packet is MqttPublishPacket publishPacket) if (packet is MqttPublishPacket publishPacket)
{ {
await HandleIncomingPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false); await HandleIncomingPublishPacketAsync(publishPacket, cancellationToken).ConfigureAwait(false);
continue;
} }
else if (packet is MqttPubRelPacket pubRelPacket)
if (packet is MqttPubRelPacket pubRelPacket)
{ {
var pubCompPacket = new MqttPubCompPacket await HandleIncomingPubRelPacketAsync(pubRelPacket, cancellationToken).ConfigureAwait(false);
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
};
await SendAsync(pubCompPacket, cancellationToken).ConfigureAwait(false);
continue;
} }
else if (packet is MqttSubscribePacket subscribePacket)
if (packet is MqttSubscribePacket subscribePacket)
{ {
await HandleIncomingSubscribePacketAsync(subscribePacket, cancellationToken).ConfigureAwait(false); await HandleIncomingSubscribePacketAsync(subscribePacket, cancellationToken).ConfigureAwait(false);
continue;
} }
else if (packet is MqttUnsubscribePacket unsubscribePacket)
if (packet is MqttUnsubscribePacket unsubscribePacket)
{ {
await HandleIncomingUnsubscribePacketAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false); await HandleIncomingUnsubscribePacketAsync(unsubscribePacket, cancellationToken).ConfigureAwait(false);
continue;
} }
else if (packet is MqttPingReqPacket)
if (packet is MqttPingReqPacket)
{ {
await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false); await SendAsync(MqttPingRespPacket.Instance, cancellationToken).ConfigureAwait(false);
continue;
} }
else if (packet is MqttDisconnectPacket)
if (packet is MqttDisconnectPacket)
{ {
Session.WillMessage = null; Session.WillMessage = null;
disconnectType = MqttClientDisconnectType.Clean; disconnectType = MqttClientDisconnectType.Clean;
@@ -227,8 +211,10 @@ namespace MQTTnet.Server
StopInternal(); StopInternal();
return; return;
} }
else
_packetDispatcher.Dispatch(packet); {
_packetDispatcher.Dispatch(packet);
}
} }
} }
catch (OperationCanceledException) catch (OperationCanceledException)
@@ -289,6 +275,17 @@ namespace MQTTnet.Server
} }
} }
Task HandleIncomingPubRelPacketAsync(MqttPubRelPacket pubRelPacket, CancellationToken cancellationToken)
{
var pubCompPacket = new MqttPubCompPacket
{
PacketIdentifier = pubRelPacket.PacketIdentifier,
ReasonCode = MqttPubCompReasonCode.Success
};
return SendAsync(pubCompPacket, cancellationToken);
}
async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken) async Task HandleIncomingSubscribePacketAsync(MqttSubscribePacket subscribePacket, CancellationToken cancellationToken)
{ {
var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket, ConnectPacket).ConfigureAwait(false); var subscribeResult = await Session.SubscriptionsManager.SubscribeAsync(subscribePacket, ConnectPacket).ConfigureAwait(false);