|
@@ -0,0 +1,2638 @@
|
|
|
+
|
|
|
+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;
|
|
|
+using UnityEngine;
|
|
|
+
|
|
|
+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
|
|
|
+ }
|
|
|
+}
|