@@ -37,10 +37,12 @@ public partial class HiveMQClient : IDisposable, IHiveMQClient
3737{
3838 private static readonly NLog . Logger Logger = NLog . LogManager . GetCurrentClassLogger ( ) ;
3939
40- private ConnectState connectState = ConnectState . Disconnected ;
40+ internal ConnectState ConnectState { get ; set ; }
4141
4242 public HiveMQClient ( HiveMQClientOptions ? options = null )
4343 {
44+ this . ConnectState = ConnectState . Disconnected ;
45+
4446 options ??= new HiveMQClientOptions ( ) ;
4547 options . Validate ( ) ;
4648
@@ -66,12 +68,12 @@ public HiveMQClient(HiveMQClientOptions? options = null)
6668 public List < Subscription > Subscriptions { get ; } = new ( ) ;
6769
6870 /// <inheritdoc />
69- public bool IsConnected ( ) => this . connectState == ConnectState . Connected ;
71+ public bool IsConnected ( ) => this . ConnectState == ConnectState . Connected ;
7072
7173 /// <inheritdoc />
7274 public async Task < ConnectResult > ConnectAsync ( )
7375 {
74- this . connectState = ConnectState . Connecting ;
76+ this . ConnectState = ConnectState . Connecting ;
7577
7678 Logger . Info ( "Connecting to broker at {0}:{1}" , this . Options . Host , this . Options . Port ) ;
7779
@@ -89,7 +91,7 @@ public async Task<ConnectResult> ConnectAsync()
8991 // Construct the MQTT Connect packet and queue to send
9092 var connPacket = new ConnectPacket ( this . Options ) ;
9193 Logger . Trace ( $ "Queuing packet for send: { connPacket } ") ;
92- this . sendQueue . Add ( connPacket ) ;
94+ this . SendQueue . Add ( connPacket ) ;
9395
9496 // FIXME: Cancellation token and better timeout value
9597 ConnAckPacket connAck ;
@@ -100,7 +102,7 @@ public async Task<ConnectResult> ConnectAsync()
100102 }
101103 catch ( TimeoutException )
102104 {
103- this . connectState = ConnectState . Disconnected ;
105+ this . ConnectState = ConnectState . Disconnected ;
104106 throw new HiveMQttClientException ( "Connect timeout. No response received in time." ) ;
105107 }
106108 finally
@@ -111,11 +113,11 @@ public async Task<ConnectResult> ConnectAsync()
111113
112114 if ( connAck . ReasonCode == ConnAckReasonCode . Success )
113115 {
114- this . connectState = ConnectState . Connected ;
116+ this . ConnectState = ConnectState . Connected ;
115117 }
116118 else
117119 {
118- this . connectState = ConnectState . Disconnected ;
120+ this . ConnectState = ConnectState . Disconnected ;
119121 }
120122
121123 connectResult = new ConnectResult ( connAck . ReasonCode , connAck . SessionPresent , connAck . Properties ) ;
@@ -133,7 +135,7 @@ public async Task<ConnectResult> ConnectAsync()
133135 /// <inheritdoc />
134136 public async Task < bool > DisconnectAsync ( DisconnectOptions ? options = null )
135137 {
136- if ( this . connectState != ConnectState . Connected )
138+ if ( this . ConnectState != ConnectState . Connected )
137139 {
138140 Logger . Warn ( "DisconnectAsync: Client is not connected." ) ;
139141 return false ;
@@ -152,15 +154,15 @@ public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
152154 } ;
153155
154156 // Once this is set, no more incoming packets or outgoing will be accepted
155- this . connectState = ConnectState . Disconnecting ;
157+ this . ConnectState = ConnectState . Disconnecting ;
156158
157159 var taskCompletionSource = new TaskCompletionSource < DisconnectPacket > ( ) ;
158160 void TaskHandler ( object ? sender , OnDisconnectSentEventArgs args ) => taskCompletionSource . SetResult ( args . DisconnectPacket ) ;
159161 EventHandler < OnDisconnectSentEventArgs > eventHandler = TaskHandler ;
160162 this . OnDisconnectSent += eventHandler ;
161163
162164 Logger . Trace ( $ "Queuing packet for send: { disconnectPacket } ") ;
163- this . sendQueue . Add ( disconnectPacket ) ;
165+ this . SendQueue . Add ( disconnectPacket ) ;
164166
165167 try
166168 {
@@ -176,26 +178,35 @@ public async Task<bool> DisconnectAsync(DisconnectOptions? options = null)
176178 this . OnDisconnectSent -= eventHandler ;
177179 }
178180
179- // Close the socket
181+ this . HandleDisconnection ( ) ;
182+
183+ return true ;
184+ }
185+
186+ /// <summary>
187+ /// Close the socket and set the connect state to disconnected.
188+ /// </summary>
189+ private void HandleDisconnection ( )
190+ {
191+ Logger . Debug ( "HandleDisconnection: Connection lost. Handling Disconnection." ) ;
192+
180193 this . CloseSocket ( ) ;
181194
182195 // Fire the corresponding event
183- this . AfterDisconnectEventLauncher ( true ) ;
196+ this . AfterDisconnectEventLauncher ( false ) ;
184197
185- this . connectState = ConnectState . Disconnected ;
198+ this . ConnectState = ConnectState . Disconnected ;
186199
187200 // FIXME
188- if ( this . sendQueue . Count > 0 )
201+ if ( this . SendQueue . Count > 0 )
189202 {
190- Logger . Warn ( "Disconnect : Send queue not empty. Packets pending but we are disconnecting.") ;
203+ Logger . Warn ( $ "HandleDisconnection : Send queue not empty. { this . SendQueue . Count } packets pending but we are disconnecting (or were disconnected) .") ;
191204 }
192205
193206 // We only clear the send queue on explicit disconnect
194- while ( this . sendQueue . TryTake ( out _ ) )
207+ while ( this . SendQueue . TryTake ( out _ ) )
195208 {
196209 }
197-
198- return true ;
199210 }
200211
201212 /// <inheritdoc />
@@ -210,7 +221,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
210221 if ( message . QoS == QualityOfService . AtMostOnceDelivery )
211222 {
212223 Logger . Trace ( $ "Queuing packet for send: { publishPacket } ") ;
213- this . sendQueue . Add ( publishPacket ) ;
224+ this . SendQueue . Add ( publishPacket ) ;
214225 return new PublishResult ( publishPacket . Message ) ;
215226 }
216227 else if ( message . QoS == QualityOfService . AtLeastOnceDelivery )
@@ -223,7 +234,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
223234
224235 // Construct the MQTT Connect packet and queue to send
225236 Logger . Trace ( $ "Queuing packet for send: { publishPacket } ") ;
226- this . sendQueue . Add ( publishPacket ) ;
237+ this . SendQueue . Add ( publishPacket ) ;
227238
228239 var pubAckPacket = await taskCompletionSource . Task . WaitAsync ( TimeSpan . FromSeconds ( 120 ) ) . ConfigureAwait ( false ) ;
229240
@@ -239,7 +250,7 @@ public async Task<PublishResult> PublishAsync(MQTT5PublishMessage message)
239250 publishPacket . OnPublishQoS2Complete += eventHandler ;
240251
241252 // Construct the MQTT Connect packet and queue to send
242- this . sendQueue . Add ( publishPacket ) ;
253+ this . SendQueue . Add ( publishPacket ) ;
243254
244255 // Wait on the QoS 2 handshake
245256 var packetList = await taskCompletionSource . Task . WaitAsync ( TimeSpan . FromSeconds ( 120 ) ) . ConfigureAwait ( false ) ;
@@ -322,7 +333,7 @@ public async Task<SubscribeResult> SubscribeAsync(SubscribeOptions options)
322333 this . OnSubAckReceived += eventHandler ;
323334
324335 // Queue the constructed packet to be sent on the wire
325- this . sendQueue . Add ( subscribePacket ) ;
336+ this . SendQueue . Add ( subscribePacket ) ;
326337
327338 SubAckPacket subAck ;
328339 SubscribeResult subscribeResult ;
@@ -426,7 +437,7 @@ public async Task<UnsubscribeResult> UnsubscribeAsync(UnsubscribeOptions unsubOp
426437 EventHandler < OnUnsubAckReceivedEventArgs > eventHandler = TaskHandler ;
427438 this . OnUnsubAckReceived += eventHandler ;
428439
429- this . sendQueue . Add ( unsubscribePacket ) ;
440+ this . SendQueue . Add ( unsubscribePacket ) ;
430441
431442 // FIXME: Cancellation token and better timeout value
432443 UnsubAckPacket unsubAck ;
0 commit comments