|
@@ -1,2637 +0,0 @@
|
|
|
-
|
|
|
-Copyright (c) 2013, 2014 Paolo Patierno
|
|
|
-
|
|
|
-All rights reserved. This program and the accompanying materials
|
|
|
-are made available under the terms of the Eclipse Public License v1.0
|
|
|
-and Eclipse Distribution License v1.0 which accompany this distribution.
|
|
|
-
|
|
|
-The Eclipse Public License is available at
|
|
|
- http:
|
|
|
-and the Eclipse Distribution License is available at
|
|
|
- http:
|
|
|
-
|
|
|
-Contributors:
|
|
|
- Paolo Patierno - initial API and implementation and/or initial documentation
|
|
|
-
|
|
|
- ----------------------------------------------------------------------------
|
|
|
-
|
|
|
- Giovanni Paolo Vigano' - preprocessor directives for platform dependent compilation in Unity
|
|
|
-*/
|
|
|
-
|
|
|
-using System;
|
|
|
-using System.Net;
|
|
|
-#if !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
-using System.Net.Sockets;
|
|
|
-using System.Security.Cryptography.X509Certificates;
|
|
|
-#endif
|
|
|
-using System.Threading;
|
|
|
-using uPLibrary.Networking.M2Mqtt.Exceptions;
|
|
|
-using uPLibrary.Networking.M2Mqtt.Messages;
|
|
|
-using uPLibrary.Networking.M2Mqtt.Session;
|
|
|
-using uPLibrary.Networking.M2Mqtt.Utility;
|
|
|
-using uPLibrary.Networking.M2Mqtt.Internal;
|
|
|
-
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3)
|
|
|
-using Microsoft.SPOT;
|
|
|
-#if SSL
|
|
|
-using Microsoft.SPOT.Net.Security;
|
|
|
-#endif
|
|
|
-
|
|
|
-#else
|
|
|
-using System.Collections.Generic;
|
|
|
-#if (SSL && !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP)))
|
|
|
-using System.Security.Authentication;
|
|
|
-using System.Net.Security;
|
|
|
-#endif
|
|
|
-#endif
|
|
|
-
|
|
|
-#if (WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR && UNITY_WSA_10_0 && !ENABLE_IL2CPP))
|
|
|
-using Windows.Networking.Sockets;
|
|
|
-#endif
|
|
|
-
|
|
|
-using System.Collections;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-using MqttUtility = uPLibrary.Networking.M2Mqtt.Utility;
|
|
|
-using System.IO;
|
|
|
-using System.Net.Security;
|
|
|
-
|
|
|
-namespace uPLibrary.Networking.M2Mqtt
|
|
|
-{
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public class MqttClient
|
|
|
- {
|
|
|
-#if BROKER
|
|
|
- #region Constants ...
|
|
|
-
|
|
|
-
|
|
|
- private const string RECEIVE_THREAD_NAME = "ReceiveThread";
|
|
|
- private const string RECEIVE_EVENT_THREAD_NAME = "DispatchEventThread";
|
|
|
- private const string PROCESS_INFLIGHT_THREAD_NAME = "ProcessInflightThread";
|
|
|
- private const string KEEP_ALIVE_THREAD = "KeepAliveThread";
|
|
|
-
|
|
|
- #endregion
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgPublishEventHandler(object sender, MqttMsgPublishEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgPublishedEventHandler(object sender, MqttMsgPublishedEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgSubscribedEventHandler(object sender, MqttMsgSubscribedEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgUnsubscribedEventHandler(object sender, MqttMsgUnsubscribedEventArgs e);
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgSubscribeEventHandler(object sender, MqttMsgSubscribeEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgUnsubscribeEventHandler(object sender, MqttMsgUnsubscribeEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgConnectEventHandler(object sender, MqttMsgConnectEventArgs e);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void MqttMsgDisconnectEventHandler(object sender, EventArgs e);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public delegate void ConnectionClosedEventHandler(object sender, EventArgs e);
|
|
|
-
|
|
|
-
|
|
|
- private string brokerHostName;
|
|
|
- private int brokerPort;
|
|
|
-
|
|
|
-
|
|
|
- private bool isRunning;
|
|
|
-
|
|
|
- private AutoResetEvent receiveEventWaitHandle;
|
|
|
-
|
|
|
-
|
|
|
- private AutoResetEvent inflightWaitHandle;
|
|
|
-
|
|
|
-
|
|
|
- AutoResetEvent syncEndReceiving;
|
|
|
-
|
|
|
- MqttMsgBase msgReceived;
|
|
|
-
|
|
|
-
|
|
|
- Exception exReceiving;
|
|
|
-
|
|
|
-
|
|
|
- private int keepAlivePeriod;
|
|
|
-
|
|
|
- private AutoResetEvent keepAliveEvent;
|
|
|
- private AutoResetEvent keepAliveEventEnd;
|
|
|
-
|
|
|
- private int lastCommTime;
|
|
|
-
|
|
|
-
|
|
|
- public event MqttMsgPublishEventHandler MqttMsgPublishReceived;
|
|
|
-
|
|
|
- public event MqttMsgPublishedEventHandler MqttMsgPublished;
|
|
|
-
|
|
|
- public event MqttMsgSubscribedEventHandler MqttMsgSubscribed;
|
|
|
-
|
|
|
- public event MqttMsgUnsubscribedEventHandler MqttMsgUnsubscribed;
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- public event MqttMsgSubscribeEventHandler MqttMsgSubscribeReceived;
|
|
|
-
|
|
|
- public event MqttMsgUnsubscribeEventHandler MqttMsgUnsubscribeReceived;
|
|
|
-
|
|
|
- public event MqttMsgConnectEventHandler MqttMsgConnected;
|
|
|
-
|
|
|
- public event MqttMsgDisconnectEventHandler MqttMsgDisconnected;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- public event ConnectionClosedEventHandler ConnectionClosed;
|
|
|
-
|
|
|
-
|
|
|
- private IMqttNetworkChannel channel;
|
|
|
-
|
|
|
-
|
|
|
- private Queue inflightQueue;
|
|
|
-
|
|
|
- private Queue internalQueue;
|
|
|
-
|
|
|
- private Queue eventQueue;
|
|
|
-
|
|
|
- private MqttClientSession session;
|
|
|
-
|
|
|
-
|
|
|
- private MqttSettings settings;
|
|
|
-
|
|
|
-
|
|
|
- private ushort messageIdCounter = 0;
|
|
|
-
|
|
|
-
|
|
|
- private bool isConnectionClosing;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public bool IsConnected { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public string ClientId { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public bool CleanSession { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public bool WillFlag { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public byte WillQosLevel { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public string WillTopic { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public string WillMessage { get; private set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttProtocolVersion ProtocolVersion { get; set; }
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClientSession Session
|
|
|
- {
|
|
|
- get { return this.session; }
|
|
|
- set { this.session = value; }
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttSettings Settings
|
|
|
- {
|
|
|
- get { return this.settings; }
|
|
|
- }
|
|
|
-
|
|
|
-#if !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- [Obsolete("Use this ctor MqttClient(string brokerHostName) insted")]
|
|
|
- public MqttClient(IPAddress brokerIpAddress) :
|
|
|
- this(brokerIpAddress, MqttSettings.MQTT_BROKER_DEFAULT_PORT, false, null, null, MqttSslProtocols.None)
|
|
|
- {
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- [Obsolete("Use this ctor MqttClient(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert) insted")]
|
|
|
- public MqttClient(IPAddress brokerIpAddress, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol)
|
|
|
- {
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- this.Init(brokerIpAddress.ToString(), brokerPort, secure, caCert, clientCert, sslProtocol, null, null);
|
|
|
-#else
|
|
|
- this.Init(brokerIpAddress.ToString(), brokerPort, secure, caCert, clientCert, sslProtocol);
|
|
|
-#endif
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(string brokerHostName) :
|
|
|
-#if !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- this(brokerHostName, MqttSettings.MQTT_BROKER_DEFAULT_PORT, false, null, null, MqttSslProtocols.None)
|
|
|
-#else
|
|
|
- this(brokerHostName, MqttSettings.MQTT_BROKER_DEFAULT_PORT, false, MqttSslProtocols.None)
|
|
|
-#endif
|
|
|
- {
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-#if !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR && UNITY_WSA_10_0 && !ENABLE_IL2CPP))
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol)
|
|
|
-#else
|
|
|
- public MqttClient(string brokerHostName, int brokerPort, bool secure, MqttSslProtocols sslProtocol)
|
|
|
-#endif
|
|
|
- {
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK || WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- this.Init(brokerHostName, brokerPort, secure, caCert, clientCert, sslProtocol, null, null);
|
|
|
-#elif (WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- this.Init(brokerHostName, brokerPort, secure, sslProtocol);
|
|
|
-#else
|
|
|
- this.Init(brokerHostName, brokerPort, secure, caCert, clientCert, sslProtocol);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK || WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol,
|
|
|
- RemoteCertificateValidationCallback userCertificateValidationCallback)
|
|
|
- : this(brokerHostName, brokerPort, secure, caCert, clientCert, sslProtocol, userCertificateValidationCallback, null)
|
|
|
- {
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(string brokerHostName, int brokerPort, bool secure, MqttSslProtocols sslProtocol,
|
|
|
- RemoteCertificateValidationCallback userCertificateValidationCallback,
|
|
|
- LocalCertificateSelectionCallback userCertificateSelectionCallback)
|
|
|
- : this(brokerHostName, brokerPort, secure, null, null, sslProtocol, userCertificateValidationCallback, userCertificateSelectionCallback)
|
|
|
- {
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol,
|
|
|
- RemoteCertificateValidationCallback userCertificateValidationCallback,
|
|
|
- LocalCertificateSelectionCallback userCertificateSelectionCallback)
|
|
|
- {
|
|
|
- this.Init(brokerHostName, brokerPort, secure, caCert, clientCert, sslProtocol, userCertificateValidationCallback, userCertificateSelectionCallback);
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public MqttClient(IMqttNetworkChannel channel)
|
|
|
- {
|
|
|
-
|
|
|
- this.ProtocolVersion = MqttProtocolVersion.Version_3_1_1;
|
|
|
-
|
|
|
- this.channel = channel;
|
|
|
-
|
|
|
-
|
|
|
- this.settings = MqttSettings.Instance;
|
|
|
-
|
|
|
-
|
|
|
- this.IsConnected = false;
|
|
|
- this.ClientId = null;
|
|
|
- this.CleanSession = true;
|
|
|
-
|
|
|
- this.keepAliveEvent = new AutoResetEvent(false);
|
|
|
-
|
|
|
-
|
|
|
- this.inflightWaitHandle = new AutoResetEvent(false);
|
|
|
- this.inflightQueue = new Queue();
|
|
|
-
|
|
|
-
|
|
|
- this.receiveEventWaitHandle = new AutoResetEvent(false);
|
|
|
- this.eventQueue = new Queue();
|
|
|
- this.internalQueue = new Queue();
|
|
|
-
|
|
|
-
|
|
|
- this.session = null;
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK || WINDOWS_APP || WINDOWS_PHONE_APP || ((!UNITY_EDITOR && UNITY_WSA_10_0 && !ENABLE_IL2CPP)))
|
|
|
-
|
|
|
-
|
|
|
- private void Init(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol,
|
|
|
- RemoteCertificateValidationCallback userCertificateValidationCallback,
|
|
|
- LocalCertificateSelectionCallback userCertificateSelectionCallback)
|
|
|
-#elif (WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- private void Init(string brokerHostName, int brokerPort, bool secure, MqttSslProtocols sslProtocol)
|
|
|
-#else
|
|
|
- private void Init(string brokerHostName, int brokerPort, bool secure, X509Certificate caCert, X509Certificate clientCert, MqttSslProtocols sslProtocol)
|
|
|
-#endif
|
|
|
- {
|
|
|
-
|
|
|
- this.ProtocolVersion = MqttProtocolVersion.Version_3_1_1;
|
|
|
-#if !SSL
|
|
|
-
|
|
|
- if (secure)
|
|
|
- throw new ArgumentException("Library compiled without SSL support");
|
|
|
-#endif
|
|
|
-
|
|
|
- this.brokerHostName = brokerHostName;
|
|
|
- this.brokerPort = brokerPort;
|
|
|
-
|
|
|
-
|
|
|
- this.settings = MqttSettings.Instance;
|
|
|
-
|
|
|
- if (!secure)
|
|
|
- this.settings.Port = this.brokerPort;
|
|
|
- else
|
|
|
- this.settings.SslPort = this.brokerPort;
|
|
|
-
|
|
|
- this.syncEndReceiving = new AutoResetEvent(false);
|
|
|
- this.keepAliveEvent = new AutoResetEvent(false);
|
|
|
-
|
|
|
-
|
|
|
- this.inflightWaitHandle = new AutoResetEvent(false);
|
|
|
- this.inflightQueue = new Queue();
|
|
|
-
|
|
|
-
|
|
|
- this.receiveEventWaitHandle = new AutoResetEvent(false);
|
|
|
- this.eventQueue = new Queue();
|
|
|
- this.internalQueue = new Queue();
|
|
|
-
|
|
|
-
|
|
|
- this.session = null;
|
|
|
-
|
|
|
-
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK || WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- this.channel = new MqttNetworkChannel(this.brokerHostName, this.brokerPort, secure, caCert, clientCert, sslProtocol, userCertificateValidationCallback, userCertificateSelectionCallback);
|
|
|
-#elif (WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- this.channel = new MqttNetworkChannel(this.brokerHostName, this.brokerPort, secure, sslProtocol);
|
|
|
-#else
|
|
|
- this.channel = new MqttNetworkChannel(this.brokerHostName, this.brokerPort, secure, caCert, clientCert, sslProtocol);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public byte Connect(string clientId)
|
|
|
- {
|
|
|
- return this.Connect(clientId, null, null, false, MqttMsgConnect.QOS_LEVEL_AT_MOST_ONCE, false, null, null, true, MqttMsgConnect.KEEP_ALIVE_PERIOD_DEFAULT);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public byte Connect(string clientId,
|
|
|
- string username,
|
|
|
- string password)
|
|
|
- {
|
|
|
- return this.Connect(clientId, username, password, false, MqttMsgConnect.QOS_LEVEL_AT_MOST_ONCE, false, null, null, true, MqttMsgConnect.KEEP_ALIVE_PERIOD_DEFAULT);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public byte Connect(string clientId,
|
|
|
- string username,
|
|
|
- string password,
|
|
|
- bool cleanSession,
|
|
|
- ushort keepAlivePeriod)
|
|
|
- {
|
|
|
- return this.Connect(clientId, username, password, false, MqttMsgConnect.QOS_LEVEL_AT_MOST_ONCE, false, null, null, cleanSession, keepAlivePeriod);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public byte Connect(string clientId,
|
|
|
- string username,
|
|
|
- string password,
|
|
|
- bool willRetain,
|
|
|
- byte willQosLevel,
|
|
|
- bool willFlag,
|
|
|
- string willTopic,
|
|
|
- string willMessage,
|
|
|
- bool cleanSession,
|
|
|
- ushort keepAlivePeriod)
|
|
|
- {
|
|
|
-
|
|
|
- MqttMsgConnect connect = new MqttMsgConnect(clientId,
|
|
|
- username,
|
|
|
- password,
|
|
|
- willRetain,
|
|
|
- willQosLevel,
|
|
|
- willFlag,
|
|
|
- willTopic,
|
|
|
- willMessage,
|
|
|
- cleanSession,
|
|
|
- keepAlivePeriod,
|
|
|
- (byte)this.ProtocolVersion);
|
|
|
-
|
|
|
- try
|
|
|
- {
|
|
|
-
|
|
|
- this.channel.Connect();
|
|
|
- }
|
|
|
- catch (Exception ex)
|
|
|
- {
|
|
|
- throw new MqttConnectionException("Exception connecting to the broker", ex);
|
|
|
- }
|
|
|
-
|
|
|
- this.lastCommTime = 0;
|
|
|
- this.isRunning = true;
|
|
|
- this.isConnectionClosing = false;
|
|
|
-
|
|
|
- Fx.StartThread(this.ReceiveThread);
|
|
|
-
|
|
|
- MqttMsgConnack connack = (MqttMsgConnack)this.SendReceive(connect);
|
|
|
-
|
|
|
- if (connack.ReturnCode == MqttMsgConnack.CONN_ACCEPTED)
|
|
|
- {
|
|
|
-
|
|
|
- this.ClientId = clientId;
|
|
|
- this.CleanSession = cleanSession;
|
|
|
- this.WillFlag = willFlag;
|
|
|
- this.WillTopic = willTopic;
|
|
|
- this.WillMessage = willMessage;
|
|
|
- this.WillQosLevel = willQosLevel;
|
|
|
-
|
|
|
- this.keepAlivePeriod = keepAlivePeriod * 1000;
|
|
|
-
|
|
|
-
|
|
|
- this.RestoreSession();
|
|
|
-
|
|
|
-
|
|
|
- if (this.keepAlivePeriod != 0)
|
|
|
- {
|
|
|
-
|
|
|
- Fx.StartThread(this.KeepAliveThread);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.DispatchEventThread);
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.ProcessInflightThread);
|
|
|
-
|
|
|
- this.IsConnected = true;
|
|
|
- }
|
|
|
- return connack.ReturnCode;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void Disconnect()
|
|
|
- {
|
|
|
- MqttMsgDisconnect disconnect = new MqttMsgDisconnect();
|
|
|
- this.Send(disconnect);
|
|
|
-
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
- }
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void Open()
|
|
|
- {
|
|
|
- this.isRunning = true;
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.ReceiveThread);
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.DispatchEventThread);
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.ProcessInflightThread);
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- public void Close()
|
|
|
-#else
|
|
|
- private void Close()
|
|
|
-#endif
|
|
|
- {
|
|
|
-
|
|
|
- this.isRunning = false;
|
|
|
-
|
|
|
-
|
|
|
- if (this.receiveEventWaitHandle != null)
|
|
|
- this.receiveEventWaitHandle.Set();
|
|
|
-
|
|
|
-
|
|
|
- if (this.inflightWaitHandle != null)
|
|
|
- this.inflightWaitHandle.Set();
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- this.keepAliveEvent.Set();
|
|
|
-#else
|
|
|
-
|
|
|
- this.keepAliveEvent.Set();
|
|
|
-
|
|
|
- if (this.keepAliveEventEnd != null)
|
|
|
- this.keepAliveEventEnd.WaitOne();
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Clear();
|
|
|
- this.internalQueue.Clear();
|
|
|
- this.eventQueue.Clear();
|
|
|
-
|
|
|
-
|
|
|
- this.channel.Close();
|
|
|
-
|
|
|
- this.IsConnected = false;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private MqttMsgPingResp Ping()
|
|
|
- {
|
|
|
- MqttMsgPingReq pingreq = new MqttMsgPingReq();
|
|
|
- try
|
|
|
- {
|
|
|
-
|
|
|
- return (MqttMsgPingResp)this.SendReceive(pingreq, this.keepAlivePeriod);
|
|
|
- }
|
|
|
- catch (Exception e)
|
|
|
- {
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
- return null;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void Connack(MqttMsgConnect connect, byte returnCode, string clientId, bool sessionPresent)
|
|
|
- {
|
|
|
- this.lastCommTime = 0;
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgConnack connack = new MqttMsgConnack();
|
|
|
- connack.ReturnCode = returnCode;
|
|
|
-
|
|
|
- if (this.ProtocolVersion == MqttProtocolVersion.Version_3_1_1)
|
|
|
- connack.SessionPresent = sessionPresent;
|
|
|
-
|
|
|
- this.Send(connack);
|
|
|
-
|
|
|
-
|
|
|
- if (connack.ReturnCode == MqttMsgConnack.CONN_ACCEPTED)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
- this.ClientId = (clientId == null) ? connect.ClientId : clientId;
|
|
|
- this.CleanSession = connect.CleanSession;
|
|
|
- this.WillFlag = connect.WillFlag;
|
|
|
- this.WillTopic = connect.WillTopic;
|
|
|
- this.WillMessage = connect.WillMessage;
|
|
|
- this.WillQosLevel = connect.WillQosLevel;
|
|
|
-
|
|
|
- this.keepAlivePeriod = connect.KeepAlivePeriod * 1000;
|
|
|
-
|
|
|
- this.keepAlivePeriod += (this.keepAlivePeriod / 2);
|
|
|
-
|
|
|
-
|
|
|
- Fx.StartThread(this.KeepAliveThread);
|
|
|
-
|
|
|
- this.isConnectionClosing = false;
|
|
|
- this.IsConnected = true;
|
|
|
- }
|
|
|
-
|
|
|
- else
|
|
|
- {
|
|
|
- this.Close();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void Suback(ushort messageId, byte[] grantedQosLevels)
|
|
|
- {
|
|
|
- MqttMsgSuback suback = new MqttMsgSuback();
|
|
|
- suback.MessageId = messageId;
|
|
|
- suback.GrantedQoSLevels = grantedQosLevels;
|
|
|
-
|
|
|
- this.Send(suback);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void Unsuback(ushort messageId)
|
|
|
- {
|
|
|
- MqttMsgUnsuback unsuback = new MqttMsgUnsuback();
|
|
|
- unsuback.MessageId = messageId;
|
|
|
-
|
|
|
- this.Send(unsuback);
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public ushort Subscribe(string[] topics, byte[] qosLevels)
|
|
|
- {
|
|
|
- MqttMsgSubscribe subscribe =
|
|
|
- new MqttMsgSubscribe(topics, qosLevels);
|
|
|
- subscribe.MessageId = this.GetMessageId();
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInflight(subscribe, MqttMsgFlow.ToPublish);
|
|
|
-
|
|
|
- return subscribe.MessageId;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public ushort Unsubscribe(string[] topics)
|
|
|
- {
|
|
|
- MqttMsgUnsubscribe unsubscribe =
|
|
|
- new MqttMsgUnsubscribe(topics);
|
|
|
- unsubscribe.MessageId = this.GetMessageId();
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInflight(unsubscribe, MqttMsgFlow.ToPublish);
|
|
|
-
|
|
|
- return unsubscribe.MessageId;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public ushort Publish(string topic, byte[] message)
|
|
|
- {
|
|
|
- return this.Publish(topic, message, MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE, false);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public ushort Publish(string topic, byte[] message, byte qosLevel, bool retain)
|
|
|
- {
|
|
|
- MqttMsgPublish publish =
|
|
|
- new MqttMsgPublish(topic, message, false, qosLevel, retain);
|
|
|
- publish.MessageId = this.GetMessageId();
|
|
|
-
|
|
|
-
|
|
|
- bool enqueue = this.EnqueueInflight(publish, MqttMsgFlow.ToPublish);
|
|
|
-
|
|
|
-
|
|
|
- if (enqueue)
|
|
|
- return publish.MessageId;
|
|
|
-
|
|
|
- else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.InflightQueueFull);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnInternalEvent(InternalEvent internalEvent)
|
|
|
- {
|
|
|
- lock (this.eventQueue)
|
|
|
- {
|
|
|
- this.eventQueue.Enqueue(internalEvent);
|
|
|
- }
|
|
|
-
|
|
|
- this.receiveEventWaitHandle.Set();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnConnectionClosing()
|
|
|
- {
|
|
|
- if (!this.isConnectionClosing)
|
|
|
- {
|
|
|
- this.isConnectionClosing = true;
|
|
|
- this.receiveEventWaitHandle.Set();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgPublishReceived(MqttMsgPublish publish)
|
|
|
- {
|
|
|
- if (this.MqttMsgPublishReceived != null)
|
|
|
- {
|
|
|
- this.MqttMsgPublishReceived(this,
|
|
|
- new MqttMsgPublishEventArgs(publish.Topic, publish.Message, publish.DupFlag, publish.QosLevel, publish.Retain));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgPublished(ushort messageId, bool isPublished)
|
|
|
- {
|
|
|
- if (this.MqttMsgPublished != null)
|
|
|
- {
|
|
|
- this.MqttMsgPublished(this,
|
|
|
- new MqttMsgPublishedEventArgs(messageId, isPublished));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgSubscribed(MqttMsgSuback suback)
|
|
|
- {
|
|
|
- if (this.MqttMsgSubscribed != null)
|
|
|
- {
|
|
|
- this.MqttMsgSubscribed(this,
|
|
|
- new MqttMsgSubscribedEventArgs(suback.MessageId, suback.GrantedQoSLevels));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgUnsubscribed(ushort messageId)
|
|
|
- {
|
|
|
- if (this.MqttMsgUnsubscribed != null)
|
|
|
- {
|
|
|
- this.MqttMsgUnsubscribed(this,
|
|
|
- new MqttMsgUnsubscribedEventArgs(messageId));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgSubscribeReceived(ushort messageId, string[] topics, byte[] qosLevels)
|
|
|
- {
|
|
|
- if (this.MqttMsgSubscribeReceived != null)
|
|
|
- {
|
|
|
- this.MqttMsgSubscribeReceived(this,
|
|
|
- new MqttMsgSubscribeEventArgs(messageId, topics, qosLevels));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgUnsubscribeReceived(ushort messageId, string[] topics)
|
|
|
- {
|
|
|
- if (this.MqttMsgUnsubscribeReceived != null)
|
|
|
- {
|
|
|
- this.MqttMsgUnsubscribeReceived(this,
|
|
|
- new MqttMsgUnsubscribeEventArgs(messageId, topics));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgConnected(MqttMsgConnect connect)
|
|
|
- {
|
|
|
- if (this.MqttMsgConnected != null)
|
|
|
- {
|
|
|
- this.ProtocolVersion = (MqttProtocolVersion)connect.ProtocolVersion;
|
|
|
- this.MqttMsgConnected(this, new MqttMsgConnectEventArgs(connect));
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnMqttMsgDisconnected()
|
|
|
- {
|
|
|
- if (this.MqttMsgDisconnected != null)
|
|
|
- {
|
|
|
- this.MqttMsgDisconnected(this, EventArgs.Empty);
|
|
|
- }
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void OnConnectionClosed()
|
|
|
- {
|
|
|
- if (this.ConnectionClosed != null)
|
|
|
- {
|
|
|
- this.ConnectionClosed(this, EventArgs.Empty);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void Send(byte[] msgBytes)
|
|
|
- {
|
|
|
- try
|
|
|
- {
|
|
|
-
|
|
|
- this.channel.Send(msgBytes);
|
|
|
-
|
|
|
-#if !BROKER
|
|
|
-
|
|
|
- this.lastCommTime = Environment.TickCount;
|
|
|
-#endif
|
|
|
- }
|
|
|
- catch (Exception e)
|
|
|
- {
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
|
|
|
-#endif
|
|
|
-
|
|
|
- throw new MqttCommunicationException(e);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void Send(MqttMsgBase msg)
|
|
|
- {
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "SEND {0}", msg);
|
|
|
-#endif
|
|
|
- this.Send(msg.GetBytes((byte)this.ProtocolVersion));
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private MqttMsgBase SendReceive(byte[] msgBytes)
|
|
|
- {
|
|
|
- return this.SendReceive(msgBytes, MqttSettings.MQTT_DEFAULT_TIMEOUT);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private MqttMsgBase SendReceive(byte[] msgBytes, int timeout)
|
|
|
- {
|
|
|
-
|
|
|
- this.syncEndReceiving.Reset();
|
|
|
- try
|
|
|
- {
|
|
|
-
|
|
|
- this.channel.Send(msgBytes);
|
|
|
-
|
|
|
-
|
|
|
- this.lastCommTime = Environment.TickCount;
|
|
|
- }
|
|
|
- catch (Exception e)
|
|
|
- {
|
|
|
-#if !(MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK || WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- if (typeof(SocketException) == e.GetType())
|
|
|
- {
|
|
|
-
|
|
|
- if (((SocketException)e).SocketErrorCode == SocketError.ConnectionReset)
|
|
|
- this.IsConnected = false;
|
|
|
- }
|
|
|
-#endif
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
|
|
|
-#endif
|
|
|
-
|
|
|
- throw new MqttCommunicationException(e);
|
|
|
- }
|
|
|
-
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
-
|
|
|
- if (this.syncEndReceiving.WaitOne(timeout, false))
|
|
|
-#else
|
|
|
-
|
|
|
- if (this.syncEndReceiving.WaitOne(timeout))
|
|
|
-#endif
|
|
|
- {
|
|
|
-
|
|
|
- if (this.exReceiving == null)
|
|
|
- return this.msgReceived;
|
|
|
-
|
|
|
- else
|
|
|
- throw this.exReceiving;
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- throw new MqttCommunicationException();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private MqttMsgBase SendReceive(MqttMsgBase msg)
|
|
|
- {
|
|
|
- return this.SendReceive(msg, MqttSettings.MQTT_DEFAULT_TIMEOUT);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private MqttMsgBase SendReceive(MqttMsgBase msg, int timeout)
|
|
|
- {
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "SEND {0}", msg);
|
|
|
-#endif
|
|
|
- return this.SendReceive(msg.GetBytes((byte)this.ProtocolVersion), timeout);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private bool EnqueueInflight(MqttMsgBase msg, MqttMsgFlow flow)
|
|
|
- {
|
|
|
-
|
|
|
- bool enqueue = true;
|
|
|
-
|
|
|
-
|
|
|
- if ((msg.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (msg.QosLevel == MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE))
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgContextFinder msgCtxFinder = new MqttMsgContextFinder(msg.MessageId, MqttMsgFlow.ToAcknowledge);
|
|
|
- MqttMsgContext msgCtx = (MqttMsgContext)this.inflightQueue.Get(msgCtxFinder.Find);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if (msgCtx != null)
|
|
|
- {
|
|
|
- msgCtx.State = MqttMsgState.QueuedQos2;
|
|
|
- msgCtx.Flow = MqttMsgFlow.ToAcknowledge;
|
|
|
- enqueue = false;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (enqueue)
|
|
|
- {
|
|
|
-
|
|
|
- MqttMsgState state = MqttMsgState.QueuedQos0;
|
|
|
-
|
|
|
-
|
|
|
- switch (msg.QosLevel)
|
|
|
- {
|
|
|
-
|
|
|
- case MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE:
|
|
|
-
|
|
|
- state = MqttMsgState.QueuedQos0;
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE:
|
|
|
-
|
|
|
- state = MqttMsgState.QueuedQos1;
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE:
|
|
|
-
|
|
|
- state = MqttMsgState.QueuedQos2;
|
|
|
- break;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if (msg.Type == MqttMsgBase.MQTT_MSG_SUBSCRIBE_TYPE)
|
|
|
- state = MqttMsgState.SendSubscribe;
|
|
|
- else if (msg.Type == MqttMsgBase.MQTT_MSG_UNSUBSCRIBE_TYPE)
|
|
|
- state = MqttMsgState.SendUnsubscribe;
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgContext msgContext = new MqttMsgContext()
|
|
|
- {
|
|
|
- Message = msg,
|
|
|
- State = state,
|
|
|
- Flow = flow,
|
|
|
- Attempt = 0
|
|
|
- };
|
|
|
-
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
- enqueue = (this.inflightQueue.Count < this.settings.InflightQueueSize);
|
|
|
-
|
|
|
- if (enqueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "enqueued {0}", msg);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- if (msg.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE)
|
|
|
- {
|
|
|
-
|
|
|
- if ((msgContext.Flow == MqttMsgFlow.ToPublish) &&
|
|
|
- ((msg.QosLevel == MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE) ||
|
|
|
- (msg.QosLevel == MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE)))
|
|
|
- {
|
|
|
- if (this.session != null)
|
|
|
- this.session.InflightMessages.Add(msgContext.Key, msgContext);
|
|
|
- }
|
|
|
-
|
|
|
- else if ((msgContext.Flow == MqttMsgFlow.ToAcknowledge) &&
|
|
|
- (msg.QosLevel == MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE))
|
|
|
- {
|
|
|
- if (this.session != null)
|
|
|
- this.session.InflightMessages.Add(msgContext.Key, msgContext);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- this.inflightWaitHandle.Set();
|
|
|
-
|
|
|
- return enqueue;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void EnqueueInternal(MqttMsgBase msg)
|
|
|
- {
|
|
|
-
|
|
|
- bool enqueue = true;
|
|
|
-
|
|
|
-
|
|
|
- if (msg.Type == MqttMsgBase.MQTT_MSG_PUBREL_TYPE)
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgContextFinder msgCtxFinder = new MqttMsgContextFinder(msg.MessageId, MqttMsgFlow.ToAcknowledge);
|
|
|
- MqttMsgContext msgCtx = (MqttMsgContext)this.inflightQueue.Get(msgCtxFinder.Find);
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if (msgCtx == null)
|
|
|
- {
|
|
|
- MqttMsgPubcomp pubcomp = new MqttMsgPubcomp();
|
|
|
- pubcomp.MessageId = msg.MessageId;
|
|
|
-
|
|
|
- this.Send(pubcomp);
|
|
|
-
|
|
|
- enqueue = false;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- else if (msg.Type == MqttMsgBase.MQTT_MSG_PUBCOMP_TYPE)
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgContextFinder msgCtxFinder = new MqttMsgContextFinder(msg.MessageId, MqttMsgFlow.ToPublish);
|
|
|
- MqttMsgContext msgCtx = (MqttMsgContext)this.inflightQueue.Get(msgCtxFinder.Find);
|
|
|
-
|
|
|
-
|
|
|
- if (msgCtx == null)
|
|
|
- {
|
|
|
- enqueue = false;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- else if (msg.Type == MqttMsgBase.MQTT_MSG_PUBREC_TYPE)
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgContextFinder msgCtxFinder = new MqttMsgContextFinder(msg.MessageId, MqttMsgFlow.ToPublish);
|
|
|
- MqttMsgContext msgCtx = (MqttMsgContext)this.inflightQueue.Get(msgCtxFinder.Find);
|
|
|
-
|
|
|
-
|
|
|
- if (msgCtx == null)
|
|
|
- {
|
|
|
- enqueue = false;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- if (enqueue)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
- this.internalQueue.Enqueue(msg);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "enqueued {0}", msg);
|
|
|
-#endif
|
|
|
- this.inflightWaitHandle.Set();
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void ReceiveThread()
|
|
|
- {
|
|
|
- int readBytes = 0;
|
|
|
- byte[] fixedHeaderFirstByte = new byte[1];
|
|
|
- byte msgType;
|
|
|
-
|
|
|
- while (this.isRunning)
|
|
|
- {
|
|
|
- try
|
|
|
- {
|
|
|
-
|
|
|
- readBytes = this.channel.Receive(fixedHeaderFirstByte);
|
|
|
-
|
|
|
- if (readBytes > 0)
|
|
|
- {
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- this.lastCommTime = Environment.TickCount;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- msgType = (byte)((fixedHeaderFirstByte[0] & MqttMsgBase.MSG_TYPE_MASK) >> MqttMsgBase.MSG_TYPE_OFFSET);
|
|
|
-
|
|
|
- switch (msgType)
|
|
|
- {
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_CONNECT_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgConnect connect = MqttMsgConnect.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- Trace.WriteLine(TraceLevel.Frame, "RECV {0}", connect);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(new MsgInternalEvent(connect));
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_CONNACK_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#else
|
|
|
- this.msgReceived = MqttMsgConnack.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", this.msgReceived);
|
|
|
-#endif
|
|
|
- this.syncEndReceiving.Set();
|
|
|
- break;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PINGREQ_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- this.msgReceived = MqttMsgPingReq.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- Trace.WriteLine(TraceLevel.Frame, "RECV {0}", this.msgReceived);
|
|
|
-#endif
|
|
|
-
|
|
|
- MqttMsgPingResp pingresp = new MqttMsgPingResp();
|
|
|
- this.Send(pingresp);
|
|
|
-
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PINGRESP_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#else
|
|
|
- this.msgReceived = MqttMsgPingResp.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", this.msgReceived);
|
|
|
-#endif
|
|
|
- this.syncEndReceiving.Set();
|
|
|
- break;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_SUBSCRIBE_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgSubscribe subscribe = MqttMsgSubscribe.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- Trace.WriteLine(TraceLevel.Frame, "RECV {0}", subscribe);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(new MsgInternalEvent(subscribe));
|
|
|
-
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_SUBACK_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#else
|
|
|
-
|
|
|
- MqttMsgSuback suback = MqttMsgSuback.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", suback);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(suback);
|
|
|
-
|
|
|
- break;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBLISH_TYPE:
|
|
|
-
|
|
|
- MqttMsgPublish publish = MqttMsgPublish.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", publish);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInflight(publish, MqttMsgFlow.ToAcknowledge);
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBACK_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgPuback puback = MqttMsgPuback.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", puback);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(puback);
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBREC_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgPubrec pubrec = MqttMsgPubrec.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", pubrec);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(pubrec);
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBREL_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgPubrel pubrel = MqttMsgPubrel.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", pubrel);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(pubrel);
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBCOMP_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- MqttMsgPubcomp pubcomp = MqttMsgPubcomp.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", pubcomp);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(pubcomp);
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_UNSUBSCRIBE_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgUnsubscribe unsubscribe = MqttMsgUnsubscribe.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- Trace.WriteLine(TraceLevel.Frame, "RECV {0}", unsubscribe);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(new MsgInternalEvent(unsubscribe));
|
|
|
-
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_UNSUBACK_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#else
|
|
|
-
|
|
|
- MqttMsgUnsuback unsuback = MqttMsgUnsuback.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Frame, "RECV {0}", unsuback);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.EnqueueInternal(unsuback);
|
|
|
-
|
|
|
- break;
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgDisconnect.MQTT_MSG_DISCONNECT_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgDisconnect disconnect = MqttMsgDisconnect.Parse(fixedHeaderFirstByte[0], (byte)this.ProtocolVersion, this.channel);
|
|
|
-#if TRACE
|
|
|
- Trace.WriteLine(TraceLevel.Frame, "RECV {0}", disconnect);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(new MsgInternalEvent(disconnect));
|
|
|
-
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
- default:
|
|
|
-
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
- }
|
|
|
-
|
|
|
- this.exReceiving = null;
|
|
|
- }
|
|
|
-
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
- }
|
|
|
- }
|
|
|
- catch (Exception e)
|
|
|
- {
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
|
|
|
-#endif
|
|
|
- this.exReceiving = new MqttCommunicationException(e);
|
|
|
-
|
|
|
- bool close = false;
|
|
|
- if (e.GetType() == typeof(MqttClientException))
|
|
|
- {
|
|
|
-
|
|
|
- MqttClientException ex = e as MqttClientException;
|
|
|
- close = ((ex.ErrorCode == MqttClientErrorCode.InvalidFlagBits) ||
|
|
|
- (ex.ErrorCode == MqttClientErrorCode.InvalidProtocolName) ||
|
|
|
- (ex.ErrorCode == MqttClientErrorCode.InvalidConnectFlags));
|
|
|
- }
|
|
|
-#if !(WINDOWS_APP || WINDOWS_PHONE_APP || (!UNITY_EDITOR&&UNITY_WSA_10_0&&!ENABLE_IL2CPP))
|
|
|
- else if ((e.GetType() == typeof(IOException)) || (e.GetType() == typeof(SocketException)) ||
|
|
|
- ((e.InnerException != null) && (e.InnerException.GetType() == typeof(SocketException))))
|
|
|
- {
|
|
|
- close = true;
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
- if (close)
|
|
|
- {
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void KeepAliveThread()
|
|
|
- {
|
|
|
- int delta = 0;
|
|
|
- int wait = this.keepAlivePeriod;
|
|
|
-
|
|
|
-
|
|
|
- this.keepAliveEventEnd = new AutoResetEvent(false);
|
|
|
-
|
|
|
- while (this.isRunning)
|
|
|
- {
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
-
|
|
|
- this.keepAliveEvent.WaitOne(wait, false);
|
|
|
-#else
|
|
|
-
|
|
|
- this.keepAliveEvent.WaitOne(wait);
|
|
|
-#endif
|
|
|
-
|
|
|
- if (this.isRunning)
|
|
|
- {
|
|
|
- delta = Environment.TickCount - this.lastCommTime;
|
|
|
-
|
|
|
-
|
|
|
- if (delta >= this.keepAlivePeriod)
|
|
|
- {
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
-#else
|
|
|
-
|
|
|
- this.Ping();
|
|
|
- wait = this.keepAlivePeriod;
|
|
|
-#endif
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- wait = this.keepAlivePeriod - delta;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- this.keepAliveEventEnd.Set();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void DispatchEventThread()
|
|
|
- {
|
|
|
- while (this.isRunning)
|
|
|
- {
|
|
|
-#if BROKER
|
|
|
- if ((this.eventQueue.Count == 0) && !this.isConnectionClosing)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
- if (!this.IsConnected)
|
|
|
- {
|
|
|
-
|
|
|
- if (!this.receiveEventWaitHandle.WaitOne(this.settings.TimeoutOnConnection))
|
|
|
- {
|
|
|
-
|
|
|
- this.Close();
|
|
|
-
|
|
|
-
|
|
|
- this.OnConnectionClosed();
|
|
|
- }
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.receiveEventWaitHandle.WaitOne();
|
|
|
- }
|
|
|
- }
|
|
|
-#else
|
|
|
- if ((this.eventQueue.Count == 0) && !this.isConnectionClosing)
|
|
|
-
|
|
|
- this.receiveEventWaitHandle.WaitOne();
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- if (this.isRunning)
|
|
|
- {
|
|
|
-
|
|
|
- InternalEvent internalEvent = null;
|
|
|
- lock (this.eventQueue)
|
|
|
- {
|
|
|
- if (this.eventQueue.Count > 0)
|
|
|
- internalEvent = (InternalEvent)this.eventQueue.Dequeue();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (internalEvent != null)
|
|
|
- {
|
|
|
- MqttMsgBase msg = ((MsgInternalEvent)internalEvent).Message;
|
|
|
-
|
|
|
- if (msg != null)
|
|
|
- {
|
|
|
- switch (msg.Type)
|
|
|
- {
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_CONNECT_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- this.OnMqttMsgConnected((MqttMsgConnect)msg);
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_SUBSCRIBE_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgSubscribe subscribe = (MqttMsgSubscribe)msg;
|
|
|
-
|
|
|
- this.OnMqttMsgSubscribeReceived(subscribe.MessageId, subscribe.Topics, subscribe.QoSLevels);
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_SUBACK_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- this.OnMqttMsgSubscribed((MqttMsgSuback)msg);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBLISH_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- if (internalEvent.GetType() == typeof(MsgPublishedInternalEvent))
|
|
|
- this.OnMqttMsgPublished(msg.MessageId, false);
|
|
|
- else
|
|
|
-
|
|
|
- this.OnMqttMsgPublishReceived((MqttMsgPublish)msg);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBACK_TYPE:
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- this.OnMqttMsgPublished(msg.MessageId, true);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBREL_TYPE:
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- this.OnMqttMsgPublishReceived((MqttMsgPublish)msg);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_PUBCOMP_TYPE:
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- this.OnMqttMsgPublished(msg.MessageId, true);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_UNSUBSCRIBE_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
- MqttMsgUnsubscribe unsubscribe = (MqttMsgUnsubscribe)msg;
|
|
|
-
|
|
|
- this.OnMqttMsgUnsubscribeReceived(unsubscribe.MessageId, unsubscribe.Topics);
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgBase.MQTT_MSG_UNSUBACK_TYPE:
|
|
|
-
|
|
|
-
|
|
|
- this.OnMqttMsgUnsubscribed(msg.MessageId);
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- case MqttMsgDisconnect.MQTT_MSG_DISCONNECT_TYPE:
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
- this.OnMqttMsgDisconnected();
|
|
|
- break;
|
|
|
-#else
|
|
|
- throw new MqttClientException(MqttClientErrorCode.WrongBrokerMessage);
|
|
|
-#endif
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if ((this.eventQueue.Count == 0) && this.isConnectionClosing)
|
|
|
- {
|
|
|
-
|
|
|
- this.Close();
|
|
|
-
|
|
|
-
|
|
|
- this.OnConnectionClosed();
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void ProcessInflightThread()
|
|
|
- {
|
|
|
- MqttMsgContext msgContext = null;
|
|
|
- MqttMsgBase msgInflight = null;
|
|
|
- MqttMsgBase msgReceived = null;
|
|
|
- InternalEvent internalEvent = null;
|
|
|
- bool acknowledge = false;
|
|
|
- int timeout = Timeout.Infinite;
|
|
|
- int delta;
|
|
|
- bool msgReceivedProcessed = false;
|
|
|
-
|
|
|
- try
|
|
|
- {
|
|
|
- while (this.isRunning)
|
|
|
- {
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
-
|
|
|
- this.inflightWaitHandle.WaitOne(timeout, false);
|
|
|
-#else
|
|
|
-
|
|
|
- this.inflightWaitHandle.WaitOne(timeout);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- if (this.isRunning)
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- msgReceivedProcessed = false;
|
|
|
- acknowledge = false;
|
|
|
- msgReceived = null;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- timeout = Int32.MaxValue;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- int count = this.inflightQueue.Count;
|
|
|
-
|
|
|
- while (count > 0)
|
|
|
- {
|
|
|
- count--;
|
|
|
- acknowledge = false;
|
|
|
- msgReceived = null;
|
|
|
-
|
|
|
-
|
|
|
- if (!this.isRunning)
|
|
|
- break;
|
|
|
-
|
|
|
-
|
|
|
- msgContext = (MqttMsgContext)this.inflightQueue.Dequeue();
|
|
|
-
|
|
|
-
|
|
|
- msgInflight = (MqttMsgBase)msgContext.Message;
|
|
|
-
|
|
|
- switch (msgContext.State)
|
|
|
- {
|
|
|
- case MqttMsgState.QueuedQos0:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- this.Send(msgInflight);
|
|
|
- }
|
|
|
-
|
|
|
- else if (msgContext.Flow == MqttMsgFlow.ToAcknowledge)
|
|
|
- {
|
|
|
- internalEvent = new MsgInternalEvent(msgInflight);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
- }
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "processed {0}", msgInflight);
|
|
|
-#endif
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.QueuedQos1:
|
|
|
-
|
|
|
- case MqttMsgState.SendSubscribe:
|
|
|
- case MqttMsgState.SendUnsubscribe:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- msgContext.Timestamp = Environment.TickCount;
|
|
|
- msgContext.Attempt++;
|
|
|
-
|
|
|
- if (msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE)
|
|
|
- {
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForPuback;
|
|
|
-
|
|
|
- if (msgContext.Attempt > 1)
|
|
|
- msgInflight.DupFlag = true;
|
|
|
- }
|
|
|
- else if (msgInflight.Type == MqttMsgBase.MQTT_MSG_SUBSCRIBE_TYPE)
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForSuback;
|
|
|
- else if (msgInflight.Type == MqttMsgBase.MQTT_MSG_UNSUBSCRIBE_TYPE)
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForUnsuback;
|
|
|
-
|
|
|
- this.Send(msgInflight);
|
|
|
-
|
|
|
-
|
|
|
- timeout = (this.settings.DelayOnRetry < timeout) ? this.settings.DelayOnRetry : timeout;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
-
|
|
|
- else if (msgContext.Flow == MqttMsgFlow.ToAcknowledge)
|
|
|
- {
|
|
|
- MqttMsgPuback puback = new MqttMsgPuback();
|
|
|
- puback.MessageId = msgInflight.MessageId;
|
|
|
-
|
|
|
- this.Send(puback);
|
|
|
-
|
|
|
- internalEvent = new MsgInternalEvent(msgInflight);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "processed {0}", msgInflight);
|
|
|
-#endif
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.QueuedQos2:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- msgContext.Timestamp = Environment.TickCount;
|
|
|
- msgContext.Attempt++;
|
|
|
- msgContext.State = MqttMsgState.WaitForPubrec;
|
|
|
-
|
|
|
- if (msgContext.Attempt > 1)
|
|
|
- msgInflight.DupFlag = true;
|
|
|
-
|
|
|
- this.Send(msgInflight);
|
|
|
-
|
|
|
-
|
|
|
- timeout = (this.settings.DelayOnRetry < timeout) ? this.settings.DelayOnRetry : timeout;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
-
|
|
|
- else if (msgContext.Flow == MqttMsgFlow.ToAcknowledge)
|
|
|
- {
|
|
|
- MqttMsgPubrec pubrec = new MqttMsgPubrec();
|
|
|
- pubrec.MessageId = msgInflight.MessageId;
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForPubrel;
|
|
|
-
|
|
|
- this.Send(pubrec);
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.WaitForPuback:
|
|
|
- case MqttMsgState.WaitForSuback:
|
|
|
- case MqttMsgState.WaitForUnsuback:
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- acknowledge = false;
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
- if (this.internalQueue.Count > 0)
|
|
|
- msgReceived = (MqttMsgBase)this.internalQueue.Peek();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (msgReceived != null)
|
|
|
- {
|
|
|
-
|
|
|
- if (((msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBACK_TYPE) && (msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) && (msgReceived.MessageId == msgInflight.MessageId)) ||
|
|
|
- ((msgReceived.Type == MqttMsgBase.MQTT_MSG_SUBACK_TYPE) && (msgInflight.Type == MqttMsgBase.MQTT_MSG_SUBSCRIBE_TYPE) && (msgReceived.MessageId == msgInflight.MessageId)) ||
|
|
|
- ((msgReceived.Type == MqttMsgBase.MQTT_MSG_UNSUBACK_TYPE) && (msgInflight.Type == MqttMsgBase.MQTT_MSG_UNSUBSCRIBE_TYPE) && (msgReceived.MessageId == msgInflight.MessageId)))
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.internalQueue.Dequeue();
|
|
|
- acknowledge = true;
|
|
|
- msgReceivedProcessed = true;
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0}", msgReceived);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBACK_TYPE)
|
|
|
- internalEvent = new MsgPublishedInternalEvent(msgReceived, true);
|
|
|
- else
|
|
|
- internalEvent = new MsgInternalEvent(msgReceived);
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
-
|
|
|
-
|
|
|
- if ((msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "processed {0}", msgInflight);
|
|
|
-#endif
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (!acknowledge)
|
|
|
- {
|
|
|
- delta = Environment.TickCount - msgContext.Timestamp;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if (delta >= this.settings.DelayOnRetry)
|
|
|
- {
|
|
|
-
|
|
|
- if (msgContext.Attempt < this.settings.AttemptsOnRetry)
|
|
|
- {
|
|
|
- msgContext.State = MqttMsgState.QueuedQos1;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- timeout = 0;
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- if (msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE)
|
|
|
- {
|
|
|
-
|
|
|
- if ((this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
- internalEvent = new MsgPublishedInternalEvent(msgInflight, false);
|
|
|
-
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- }
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- int msgTimeout = (this.settings.DelayOnRetry - delta);
|
|
|
- timeout = (msgTimeout < timeout) ? msgTimeout : timeout;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.WaitForPubrec:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- acknowledge = false;
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
- if (this.internalQueue.Count > 0)
|
|
|
- msgReceived = (MqttMsgBase)this.internalQueue.Peek();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if ((msgReceived != null) && (msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBREC_TYPE))
|
|
|
- {
|
|
|
-
|
|
|
- if (msgReceived.MessageId == msgInflight.MessageId)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.internalQueue.Dequeue();
|
|
|
- acknowledge = true;
|
|
|
- msgReceivedProcessed = true;
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0}", msgReceived);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
- MqttMsgPubrel pubrel = new MqttMsgPubrel();
|
|
|
- pubrel.MessageId = msgInflight.MessageId;
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForPubcomp;
|
|
|
- msgContext.Timestamp = Environment.TickCount;
|
|
|
- msgContext.Attempt = 1;
|
|
|
-
|
|
|
- this.Send(pubrel);
|
|
|
-
|
|
|
-
|
|
|
- timeout = (this.settings.DelayOnRetry < timeout) ? this.settings.DelayOnRetry : timeout;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (!acknowledge)
|
|
|
- {
|
|
|
- delta = Environment.TickCount - msgContext.Timestamp;
|
|
|
-
|
|
|
- if (delta >= this.settings.DelayOnRetry)
|
|
|
- {
|
|
|
-
|
|
|
- if (msgContext.Attempt < this.settings.AttemptsOnRetry)
|
|
|
- {
|
|
|
- msgContext.State = MqttMsgState.QueuedQos2;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- timeout = 0;
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- if ((this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- internalEvent = new MsgPublishedInternalEvent(msgInflight, false);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
- }
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- int msgTimeout = (this.settings.DelayOnRetry - delta);
|
|
|
- timeout = (msgTimeout < timeout) ? msgTimeout : timeout;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.WaitForPubrel:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToAcknowledge)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
- if (this.internalQueue.Count > 0)
|
|
|
- msgReceived = (MqttMsgBase)this.internalQueue.Peek();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if ((msgReceived != null) && (msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBREL_TYPE))
|
|
|
- {
|
|
|
-
|
|
|
- if (msgReceived.MessageId == msgInflight.MessageId)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.internalQueue.Dequeue();
|
|
|
- msgReceivedProcessed = true;
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0}", msgReceived);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
- MqttMsgPubcomp pubcomp = new MqttMsgPubcomp();
|
|
|
- pubcomp.MessageId = msgInflight.MessageId;
|
|
|
-
|
|
|
- this.Send(pubcomp);
|
|
|
-
|
|
|
- internalEvent = new MsgInternalEvent(msgInflight);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
-
|
|
|
-
|
|
|
- if ((msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "processed {0}", msgInflight);
|
|
|
-#endif
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.WaitForPubcomp:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- acknowledge = false;
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
- if (this.internalQueue.Count > 0)
|
|
|
- msgReceived = (MqttMsgBase)this.internalQueue.Peek();
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if ((msgReceived != null) && (msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBCOMP_TYPE))
|
|
|
- {
|
|
|
-
|
|
|
- if (msgReceived.MessageId == msgInflight.MessageId)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.internalQueue.Dequeue();
|
|
|
- acknowledge = true;
|
|
|
- msgReceivedProcessed = true;
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0}", msgReceived);
|
|
|
-#endif
|
|
|
- }
|
|
|
-
|
|
|
- internalEvent = new MsgPublishedInternalEvent(msgReceived, true);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
-
|
|
|
-
|
|
|
- if ((msgInflight.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "processed {0}", msgInflight);
|
|
|
-#endif
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- else if ((msgReceived != null) && (msgReceived.Type == MqttMsgBase.MQTT_MSG_PUBREC_TYPE))
|
|
|
- {
|
|
|
-
|
|
|
-
|
|
|
- if (msgReceived.MessageId == msgInflight.MessageId)
|
|
|
- {
|
|
|
- lock (this.internalQueue)
|
|
|
- {
|
|
|
-
|
|
|
- this.internalQueue.Dequeue();
|
|
|
- acknowledge = true;
|
|
|
- msgReceivedProcessed = true;
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0}", msgReceived);
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (!acknowledge)
|
|
|
- {
|
|
|
- delta = Environment.TickCount - msgContext.Timestamp;
|
|
|
-
|
|
|
- if (delta >= this.settings.DelayOnRetry)
|
|
|
- {
|
|
|
-
|
|
|
- if (msgContext.Attempt < this.settings.AttemptsOnRetry)
|
|
|
- {
|
|
|
- msgContext.State = MqttMsgState.SendPubrel;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- timeout = 0;
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- if ((this.session != null) &&
|
|
|
-#if (MF_FRAMEWORK_VERSION_V4_2 || MF_FRAMEWORK_VERSION_V4_3 || COMPACT_FRAMEWORK)
|
|
|
- (this.session.InflightMessages.Contains(msgContext.Key)))
|
|
|
-#else
|
|
|
- (this.session.InflightMessages.ContainsKey(msgContext.Key)))
|
|
|
-#endif
|
|
|
- {
|
|
|
- this.session.InflightMessages.Remove(msgContext.Key);
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- internalEvent = new MsgPublishedInternalEvent(msgInflight, false);
|
|
|
-
|
|
|
- this.OnInternalEvent(internalEvent);
|
|
|
- }
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- int msgTimeout = (this.settings.DelayOnRetry - delta);
|
|
|
- timeout = (msgTimeout < timeout) ? msgTimeout : timeout;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.SendPubrec:
|
|
|
-
|
|
|
-
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.SendPubrel:
|
|
|
-
|
|
|
-
|
|
|
- if (msgContext.Flow == MqttMsgFlow.ToPublish)
|
|
|
- {
|
|
|
- MqttMsgPubrel pubrel = new MqttMsgPubrel();
|
|
|
- pubrel.MessageId = msgInflight.MessageId;
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.WaitForPubcomp;
|
|
|
- msgContext.Timestamp = Environment.TickCount;
|
|
|
- msgContext.Attempt++;
|
|
|
-
|
|
|
- if (this.ProtocolVersion == MqttProtocolVersion.Version_3_1)
|
|
|
- {
|
|
|
- if (msgContext.Attempt > 1)
|
|
|
- pubrel.DupFlag = true;
|
|
|
- }
|
|
|
-
|
|
|
- this.Send(pubrel);
|
|
|
-
|
|
|
-
|
|
|
- timeout = (this.settings.DelayOnRetry < timeout) ? this.settings.DelayOnRetry : timeout;
|
|
|
-
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
- }
|
|
|
- break;
|
|
|
-
|
|
|
- case MqttMsgState.SendPubcomp:
|
|
|
-
|
|
|
- break;
|
|
|
- case MqttMsgState.SendPuback:
|
|
|
-
|
|
|
- break;
|
|
|
- default:
|
|
|
- break;
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- if (timeout == Int32.MaxValue)
|
|
|
- timeout = Timeout.Infinite;
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- if ((msgReceived != null) && !msgReceivedProcessed)
|
|
|
- {
|
|
|
- this.internalQueue.Dequeue();
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Queuing, "dequeued {0} orphan", msgReceived);
|
|
|
-#endif
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- catch (MqttCommunicationException e)
|
|
|
- {
|
|
|
-
|
|
|
- if (msgContext != null)
|
|
|
-
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-#if TRACE
|
|
|
- MqttUtility.Trace.WriteLine(TraceLevel.Error, "Exception occurred: {0}", e.ToString());
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
- this.OnConnectionClosing();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private void RestoreSession()
|
|
|
- {
|
|
|
-
|
|
|
- if (!this.CleanSession)
|
|
|
- {
|
|
|
-
|
|
|
- if (this.session != null)
|
|
|
- {
|
|
|
- lock (this.inflightQueue)
|
|
|
- {
|
|
|
- foreach (MqttMsgContext msgContext in this.session.InflightMessages.Values)
|
|
|
- {
|
|
|
- this.inflightQueue.Enqueue(msgContext);
|
|
|
-
|
|
|
-
|
|
|
- if ((msgContext.Message.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (msgContext.Flow == MqttMsgFlow.ToPublish))
|
|
|
- {
|
|
|
-
|
|
|
- if ((msgContext.Message.QosLevel == MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE) &&
|
|
|
- (msgContext.State == MqttMsgState.WaitForPuback))
|
|
|
- {
|
|
|
-
|
|
|
- msgContext.State = MqttMsgState.QueuedQos1;
|
|
|
- }
|
|
|
-
|
|
|
- else if (msgContext.Message.QosLevel == MqttMsgBase.QOS_LEVEL_EXACTLY_ONCE)
|
|
|
- {
|
|
|
-
|
|
|
- if (msgContext.State == MqttMsgState.WaitForPubrec)
|
|
|
- {
|
|
|
- msgContext.State = MqttMsgState.QueuedQos2;
|
|
|
- }
|
|
|
-
|
|
|
- else if (msgContext.State == MqttMsgState.WaitForPubcomp)
|
|
|
- {
|
|
|
- msgContext.State = MqttMsgState.SendPubrel;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
- this.inflightWaitHandle.Set();
|
|
|
- }
|
|
|
- else
|
|
|
- {
|
|
|
-
|
|
|
- this.session = new MqttClientSession(this.ClientId);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- else
|
|
|
- {
|
|
|
- if (this.session != null)
|
|
|
- this.session.Clear();
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-#if BROKER
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public void LoadSession(MqttClientSession session)
|
|
|
- {
|
|
|
-
|
|
|
- if (!this.CleanSession)
|
|
|
- {
|
|
|
-
|
|
|
- this.session = session;
|
|
|
-
|
|
|
- this.RestoreSession();
|
|
|
- }
|
|
|
- }
|
|
|
-#endif
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- private ushort GetMessageId()
|
|
|
- {
|
|
|
-
|
|
|
- this.messageIdCounter = ((this.messageIdCounter % UInt16.MaxValue) != 0) ? (ushort)(this.messageIdCounter + 1) : (ushort)1;
|
|
|
- return this.messageIdCounter;
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- internal class MqttMsgContextFinder
|
|
|
- {
|
|
|
-
|
|
|
- internal ushort MessageId { get; set; }
|
|
|
-
|
|
|
- internal MqttMsgFlow Flow { get; set; }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- internal MqttMsgContextFinder(ushort messageId, MqttMsgFlow flow)
|
|
|
- {
|
|
|
- this.MessageId = messageId;
|
|
|
- this.Flow = flow;
|
|
|
- }
|
|
|
-
|
|
|
- internal bool Find(object item)
|
|
|
- {
|
|
|
- MqttMsgContext msgCtx = (MqttMsgContext)item;
|
|
|
- return ((msgCtx.Message.Type == MqttMsgBase.MQTT_MSG_PUBLISH_TYPE) &&
|
|
|
- (msgCtx.Message.MessageId == this.MessageId) &&
|
|
|
- msgCtx.Flow == this.Flow);
|
|
|
-
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
- public enum MqttProtocolVersion
|
|
|
- {
|
|
|
- Version_3_1 = MqttMsgConnect.PROTOCOL_VERSION_V3_1,
|
|
|
- Version_3_1_1 = MqttMsgConnect.PROTOCOL_VERSION_V3_1_1
|
|
|
- }
|
|
|
-}
|