1724 nullreferenceexception on mqttclientconnectasync (#1748)

Update MqttClient.cs

Author: chkr1011

Source/MQTTnet/Client/MqttClient.cs

@@ -121,19 +121,19 @@ public async Task<MqttClientConnectResult> ConnectAsync(MqttClientOptions option
  var mqttClientAliveToken = _mqttClientAlive.Token;
 
  var adapter = _adapterFactory.CreateClientAdapter(options, new MqttPacketInspector(_events.InspectPacketEvent, _rootLogger), _rootLogger);
- _adapter = adapter;
+ _adapter = adapter ?? throw new InvalidOperationException("The adapter factory did not provide an adapter.");
 
  if (cancellationToken.CanBeCanceled)
  {
- connectResult = await ConnectInternal(cancellationToken).ConfigureAwait(false);
+ connectResult = await ConnectInternal(adapter, cancellationToken).ConfigureAwait(false);
  }
  else
  {
  // Fall back to the general timeout specified in the options if the user passed
  // CancellationToken.None or similar.
  using (var timeout = new CancellationTokenSource(Options.Timeout))
  {
- connectResult = await ConnectInternal(timeout.Token).ConfigureAwait(false);
+ connectResult = await ConnectInternal(adapter, timeout.Token).ConfigureAwait(false);
  }
  }
 
@@ -421,7 +421,7 @@ Task AcknowledgeReceivedPublishPacket(MqttApplicationMessageReceivedEventArgs ev
  return CompletedTask.Instance;
  }
 
- async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, CancellationToken cancellationToken)
+ async Task<MqttClientConnectResult> Authenticate(IMqttChannelAdapter channelAdapter, MqttClientOptions options, CancellationToken cancellationToken)
  {
  MqttClientConnectResult result;
 
@@ -435,7 +435,7 @@ async Task<MqttClientConnectResult> Authenticate(MqttClientOptions options, Canc
  if (receivedPacket is MqttConnAckPacket connAckPacket)
  {
  var clientConnectResultFactory = new MqttClientConnectResultFactory();
- result = clientConnectResultFactory.Create(connAckPacket, _adapter.PacketFormatterAdapter.ProtocolVersion);
+ result = clientConnectResultFactory.Create(connAckPacket, channelAdapter.PacketFormatterAdapter.ProtocolVersion);
  }
  else
  {
@@ -484,7 +484,7 @@ MqttClientConnectionStatus CompareExchangeConnectionStatus(MqttClientConnectionS
  return (MqttClientConnectionStatus)Interlocked.CompareExchange(ref _connectionStatus, (int)value, (int)comparand);
  }
 
- async Task<MqttClientConnectResult> ConnectInternal(CancellationToken cancellationToken)
+ async Task<MqttClientConnectResult> ConnectInternal(IMqttChannelAdapter channelAdapter, CancellationToken cancellationToken)
  {
  var backgroundCancellationToken = _mqttClientAlive.Token;
 
@@ -497,7 +497,7 @@ async Task<MqttClientConnectResult> ConnectInternal(CancellationToken cancellati
  _publishPacketReceiverQueue?.Dispose();
  _publishPacketReceiverQueue = new AsyncQueue<MqttPublishPacket>();
 
- var connectResult = await Authenticate(Options, effectiveCancellationToken.Token).ConfigureAwait(false);
+ var connectResult = await Authenticate(channelAdapter, Options, effectiveCancellationToken.Token).ConfigureAwait(false);
 
  _publishPacketReceiverTask = Task.Run(() => ProcessReceivedPublishPackets(backgroundCancellationToken), backgroundCancellationToken);
  _packetReceiverTask = Task.Run(() => ReceivePacketsLoop(backgroundCancellationToken), backgroundCancellationToken);