diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2d9d044 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +*.userprefs diff --git a/MqttLib/Core/Messages/MqttExtendedackMessage.cs b/MqttLib/Core/Messages/MqttExtendedackMessage.cs index 2976c97..6c1af9f 100644 --- a/MqttLib/Core/Messages/MqttExtendedackMessage.cs +++ b/MqttLib/Core/Messages/MqttExtendedackMessage.cs @@ -32,7 +32,7 @@ public MqttExtendedackMessage(ulong msgID, byte commondId, string payloadStr) base.variableHeaderLength = 11 + _payload.Length; } - public MqttExtendedackMessage(ulong msgID, string topic, byte[] payload, QoS qos, int ttl, string apn_json) + public MqttExtendedackMessage(ulong msgID, string topic, byte[] payload, QoS qos, int ttl, string apn_json, string third_party_push) : base(MessageType.EXTENDEDACK, 8, msgID) { _commondId = 7; @@ -63,6 +63,12 @@ public MqttExtendedackMessage(ulong msgID, string topic, byte[] payload, QoS qos WriteToStream(pay, apn_json); } + if (third_party_push != null) + { + pay.WriteByte(8); + WriteToStream(pay, third_party_push); + } + byte[] paybytes = pay.ToArray(); if (paybytes.Length <= 65535) { diff --git a/MqttLib/IMqttExtend.cs b/MqttLib/IMqttExtend.cs index 74489ca..9162713 100644 --- a/MqttLib/IMqttExtend.cs +++ b/MqttLib/IMqttExtend.cs @@ -15,7 +15,7 @@ public interface IMqttExtend /// Time for message will be stored on server, in seconds. If set 0, the message will be forever on the server. /// APNS /// Message ID - ulong Publish2(string topic, MqttPayload payload, QoS qos, int ttl, string apn_json); + ulong Publish2(string topic, MqttPayload payload, QoS qos, int ttl, string apn_json, string third_party_push); /// /// Publish a message to a specific client with alias and some options @@ -26,7 +26,8 @@ public interface IMqttExtend /// Time for message will be stored on server, in seconds. If set 0, the message will be forever on the server. /// APNS /// Message ID - ulong Publish2Alias(string alias, MqttPayload payload, QoS qos, int ttl, string apn_json); + ulong Publish2Alias(string alias, MqttPayload payload, QoS qos, int ttl, string apn_json, string third_party_push); + ulong Publish2ToAliasWithToken(string alias, string token, MqttPayload payload, QoS qos, int ttl, string apn_json, string third_party_push); /// /// Request the client's alias diff --git a/MqttLib/IMqttPublisher.cs b/MqttLib/IMqttPublisher.cs index e126085..95cb785 100644 --- a/MqttLib/IMqttPublisher.cs +++ b/MqttLib/IMqttPublisher.cs @@ -39,11 +39,43 @@ public interface IMqttPublisher /// /// alias name /// Message ID - ulong SetAlias(string alias); + ulong SetAlias(string alias); - /// - /// Fired when receipt of publication is confirmed - /// - event CompleteDelegate Published; + /// + /// Publish a message to the MQTT message broker under Yunba-AccessManager control + /// + /// Destination of message + /// grant from Yunba yam server + /// Message body + /// QoS level + /// Whether the message is retained by the broker + /// Message ID + ulong PublishWithToken(string topic, string token, MqttPayload payload, QoS qos, bool retained); + + /// + /// Publish a message to the MQTT message broker under Yunba-AccessManager control + /// + /// Destination of message + /// grant from Yunba yam server + /// Message body + /// QoS level + /// Whether the message is retained by the broker + /// Message ID + ulong SetAliasWithToken(string alias, string token); + ulong Publish2WithToken(string topic, string token, MqttPayload payload, QoS qos, int ttl, string apn_json); + /// + /// Publish a message to the MQTT message broker under Yunba-AccessManager control + /// + /// Destination of message + /// grant from Yunba yam server + /// Message body + /// QoS level + /// Whether the message is retained by the broker + /// Message ID + ulong PublishToAliasWithToken(string alias, string token, MqttPayload payload, QoS qos, bool retained); + /// + /// Fired when receipt of publication is confirmed + /// + event CompleteDelegate Published; } } diff --git a/MqttLib/IMqttSubscriber.cs b/MqttLib/IMqttSubscriber.cs index b6dcb1b..b6a33d1 100644 --- a/MqttLib/IMqttSubscriber.cs +++ b/MqttLib/IMqttSubscriber.cs @@ -35,6 +35,10 @@ public interface IMqttSubscriber /// Topic Name /// Message ID ulong Unsubscribe(string[] topics); + ulong SubscribeWithToken(Subscription[] subscriptions, string token); + ulong SubscribeWithToken(Subscription subscription, string token); + ulong SubscribeWithToken(string topic, string token, QoS qos); + ulong UnsubscribeWithToken(string[] topics, string token); /// /// Fired when the Topic the MQTT client is subscribed to receives a message diff --git a/MqttLib/Mqtt.cs b/MqttLib/Mqtt.cs index d659af9..225ac69 100644 --- a/MqttLib/Mqtt.cs +++ b/MqttLib/Mqtt.cs @@ -16,452 +16,452 @@ namespace MqttLib { - internal class Mqtt : IMqtt, IMqttShared - { - - #region Member Variables - - private StreamManager manager = null; - private QoSManager qosManager = null; - private IPersistence _store = null; - - private TopicTree topicTree = null; - - private string _clientID; - private string _username; - private string _password; - private ushort _keepAlive = 30; - private Timer keepAliveTimer = null; - private DateTime EPOCH = new System.DateTime(1970, 1, 1, 0, 0, 0, 0); - private bool isActiveStop = true; - private bool autoReconnect = false; - private int autoReconnectTimeout = 500; // ms - private MqttConnectMessage connectMsg; - private Hashtable extendAckCallBacks = new Hashtable(); - private Random messageidRandom = new Random(); - - #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 bool AutoReconnect - { - get { return autoReconnect; } - set { autoReconnect = value; } - } - - public int AutoReconnectTimeout - { - get { return autoReconnectTimeout; } - set { autoReconnectTimeout = 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; - - this.ConnectionLost += Mqtt_ConnectionLost; - } - - void Mqtt_ConnectionLost(object sender, EventArgs e) - { - extendAckCallBacks.Clear(); - - if (!isActiveStop) - DoReconnect(); - } - - 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.EXTENDEDACK: - MqttExtendedackMessage mm = (MqttExtendedackMessage)e.Message; - if (mm.CommondId == 8) // publish2 ack or publish2alias ack - OnPublish2ed(mm); - else - OnExtendedAckArrived(new ExtendedAckArrivedArgs(mm.AckID, mm.CommondId, mm.Status, mm.Payload)); - break; - case MessageType.PINGRESP: - break; - case MessageType.UNSUBSCRIBE: - case MessageType.CONNECT: - case MessageType.SUBSCRIBE: - default: - throw new Exception("Unsupported Message Type"); - - } - } - - public void Start() - { - DoConnect(new MqttConnectMessage( - _clientID, _username, _password, _keepAlive, false - )); - } - - public void Start(string willTopic, QoS willQoS, MqttPayload willMsg, bool willRetain) - { - DoConnect(new MqttConnectMessage( - _clientID, _username, _password, _keepAlive, willTopic, willMsg.TrimmedBuffer, willQoS, willRetain, false - )); - } - - public void Start(bool cleanStart) - { - DoConnect(new MqttConnectMessage( - _clientID, _username, _password, _keepAlive, cleanStart - )); - } - - public void Start(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) - { - connectMsg = conmsg; - isActiveStop = false; - - 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 Stop() - { - isActiveStop = true; - extendAckCallBacks.Clear(); - - manager.SendMessage(new MqttDisconnectMessage()); - if (keepAliveTimer != null) - { - keepAliveTimer.Dispose(); - keepAliveTimer = null; - } - manager.Disconnect(); - - } - - public ulong 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); - } - ulong 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 ulong Publish(MqttParcel parcel) - { - return Publish(parcel.Topic, parcel.Payload, parcel.Qos, parcel.Retained); - } - - public ulong Publish2(string topic, MqttPayload payload, QoS qos, int ttl, string apn_json) - { - if (manager.IsConnected) - { - ulong messID = MessageID; - - manager.SendMessage(new MqttExtendedackMessage(messID, topic, payload.TrimmedBuffer, qos, ttl, apn_json)); - return messID; - } - else - { - throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish"); - } - } - - public ulong PublishToAlias(string alias, MqttPayload payload, QoS qos, bool retained) - { - return Publish(",yta/" + alias, payload, qos, retained); - } - - public ulong Publish2Alias(string alias, MqttPayload payload, QoS qos, int ttl, string apn_json) - { - return Publish2(",yta/" + alias, payload, qos, ttl, apn_json); - } - - public ulong Subscribe(Subscription[] subscriptions) - { - if (manager.IsConnected) - { - ulong 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 ulong Subscribe(Subscription subscription) - { - return Subscribe(new Subscription[] { subscription }); - } - - public ulong Subscribe(string topic, QoS qos) - { - return Subscribe(new Subscription(topic, qos)); - } - - public ulong Unsubscribe(string[] topics) - { - if (manager.IsConnected) - { - ulong 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 ulong SetAlias(string alias) - { - return Publish(",yali", alias, QoS.AtLeastOnce, false); - } - - public void GetAlias(ExtendedAckArrivedDelegate cb) - { - SendExtendMessage(1, "", cb); - } - - public void GetTopicList(ExtendedAckArrivedDelegate cb) - { - SendExtendMessage(3, "", cb); - } - - public void GetTopicList(string alias, ExtendedAckArrivedDelegate cb) - { - SendExtendMessage(3, alias, cb); - } - - public void GetAliasList(string topic, ExtendedAckArrivedDelegate cb) - { - SendExtendMessage(5, topic, cb); - } - - public void GetState(string alias, ExtendedAckArrivedDelegate cb) - { - SendExtendMessage(9, alias, cb); - } - - 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) - { - topicTree = new TopicTree(); - } - - topicTree.Add(subscription.Topic, subscriber); - - // TODO: Check if we're already subscribed. - Subscribe(subscription); - } - - public void Unsubscribe(string topic, PublishArrivedDelegate subscriber) - { - topicTree.Remove(topic, subscriber); - - // TODO: Check if this is the last subscriber - Unsubscribe(new string[] { topic }); - } - - #endregion - - #region Event Raising functions - - protected void OnPublishArrived(MqttPublishMessage m) - { - bool accepted = false; - PublishArrivedArgs e = new PublishArrivedArgs(m.Topic, m.Payload, m.Retained, m.QualityOfService); - - if (PublishArrived != null) - { - try - { - accepted |= PublishArrived(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - - if (topicTree != null) - { - List subscribers = topicTree.CollectMatches(new Topic(m.Topic)); - foreach (PublishArrivedDelegate pad in subscribers) - { - try - { - accepted |= pad(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - if (m.QualityOfService > QoS.BestEfforts) - { - qosManager.PublishAccepted(m.MessageID, accepted); - } - - } - - protected void OnPublished(CompleteArgs e) - { - if (Published != null) - { - try - { - Published(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - protected void OnPublish2ed(MqttExtendedackMessage m) - { - OnPublished(new CompleteArgs(m.MessageID)); - } - - protected void OnExtendedAckArrived(ExtendedAckArrivedArgs e) - { - /* + internal class Mqtt : IMqtt, IMqttShared + { + + #region Member Variables + + private StreamManager manager = null; + private QoSManager qosManager = null; + private IPersistence _store = null; + + private TopicTree topicTree = null; + + private string _clientID; + private string _username; + private string _password; + private ushort _keepAlive = 30; + private Timer keepAliveTimer = null; + private DateTime EPOCH = new System.DateTime(1970, 1, 1, 0, 0, 0, 0); + private bool isActiveStop = true; + private bool autoReconnect = false; + private int autoReconnectTimeout = 500; // ms + private MqttConnectMessage connectMsg; + private Hashtable extendAckCallBacks = new Hashtable(); + private Random messageidRandom = new Random(); + + #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 bool AutoReconnect + { + get { return autoReconnect; } + set { autoReconnect = value; } + } + + public int AutoReconnectTimeout + { + get { return autoReconnectTimeout; } + set { autoReconnectTimeout = 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; + + this.ConnectionLost += Mqtt_ConnectionLost; + } + + void Mqtt_ConnectionLost(object sender, EventArgs e) + { + extendAckCallBacks.Clear(); + + if (!isActiveStop) + DoReconnect(); + } + + 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.EXTENDEDACK: + MqttExtendedackMessage mm = (MqttExtendedackMessage)e.Message; + if (mm.CommondId == 8) // publish2 ack or publish2alias ack + OnPublish2ed(mm); + else + OnExtendedAckArrived(new ExtendedAckArrivedArgs(mm.AckID, mm.CommondId, mm.Status, mm.Payload)); + break; + case MessageType.PINGRESP: + break; + case MessageType.UNSUBSCRIBE: + case MessageType.CONNECT: + case MessageType.SUBSCRIBE: + default: + throw new Exception("Unsupported Message Type"); + + } + } + + public void Start() + { + DoConnect(new MqttConnectMessage( + _clientID, _username, _password, _keepAlive, false + )); + } + + public void Start(string willTopic, QoS willQoS, MqttPayload willMsg, bool willRetain) + { + DoConnect(new MqttConnectMessage( + _clientID, _username, _password, _keepAlive, willTopic, willMsg.TrimmedBuffer, willQoS, willRetain, false + )); + } + + public void Start(bool cleanStart) + { + DoConnect(new MqttConnectMessage( + _clientID, _username, _password, _keepAlive, cleanStart + )); + } + + public void Start(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) + { + connectMsg = conmsg; + isActiveStop = false; + + 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 Stop() + { + isActiveStop = true; + extendAckCallBacks.Clear(); + + manager.SendMessage(new MqttDisconnectMessage()); + if (keepAliveTimer != null) + { + keepAliveTimer.Dispose(); + keepAliveTimer = null; + } + manager.Disconnect(); + + } + + public ulong 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); + } + ulong 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 ulong Publish(MqttParcel parcel) + { + return Publish(parcel.Topic, parcel.Payload, parcel.Qos, parcel.Retained); + } + + public ulong Publish2(string topic, MqttPayload payload, QoS qos, int ttl, string apn_json, string third_party_push) + { + if (manager.IsConnected) + { + ulong messID = MessageID; + + manager.SendMessage(new MqttExtendedackMessage(messID, topic, payload.TrimmedBuffer, qos, ttl, apn_json, third_party_push)); + return messID; + } + else + { + throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish"); + } + } + + public ulong PublishToAlias(string alias, MqttPayload payload, QoS qos, bool retained) + { + return Publish(",yta/" + alias, payload, qos, retained); + } + + public ulong Publish2Alias(string alias, MqttPayload payload, QoS qos, int ttl, string apn_json, string third_party_push) + { + return Publish2(",yta/" + alias, payload, qos, ttl, apn_json, third_party_push); + } + + public ulong Subscribe(Subscription[] subscriptions) + { + if (manager.IsConnected) + { + ulong 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 ulong Subscribe(Subscription subscription) + { + return Subscribe(new Subscription[] { subscription }); + } + + public ulong Subscribe(string topic, QoS qos) + { + return Subscribe(new Subscription(topic, qos)); + } + + public ulong Unsubscribe(string[] topics) + { + if (manager.IsConnected) + { + ulong 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 ulong SetAlias(string alias) + { + return Publish(",yali", alias, QoS.AtLeastOnce, false); + } + + public void GetAlias(ExtendedAckArrivedDelegate cb) + { + SendExtendMessage(1, "", cb); + } + + public void GetTopicList(ExtendedAckArrivedDelegate cb) + { + SendExtendMessage(3, "", cb); + } + + public void GetTopicList(string alias, ExtendedAckArrivedDelegate cb) + { + SendExtendMessage(3, alias, cb); + } + + public void GetAliasList(string topic, ExtendedAckArrivedDelegate cb) + { + SendExtendMessage(5, topic, cb); + } + + public void GetState(string alias, ExtendedAckArrivedDelegate cb) + { + SendExtendMessage(9, alias, cb); + } + + 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) + { + topicTree = new TopicTree(); + } + + topicTree.Add(subscription.Topic, subscriber); + + // TODO: Check if we're already subscribed. + Subscribe(subscription); + } + + public void Unsubscribe(string topic, PublishArrivedDelegate subscriber) + { + topicTree.Remove(topic, subscriber); + + // TODO: Check if this is the last subscriber + Unsubscribe(new string[] { topic }); + } + + #endregion + + #region Event Raising functions + + protected void OnPublishArrived(MqttPublishMessage m) + { + bool accepted = false; + PublishArrivedArgs e = new PublishArrivedArgs(m.Topic, m.Payload, m.Retained, m.QualityOfService); + + if (PublishArrived != null) + { + try + { + accepted |= PublishArrived(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + + if (topicTree != null) + { + List subscribers = topicTree.CollectMatches(new Topic(m.Topic)); + foreach (PublishArrivedDelegate pad in subscribers) + { + try + { + accepted |= pad(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + if (m.QualityOfService > QoS.BestEfforts) + { + qosManager.PublishAccepted(m.MessageID, accepted); + } + + } + + protected void OnPublished(CompleteArgs e) + { + if (Published != null) + { + try + { + Published(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + protected void OnPublish2ed(MqttExtendedackMessage m) + { + OnPublished(new CompleteArgs(m.MessageID)); + } + + protected void OnExtendedAckArrived(ExtendedAckArrivedArgs e) + { + /* Console.WriteLine("Extended ack arrived"); Console.WriteLine("Commond id: " + e.CommondID); Console.WriteLine("Commond status: " + e.Status); @@ -469,127 +469,197 @@ protected void OnExtendedAckArrived(ExtendedAckArrivedArgs e) Console.WriteLine(); */ - if(extendAckCallBacks.Contains(e.MessageID)) - { - ((ExtendedAckArrivedDelegate)extendAckCallBacks[e.MessageID])(this, e); - extendAckCallBacks.Remove(e.MessageID); - } - } - - protected void OnSubscribed(CompleteArgs e) - { - if (Subscribed != null) - { - try - { - Subscribed(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - protected void OnUnsubscribed(CompleteArgs e) - { - if (Unsubscribed != null) - { - try - { - Unsubscribed(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - protected void OnConnectionLost(EventArgs e) - { - if (keepAliveTimer != null) - { - keepAliveTimer.Dispose(); - keepAliveTimer = null; - } - - if (ConnectionLost != null) - { - try - { - ConnectionLost(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - protected void OnConnected(EventArgs e) - { - if (Connected != null) - { - try - { - Connected(this, e); - } - catch (Exception ex) - { - MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); - } - } - } - - - - #endregion - - private ulong MessageID - { - get - { - ulong ms = (ulong)Math.Round((DateTime.Now - EPOCH).TotalMilliseconds); - ms = ms << 23; - - ulong id = ms | (uint)(messageidRandom.Next() & 0x7FFFFF); - return id; - } - } - - private void DoReconnect() - { - while (autoReconnect) - { - // Log.Write(LogLevel.INFO, "Client reconnect."); - - try - { - DoConnect(connectMsg); - break; - } - catch (Exception) - { - Thread.Sleep(autoReconnectTimeout); - } - } - } - - private void SendExtendMessage(byte cmdID, string msg, ExtendedAckArrivedDelegate cb) - { - if (manager.IsConnected) - { - ulong messID = MessageID; - - manager.SendMessage(new MqttExtendedackMessage(messID, cmdID, msg)); - - extendAckCallBacks.Add(messID, cb); - } - else - { - throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish"); - } - } - } -} + if (extendAckCallBacks.Contains(e.MessageID)) + { + ((ExtendedAckArrivedDelegate)extendAckCallBacks[e.MessageID])(this, e); + extendAckCallBacks.Remove(e.MessageID); + } + } + + protected void OnSubscribed(CompleteArgs e) + { + if (Subscribed != null) + { + try + { + Subscribed(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + protected void OnUnsubscribed(CompleteArgs e) + { + if (Unsubscribed != null) + { + try + { + Unsubscribed(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + protected void OnConnectionLost(EventArgs e) + { + if (keepAliveTimer != null) + { + keepAliveTimer.Dispose(); + keepAliveTimer = null; + } + + if (ConnectionLost != null) + { + try + { + ConnectionLost(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + protected void OnConnected(EventArgs e) + { + if (Connected != null) + { + try + { + Connected(this, e); + } + catch (Exception ex) + { + MqttLib.Logger.Log.Write(LogLevel.ERROR, "MqttLib: Uncaught exception from user delegate: " + ex.ToString()); + } + } + } + + + + #endregion + + private ulong MessageID + { + get + { + ulong ms = (ulong)Math.Round((DateTime.Now - EPOCH).TotalMilliseconds); + ms = ms << 23; + + ulong id = ms | (uint)(messageidRandom.Next() & 0x7FFFFF); + return id; + } + } + + private void DoReconnect() + { + while (autoReconnect) + { + // Log.Write(LogLevel.INFO, "Client reconnect."); + + try + { + DoConnect(connectMsg); + break; + } + catch (Exception) + { + Thread.Sleep(autoReconnectTimeout); + } + } + } + + private void SendExtendMessage(byte cmdID, string msg, ExtendedAckArrivedDelegate cb) + { + if (manager.IsConnected) + { + ulong messID = MessageID; + + manager.SendMessage(new MqttExtendedackMessage(messID, cmdID, msg)); + + extendAckCallBacks.Add(messID, cb); + } + else + { + throw new MqttNotConnectedException("You need to connect to a broker before trying to Publish"); + } + } + + #region pub/pubToAlias/pub2/pub2toAlias; /sub/unsub/subP/unsubP; setAlias with token + + public ulong PublishWithToken(string topic, string token, MqttPayload payload, QoS qos, bool retained) + { + string topicWithToken = ",yam" + token + "_" + topic; + return Publish(topicWithToken, payload, qos, retained); + } + + public ulong PublishToAliasWithToken(string alias, string token, MqttPayload payload, QoS qos, bool retained) + { + string AliasWithToken = ",yam" + token + "_" + ",yta/" + alias; + return Publish(AliasWithToken, payload, qos, retained); + } + + public ulong Publish2WithToken(string topic, string token, MqttPayload payload, QoS qos, int ttl, string apn_json) + { + string topicWithToken = ",yam" + token + "_" + topic; + return Publish2(topicWithToken, payload, qos, ttl, apn_json); + } + + public ulong Publish2ToAliasWithToken(string alias, string token, MqttPayload payload, QoS qos, int ttl, string apn_json) + { + string AliasWithToken = ",yam" + token + "_" + ",yta/" + alias; + return Publish2(AliasWithToken, payload, qos, ttl, apn_json); + } + + public ulong SubscribeWithToken(Subscription[] subscriptions, string token) + { + List subscriptionsWithToken = new List(); + foreach (var subscription in subscriptions) + { + string topicWithToken = ",yam" + token + "_" + subscription.Topic; + Subscription subscriptionWithToken = new Subscription(topicWithToken, subscription.QualityOfService); + subscriptionsWithToken.Add(subscriptionWithToken); + } + Subscription[] SubscriptionsWithTokens = subscriptionsWithToken.ToArray(); + return Subscribe(SubscriptionsWithTokens); + } + + public ulong SubscribeWithToken(Subscription subscription, string token) + { + string topicWithToken = ",yam" + token + "_" + subscription.Topic; + Subscription SubWithToken = new Subscription(topicWithToken, subscription.QualityOfService); + return Subscribe(new Subscription[] { SubWithToken }); + } + + public ulong SubscribeWithToken(string topic, string token, QoS qos) + { + return SubscribeWithToken(new Subscription(topic, qos), token); + } + + public ulong UnsubscribeWithToken(string[] topics, string token) + { + List topicsWithToken = new List(); + foreach (var topic in topics) + { + string topicWithToken = ",yam" + token + "_" + topic; + topicsWithToken.Add(topicWithToken); + } + string[] topicsWithTokens = topicsWithToken.ToArray(); + return Unsubscribe(topicsWithTokens); + } + + public ulong SetAliasWithToken(string alias, string token) + { + string AliasWithToken = ",yam" + token + "_" + ",yali"; + return Publish(AliasWithToken, alias, QoS.AtLeastOnce, false);; + } + #endregion + } +} \ No newline at end of file diff --git a/Sample/Program.cs b/Sample/Program.cs old mode 100644 new mode 100755 index 8144bd1..04c2f9d --- a/Sample/Program.cs +++ b/Sample/Program.cs @@ -170,7 +170,7 @@ static void Main(string[] args) Console.Write("The message: "); string m = Console.ReadLine(); - _client.Publish2(a, m, QoS.AtLeastOnce, 30, ""); + _client.Publish2(a, m, QoS.AtLeastOnce, 30, "", "{\"notification_title\":\"ÄãºÃ\", \"notification_content\":\"ÔÆ°ÍÍÆËÍ\"}"); } break; case 'c': @@ -181,7 +181,7 @@ static void Main(string[] args) Console.Write("The message: "); string m = Console.ReadLine(); - _client.Publish2Alias(a, m, QoS.AtLeastOnce, 30, ""); + _client.Publish2Alias(a, m, QoS.AtLeastOnce, 30, "", "{\"notification_title\":\"ÄãºÃ\", \"notification_content\":\"ÔÆ°ÍÍÆËÍ\"}"); } break; default: diff --git a/Sample/Properties/AssemblyInfo.cs b/Sample/Properties/AssemblyInfo.cs old mode 100644 new mode 100755 diff --git a/Sample/Sample.csproj b/Sample/Sample.csproj old mode 100644 new mode 100755 diff --git a/Sample/app.config b/Sample/app.config old mode 100644 new mode 100755