context: using System;
  using System.Collections.Generic;
  using MqttClient.Core.Messages;
  using MqttClient.Core;
  using MqttClient.Logger;
  using MqttClient.MatchTree;
  
  namespace MqttClient
  {
      internal class Mqtt : IMqtt, IMqttShared
      {
  
          #region Member Variables
  
          private StreamManager manager = null;
          private QoSManager qosManager = null;
          private IPersistence _store = null;
  
          private TopicTree<PublishArrivedDelegate> topicTree = null;
  
          private string _clientID;
          private string _username;
          private string _password;
          private ushort _keepAlive = 30;
          private ushort messageID = 1;
          private Timer keepAliveTimer = null;
  
          #endregion
  
          #region IMqtt Members
  
          public ushort KeepAliveInterval
          {
              get { return _keepAlive; }
              set { _keepAlive = value; }
          }
  
          public long ResendInterval
          {
              get { return qosManager.ResendInterval; }
              set { qosManager.ResendInterval = value; }
          }
  
          public Mqtt(string connString, string clientID, string username, string password, IPersistence store)
          {
              _store = store;
              qosManager = new QoSManager(_store);
              manager = new StreamManager(connString, qosManager);
              qosManager.MessageReceived += new QoSManager.MessageReceivedDelegate(qosManager_MessageReceived);
              _clientID = clientID;
              _username = username;
              _password = password;
          }
  
          void tmrCallback(object args)
          {
              try
              {
                  manager.SendMessage(new MqttPingReqMessage());
              }
              catch (Exception e)
              {
                  Log.Write(LogLevel.ERROR, e.ToString());
                  // We've probably lost a connection. The time will be cancelled when we are
                  // notified by the stream manager. In the mean time, just ignore the Exception.
              }
          }
  
          void qosManager_MessageReceived(object sender, MqttMessageReceivedEventArgs e)
          {
              if (e.Message == null)
              {
                  //a null message means we have disconnected from the broker
                  OnConnectionLost(new EventArgs());
                  return;
              }
  
              switch (e.Message.MsgType)
              {
                  case MessageType.CONNACK:
                      var connack = ((MqttConnackMessage)e.Message);
                      if (connack.Response == MqttConnectionResponse.Accepted)
                          OnConnected(new EventArgs());
                      else
                          OnConnectionLost(new MqttConnackEventArgs(connack.Response));
                      break;
                  case MessageType.DISCONNECT:
                      break;
                  case MessageType.PINGREQ:
                      manager.SendMessage(new MqttPingRespMessage());
                      break;
                  case MessageType.PUBACK:
                      MqttPubackMessage puback = (MqttPubackMessage)e.Message;
                      OnPublished(new CompleteArgs(puback.AckID));
                      break;
                  case MessageType.PUBCOMP:
                      break;
                  case MessageType.PUBLISH:
                      MqttPublishMessage m = (MqttPublishMessage)e.Message;
                      OnPublishArrived(m);
                      break;
                  case MessageType.PUBREC:
                      break;
                  case MessageType.PUBREL:
                      MqttPubrelMessage pubrel = (MqttPubrelMessage)e.Message;
                      OnPublished(new CompleteArgs(pubrel.AckID));
                      break;
                  case MessageType.SUBACK:
                      MqttSubackMessage m1 = (MqttSubackMessage)e.Message;
                      OnSubscribed(new CompleteArgs(m1.AckID));
                      break;
                  case MessageType.UNSUBACK:
                      MqttUnsubackMessage m2 = (MqttUnsubackMessage)e.Message;
                      OnUnsubscribed(new CompleteArgs(m2.AckID));
                      break;
                  case MessageType.PINGRESP:
                      break;
                  case MessageType.UNSUBSCRIBE:
                  case MessageType.CONNECT:
                  case MessageType.SUBSCRIBE:
                  default:
                      throw new Exception("Unsupported Message Type");
  
              }
          }
  
          public void Connect()
          {
              DoConnect(new MqttConnectMessage(
                _clientID, _username, _password, _keepAlive, false
              ));
          }
  
          public void Connect(string willTopic, QoS willQoS, MqttPayload willMsg, bool willRetain)
          {
              DoConnect(new MqttConnectMessage(
                _clientID, _username, _password, _keepAlive, willTopic, willMsg.TrimmedBuffer, willQoS, willRetain, false
              ));
          }
  
          public void Connect(bool cleanStart)
          {
              DoConnect(new MqttConnectMessage(
                _clientID, _username, _password, _keepAlive, cleanStart
              ));
          }
  
          public void Connect(string willTopic, QoS willQoS, MqttPayload willMsg, bool willRetain, bool cleanStart)
          {
              DoConnect(new MqttConnectMessage(
                _clientID, _username, _password, _keepAlive, willTopic, willMsg.TrimmedBuffer, willQoS, willRetain, cleanStart
              ));
          }
  
          private void DoConnect(MqttConnectMessage conmsg)
          {
              try
              {
                  manager.Connect();
                  manager.SendMessage(conmsg);
                  manager.WaitForResponse();
                  TimerCallback callback = new TimerCallback(tmrCallback);
                  // TODO: Set Keep Alive interval and keepAlive time as property of client
                  int keepAliveInterval = 1000 * _keepAlive;
                  keepAliveTimer = new Timer(callback, null, keepAliveInterval, keepAliveInterval);
              }
              catch (Exception e)
              {
                  throw new MqttBrokerUnavailableException("Unable to connect to the broker", e);
              }
          }
  
          public void Disconnect()
          {
              manager.SendMessage(new MqttDisconnectMessage());
              if (keepAliveTimer != null)
              {
                  keepAliveTimer.Dispose();
                  keepAliveTimer = null;
              }
              manager.Disconnect();
  
          }
  
          public int Publish(string topic, MqttPayload payload, QoS qos, bool retained)
          {
              if (manager.IsConnected)
              {
                  // Reset the PINGREQ timer as this publish will reset the server's counter
                  if (keepAliveTimer != null)
                  {
                      int kmillis = 1000 * _keepAlive;
                      keepAliveTimer.Change(kmillis, kmillis);
                  }
                  ushort messID = MessageID;
                  manager.SendMessage(new MqttPublishMessage(messID, topic, payload.TrimmedBuffer, qos, retained));
                  return messID;
              }
              else
              {
                  throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish");
              }
          }
  
          public int Publish(MqttParcel parcel)
          {
              return Publish(parcel.Topic, parcel.Payload, parcel.Qos, parcel.Retained);
          }
  
          public int Subscribe(Subscription[] subscriptions)
          {
              if (manager.IsConnected)
              {
                  ushort messID = MessageID;
                  manager.SendMessage(new MqttSubscribeMessage(messID, subscriptions));
                  return messID;
              }
              else
              {
                  throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish");
              }
          }
  
          public int Subscribe(Subscription subscription)
          {
              return Subscribe(new Subscription[] { subscription });
          }
  
          public int Subscribe(string topic, QoS qos)
          {
              return Subscribe(new Subscription(topic, qos));
          }
  
          public int Unsubscribe(string[] topics)
          {
              if (manager.IsConnected)
              {
                  ushort messID = MessageID;
                  manager.SendMessage(new MqttUnsubscribeMessage(messID, topics));
                  return messID;
              }
              else
              {
                  throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish");
              }
          }
  
          public bool IsConnected
          {
              get
              {
                  return manager.IsConnected;
              }
          }
  
          public event PublishArrivedDelegate PublishArrived;
  
          public event CompleteDelegate Published;
  
          public event CompleteDelegate Subscribed;
  
          public event CompleteDelegate Unsubscribed;
  
          public event ConnectionDelegate ConnectionLost;
  
          public event ConnectionDelegate Connected;
  
          #endregion
  
          #region IMqttSharedSubscriber Members
  
          public void Subscribe(Subscription subscription, PublishArrivedDelegate subscriber)
          {
              if (topicTree == null)
              {
  
gt:                 topicTree = new TopicTree<PublishArrivedDelegate>();