<commit_before>using System;
using System.Collections.Generic;
using System.Text;
using MqttLib.Core.Messages;
using MqttLib.Core;
using MqttLib;
//using System.Timers;
using System.Threading;
using System.Diagnostics;
using MqttLib.Logger;
using MqttLib.MatchTree;

namespace MqttLib
{
    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 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, IPersistence store)
        {
            _store = store;
            qosManager = new QoSManager(_store);
            manager = new StreamManager(connString, qosManager);
            qosManager.MessageReceived += new QoSManager.MessageReceivedDelegate(qosManager_MessageReceived);
            _clientID = clientID;
        }

        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:
                    OnConnected(new EventArgs());
                    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:
                    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, _keepAlive, false
          ));
        }<commit_msg>Fix to issue identified by <NAME> - QoS1 missing published event<commit_after>using System;
using System.Collections.Generic;
using System.Text;
using MqttLib.Core.Messages;
using MqttLib.Core;
using MqttLib;
//using System.Timers;
using System.Threading;
using System.Diagnostics;
using MqttLib.Logger;
using MqttLib.MatchTree;

namespace MqttLib
{
    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 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, IPersistence store)
        {
            _store = store;
            qosManager = new QoSManager(_store);
            manager = new StreamManager(connString, qosManager);
            qosManager.MessageReceived += new QoSManager.MessageReceivedDelegate(qosManager_MessageReceived);
            _clientID = clientID;
        }

        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:
                    OnConnected(new EventArgs());
                    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, _keepAlive, false
          ));
        }
{'fraction_non_alphanumeric': 0.06495522388059702, 'fraction_numerical': 0.001791044776119403, 'mean_word_length': 1.5138055222088835, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '568722', 'n_tokens_mistral': 2230, 'n_tokens_neox': 2092, 'n_words': 498}