Fix breaking changes

This commit is contained in:
Christian
2023-05-06 13:09:02 +02:00
parent 3aaf101b3b
commit 533b8e22c2
3 changed files with 69 additions and 127 deletions
@@ -57,7 +57,7 @@ namespace MQTTnet.Client.Internal
{ {
// Send the CONNECT packet.. // Send the CONNECT packet..
var connectPacket = MqttPacketFactories.Connect.Create(options); var connectPacket = MqttPacketFactories.Connect.Create(options);
await _client.SendAsync(connectPacket, cancellationToken).ConfigureAwait(false); await _client.Send(connectPacket, cancellationToken).ConfigureAwait(false);
// Wait for the CONNACK packet... // Wait for the CONNACK packet...
MqttConnAckPacket connAckPacket = null; MqttConnAckPacket connAckPacket = null;
@@ -77,7 +77,7 @@ namespace MQTTnet.Client.Internal
var response = await OnExtendedAuthentication(authPacket).ConfigureAwait(false); var response = await OnExtendedAuthentication(authPacket).ConfigureAwait(false);
authPacket = MqttPacketFactories.Auth.Create(authPacket, response); authPacket = MqttPacketFactories.Auth.Create(authPacket, response);
await _client.SendAsync(authPacket, cancellationToken).ConfigureAwait(false); await _client.Send(authPacket, cancellationToken).ConfigureAwait(false);
} }
else if (receivedPacket is MqttConnAckPacket connAckPacketBuffer) else if (receivedPacket is MqttConnAckPacket connAckPacketBuffer)
{ {
@@ -14,10 +14,10 @@ namespace MQTTnet.Client.Internal
{ {
readonly MqttClient _client; readonly MqttClient _client;
readonly MqttNetSourceLogger _logger; readonly MqttNetSourceLogger _logger;
CancellationTokenSource _cancellationToken; CancellationTokenSource _cancellationToken;
TimeSpan _keepAliveInterval; TimeSpan _keepAliveInterval;
Task _keepAlivePacketsSenderTask; Task _keepAlivePacketsSenderTask;
DateTime _lastPacketSentTimestamp; DateTime _lastPacketSentTimestamp;
public MqttClientKeepAliveHandler(MqttClient client, IMqttNetLogger logger) public MqttClientKeepAliveHandler(MqttClient client, IMqttNetLogger logger)
+66 -124
View File
@@ -21,13 +21,11 @@ namespace MQTTnet.Client
public sealed class MqttClient : Disposable, IMqttClient public sealed class MqttClient : Disposable, IMqttClient
{ {
readonly IMqttClientAdapterFactory _adapterFactory; readonly IMqttClientAdapterFactory _adapterFactory;
readonly MqttClientAuthenticationHandler _authenticationHandler;
readonly object _disconnectLock = new object(); readonly object _disconnectLock = new object();
readonly MqttClientEvents _events = new MqttClientEvents(); readonly MqttClientEvents _events = new MqttClientEvents();
readonly MqttNetSourceLogger _logger;
readonly MqttClientAuthenticationHandler _authenticationHandler;
readonly MqttClientKeepAliveHandler _keepAliveHandler; readonly MqttClientKeepAliveHandler _keepAliveHandler;
readonly MqttNetSourceLogger _logger;
readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider(); readonly MqttPacketIdentifierProvider _packetIdentifierProvider = new MqttPacketIdentifierProvider();
readonly IMqttNetLogger _rootLogger; readonly IMqttNetLogger _rootLogger;
@@ -143,17 +141,6 @@ namespace MQTTnet.Client
} }
} }
if (connectResult.ResultCode != MqttClientConnectResultCode.Success)
{
_logger.Info($"Using keep alive value ({connectResult.ServerKeepAlive}) sent from the server");
keepAliveInterval = TimeSpan.FromSeconds(connectResult.ServerKeepAlive);
}
if (keepAliveInterval != TimeSpan.Zero)
{
_keepAlivePacketsSenderTask = Task.Run(() => TrySendKeepAliveMessages(mqttClientAliveToken), mqttClientAliveToken);
}
CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting); CompareExchangeConnectionStatus(MqttClientConnectionStatus.Connected, MqttClientConnectionStatus.Connecting);
_logger.Info("Connected"); _logger.Info("Connected");
@@ -390,13 +377,13 @@ namespace MQTTnet.Client
if (!DisconnectIsPendingOrFinished()) if (!DisconnectIsPendingOrFinished())
{ {
return DisconnectCoreAsync(sender, exception, connectResult, clientWasConnected); return DisconnectCore(sender, exception, connectResult, clientWasConnected);
} }
return CompletedTask.Instance; return CompletedTask.Instance;
} }
internal Task SendAsync(MqttPacket packet, CancellationToken cancellationToken) internal Task Send(MqttPacket packet, CancellationToken cancellationToken)
{ {
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
@@ -445,42 +432,6 @@ namespace MQTTnet.Client
return CompletedTask.Instance; return CompletedTask.Instance;
} }
async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken)
{
MqttClientConnectResult result;
try
{
var connectPacket = MqttPacketFactories.Connect.Create(options);
await Send(connectPacket, cancellationToken).ConfigureAwait(false);
var receivedPacket = await Receive(cancellationToken).ConfigureAwait(false);
if (receivedPacket is MqttConnAckPacket connAckPacket)
{
var clientConnectResultFactory = new MqttClientConnectResultFactory();
result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
}
else
{
throw new NotSupportedException("Extended authentication handler is not yet supported");
}
}
catch (Exception exception)
{
throw new MqttConnectingFailedException($"Error while authenticating. {exception.Message}", exception, null);
}
if (result.ResultCode != MqttClientConnectResultCode.Success)
{
throw new MqttConnectingFailedException($"Connecting with MQTT server failed ({result.ResultCode}).", null, result);
}
_logger.Verbose("Authenticated MQTT connection with server established.");
return result;
}
void Cleanup() void Cleanup()
{ {
try try
@@ -527,8 +478,8 @@ namespace MQTTnet.Client
_publishPacketReceiverQueue?.Dispose(); _publishPacketReceiverQueue?.Dispose();
_publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>(); _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
_publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(clientAliveToken), clientAliveToken); _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPacketsLoop(clientAliveToken), clientAliveToken);
_packetReceiverTask = Task.Run(() => TryReceivePacketsAsync(clientAliveToken), clientAliveToken); _packetReceiverTask = Task.Run(() => ReceivePacketsLoop(clientAliveToken), clientAliveToken);
_keepAliveHandler.Enable(connectResult); _keepAliveHandler.Enable(connectResult);
} }
@@ -691,7 +642,7 @@ namespace MQTTnet.Client
return DisconnectInternal(_packetReceiverTask, null, null); return DisconnectInternal(_packetReceiverTask, null, null);
} }
async Task ProcessReceivedPublishPackets(CancellationToken cancellationToken) async Task ProcessReceivedPublishPacketsLoop(CancellationToken cancellationToken)
{ {
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
{ {
@@ -786,6 +737,65 @@ namespace MQTTnet.Client
return packet; return packet;
} }
async Task ReceivePacketsLoop(CancellationToken cancellationToken)
{
try
{
_logger.Verbose("Start receiving packets.");
while (!cancellationToken.IsCancellationRequested)
{
var packet = await Receive(cancellationToken).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (packet == null)
{
await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
return;
}
await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
{
if (_cleanDisconnectInitiated)
{
return;
}
if (exception is AggregateException aggregateException)
{
exception = aggregateException.GetBaseException();
}
if (exception is OperationCanceledException)
{
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
}
else
{
_logger.Error(exception, "Error while receiving packets.");
}
_packetDispatcher.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped receiving packets.");
}
}
async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket async Task<TResponsePacket> Request<TResponsePacket>(MqttPacket requestPacket, CancellationToken cancellationToken) where TResponsePacket : MqttPacket
{ {
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
@@ -824,15 +834,6 @@ namespace MQTTnet.Client
} }
} }
Task Send(MqttPacket packet, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
_lastPacketSentTimestamp = DateTime.UtcNow;
return _adapter.SendPacketAsync(packet, cancellationToken);
}
void ThrowIfConnected(string message) void ThrowIfConnected(string message)
{ {
if (IsConnected) if (IsConnected)
@@ -951,64 +952,5 @@ namespace MQTTnet.Client
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false); await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
} }
} }
async Task ReceivePacketsLoop(CancellationToken cancellationToken)
{
try
{
_logger.Verbose("Start receiving packets.");
while (!cancellationToken.IsCancellationRequested)
{
var packet = await Receive(cancellationToken).ConfigureAwait(false);
if (cancellationToken.IsCancellationRequested)
{
return;
}
if (packet == null)
{
await DisconnectInternal(_packetReceiverTask, null, null).ConfigureAwait(false);
return;
}
await TryProcessReceivedPacket(packet, cancellationToken).ConfigureAwait(false);
}
}
catch (Exception exception)
{
if (_cleanDisconnectInitiated)
{
return;
}
if (exception is AggregateException aggregateException)
{
exception = aggregateException.GetBaseException();
}
if (exception is OperationCanceledException)
{
}
else if (exception is MqttCommunicationException)
{
_logger.Warning(exception, "Communication error while receiving packets.");
}
else
{
_logger.Error(exception, "Error while receiving packets.");
}
_packetDispatcher.FailAll(exception);
await DisconnectInternal(_packetReceiverTask, exception, null).ConfigureAwait(false);
}
finally
{
_logger.Verbose("Stopped receiving packets.");
}
}
} }
} }