@@ -291,13 +291,13 @@ namespace MQTTnet.Client | |||||
private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) | private Task SendAsync(MqttBasePacket packet, CancellationToken cancellationToken) | ||||
{ | { | ||||
_sendTracker.Restart(); | _sendTracker.Restart(); | ||||
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, new[] { packet }); | |||||
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, new[] { packet }, cancellationToken); | |||||
} | } | ||||
private Task SendAsync(IEnumerable<MqttBasePacket> packets, CancellationToken cancellationToken) | private Task SendAsync(IEnumerable<MqttBasePacket> packets, CancellationToken cancellationToken) | ||||
{ | { | ||||
_sendTracker.Restart(); | _sendTracker.Restart(); | ||||
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, cancellationToken, packets); | |||||
return _adapter.SendPacketsAsync(_options.CommunicationTimeout, packets, cancellationToken); | |||||
} | } | ||||
private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket | private async Task<TResponsePacket> SendAndReceiveAsync<TResponsePacket>(MqttBasePacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttBasePacket | ||||
@@ -313,8 +313,8 @@ namespace MQTTnet.Client | |||||
var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier); | var packetAwaiter = _packetDispatcher.AddPacketAwaiter<TResponsePacket>(identifier); | ||||
try | try | ||||
{ | { | ||||
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, new[] { requestPacket }, _cancellationTokenSource.Token).ConfigureAwait(false); | |||||
var respone = await Internal.TaskExtensions.TimeoutAfter(ct => packetAwaiter.Task, _options.CommunicationTimeout, _cancellationTokenSource.Token).ConfigureAwait(false); | |||||
await _adapter.SendPacketsAsync(_options.CommunicationTimeout, new[] { requestPacket }, cancellationToken).ConfigureAwait(false); | |||||
var respone = await Internal.TaskExtensions.TimeoutAfter(ct => packetAwaiter.Task, _options.CommunicationTimeout, cancellationToken).ConfigureAwait(false); | |||||
return (TResponsePacket)respone; | return (TResponsePacket)respone; | ||||
} | } | ||||
@@ -13,12 +13,12 @@ namespace MQTTnet.Server | |||||
private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch(); | private readonly Stopwatch _lastNonKeepAlivePacketReceivedTracker = new Stopwatch(); | ||||
private readonly string _clientId; | private readonly string _clientId; | ||||
private readonly Func<Task> _timeoutCallback; | |||||
private readonly Action _timeoutCallback; | |||||
private readonly IMqttNetLogger _logger; | private readonly IMqttNetLogger _logger; | ||||
private Task _workerTask; | private Task _workerTask; | ||||
public MqttClientKeepAliveMonitor(string clientId, Func<Task> timeoutCallback, IMqttNetLogger logger) | |||||
public MqttClientKeepAliveMonitor(string clientId, Action timeoutCallback, IMqttNetLogger logger) | |||||
{ | { | ||||
_clientId = clientId; | _clientId = clientId; | ||||
_timeoutCallback = timeoutCallback; | _timeoutCallback = timeoutCallback; | ||||
@@ -61,10 +61,7 @@ namespace MQTTnet.Server | |||||
{ | { | ||||
_logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", _clientId); | _logger.Warning<MqttClientSession>("Client '{0}': Did not receive any packet or keep alive signal.", _clientId); | ||||
if (_timeoutCallback != null) | |||||
{ | |||||
await _timeoutCallback().ConfigureAwait(false); | |||||
} | |||||
_timeoutCallback?.Invoke(); | |||||
return; | return; | ||||
} | } | ||||
@@ -127,7 +127,7 @@ namespace MQTTnet.Server | |||||
if (!cancellationToken.IsCancellationRequested) | if (!cancellationToken.IsCancellationRequested) | ||||
{ | { | ||||
await _clientSession.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); | |||||
_clientSession.Stop(MqttClientDisconnectType.NotClean); | |||||
} | } | ||||
} | } | ||||
} | } | ||||
@@ -39,7 +39,7 @@ namespace MQTTnet.Server | |||||
ClientId = clientId; | ClientId = clientId; | ||||
KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeoutAsync, _logger); | |||||
KeepAliveMonitor = new MqttClientKeepAliveMonitor(clientId, StopDueToKeepAliveTimeout, _logger); | |||||
SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server); | SubscriptionsManager = new MqttClientSubscriptionsManager(clientId, _options, sessionsManager.Server); | ||||
PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger); | PendingMessagesQueue = new MqttClientPendingMessagesQueue(_options, this, _logger); | ||||
} | } | ||||
@@ -101,13 +101,13 @@ namespace MQTTnet.Server | |||||
return _wasCleanDisconnect; | return _wasCleanDisconnect; | ||||
} | } | ||||
public async Task StopAsync(MqttClientDisconnectType type) | |||||
public void Stop(MqttClientDisconnectType type) | |||||
{ | { | ||||
try | try | ||||
{ | { | ||||
if (_cancellationTokenSource == null) | if (_cancellationTokenSource == null) | ||||
{ | { | ||||
return Task.FromResult(0); | |||||
return; | |||||
} | } | ||||
_wasCleanDisconnect = type == MqttClientDisconnectType.Clean; | _wasCleanDisconnect = type == MqttClientDisconnectType.Clean; | ||||
@@ -128,8 +128,6 @@ namespace MQTTnet.Server | |||||
{ | { | ||||
_logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId); | _logger.Info<MqttClientSession>("Client '{0}': Session stopped.", ClientId); | ||||
} | } | ||||
return Task.FromResult(0); | |||||
} | } | ||||
public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage) | public async Task EnqueueApplicationMessageAsync(MqttApplicationMessage applicationMessage) | ||||
@@ -183,10 +181,10 @@ namespace MQTTnet.Server | |||||
_cancellationTokenSource?.Dispose(); | _cancellationTokenSource?.Dispose(); | ||||
} | } | ||||
private Task StopDueToKeepAliveTimeoutAsync() | |||||
private void StopDueToKeepAliveTimeout() | |||||
{ | { | ||||
_logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); | _logger.Info<MqttClientSession>("Client '{0}': Timeout while waiting for KeepAlive packet.", ClientId); | ||||
return StopAsync(MqttClientDisconnectType.NotClean); | |||||
Stop(MqttClientDisconnectType.NotClean); | |||||
} | } | ||||
private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) | private async Task ReceivePacketsAsync(IMqttChannelAdapter adapter, CancellationToken cancellationToken) | ||||
@@ -214,7 +212,7 @@ namespace MQTTnet.Server | |||||
_logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); | _logger.Error<MqttClientSession>(exception, "Client '{0}': Unhandled exception while processing client packets.", ClientId); | ||||
} | } | ||||
await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); | |||||
Stop(MqttClientDisconnectType.NotClean); | |||||
} | } | ||||
} | } | ||||
@@ -263,16 +261,20 @@ namespace MQTTnet.Server | |||||
if (packet is MqttDisconnectPacket) | if (packet is MqttDisconnectPacket) | ||||
{ | { | ||||
return StopAsync(MqttClientDisconnectType.Clean); | |||||
Stop(MqttClientDisconnectType.Clean); | |||||
return Task.FromResult(0); | |||||
} | } | ||||
if (packet is MqttConnectPacket) | if (packet is MqttConnectPacket) | ||||
{ | { | ||||
return StopAsync(MqttClientDisconnectType.NotClean); | |||||
Stop(MqttClientDisconnectType.NotClean); | |||||
return Task.FromResult(0); | |||||
} | } | ||||
_logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); | _logger.Warning<MqttClientSession>("Client '{0}': Received not supported packet ({1}). Closing connection.", ClientId, packet); | ||||
return StopAsync(MqttClientDisconnectType.NotClean); | |||||
Stop(MqttClientDisconnectType.NotClean); | |||||
return Task.FromResult(0); | |||||
} | } | ||||
private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters) | private async Task EnqueueSubscribedRetainedMessagesAsync(ICollection<TopicFilter> topicFilters) | ||||
@@ -291,7 +293,8 @@ namespace MQTTnet.Server | |||||
if (subscribeResult.CloseConnection) | if (subscribeResult.CloseConnection) | ||||
{ | { | ||||
await StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); | |||||
Stop(MqttClientDisconnectType.NotClean); | |||||
return; | |||||
} | } | ||||
await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); | await EnqueueSubscribedRetainedMessagesAsync(subscribePacket.TopicFilters).ConfigureAwait(false); | ||||
@@ -120,7 +120,7 @@ namespace MQTTnet.Server | |||||
{ | { | ||||
foreach (var session in _sessions) | foreach (var session in _sessions) | ||||
{ | { | ||||
await session.Value.StopAsync(MqttClientDisconnectType.NotClean).ConfigureAwait(false); | |||||
session.Value.Stop(MqttClientDisconnectType.NotClean); | |||||
} | } | ||||
_sessions.Clear(); | _sessions.Clear(); | ||||
@@ -232,7 +232,7 @@ namespace MQTTnet.Server | |||||
{ | { | ||||
_sessions.Remove(connectPacket.ClientId); | _sessions.Remove(connectPacket.ClientId); | ||||
await clientSession.StopAsync(MqttClientDisconnectType.Clean).ConfigureAwait(false); | |||||
clientSession.Stop(MqttClientDisconnectType.Clean); | |||||
clientSession.Dispose(); | clientSession.Dispose(); | ||||
clientSession = null; | clientSession = null; | ||||
@@ -270,7 +270,7 @@ namespace MQTTnet.Server | |||||
var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); | var interceptorContext = InterceptApplicationMessage(senderClientSession, applicationMessage); | ||||
if (interceptorContext.CloseConnection) | if (interceptorContext.CloseConnection) | ||||
{ | { | ||||
await senderClientSession.StopAsync().ConfigureAwait(false); | |||||
senderClientSession.Stop(MqttClientDisconnectType.NotClean); | |||||
} | } | ||||
if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) | if (interceptorContext.ApplicationMessage == null || !interceptorContext.AcceptPublish) | ||||
@@ -1,5 +1,4 @@ | |||||
using System.Threading; | using System.Threading; | ||||
using System.Threading.Tasks; | |||||
using Microsoft.VisualStudio.TestTools.UnitTesting; | using Microsoft.VisualStudio.TestTools.UnitTesting; | ||||
using MQTTnet.Diagnostics; | using MQTTnet.Diagnostics; | ||||
using MQTTnet.Packets; | using MQTTnet.Packets; | ||||
@@ -18,7 +17,6 @@ namespace MQTTnet.Core.Tests | |||||
var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate | var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate | ||||
{ | { | ||||
timeoutCalledCount++; | timeoutCalledCount++; | ||||
return Task.FromResult(0); | |||||
}, new MqttNetLogger()); | }, new MqttNetLogger()); | ||||
Assert.AreEqual(0, timeoutCalledCount); | Assert.AreEqual(0, timeoutCalledCount); | ||||
@@ -40,7 +38,6 @@ namespace MQTTnet.Core.Tests | |||||
var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate | var monitor = new MqttClientKeepAliveMonitor(string.Empty, delegate | ||||
{ | { | ||||
timeoutCalledCount++; | timeoutCalledCount++; | ||||
return Task.FromResult(0); | |||||
}, new MqttNetLogger()); | }, new MqttNetLogger()); | ||||
Assert.AreEqual(0, timeoutCalledCount); | Assert.AreEqual(0, timeoutCalledCount); | ||||