Initial commit
diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..11ecb79 --- /dev/null +++ b/LICENSE
@@ -0,0 +1,198 @@ +Eclipse Public License - v 1.0 + +THE ACCOMPANYING PROGRAM IS PROVIDED UNDER THE TERMS OF THIS ECLIPSE PUBLIC +LICENSE ("AGREEMENT"). ANY USE, REPRODUCTION OR DISTRIBUTION OF THE PROGRAM +CONSTITUTES RECIPIENT'S ACCEPTANCE OF THIS AGREEMENT. + +1. DEFINITIONS + +"Contribution" means: + +a) in the case of the initial Contributor, the initial code and documentation + distributed under this Agreement, and +b) in the case of each subsequent Contributor: + i) changes to the Program, and + ii) additions to the Program; + + where such changes and/or additions to the Program originate from and are + distributed by that particular Contributor. A Contribution 'originates' from + a Contributor if it was added to the Program by such Contributor itself or + anyone acting on such Contributor's behalf. Contributions do not include + additions to the Program which: (i) are separate modules of software + distributed in conjunction with the Program under their own license + agreement, and (ii) are not derivative works of the Program. + +"Contributor" means any person or entity that distributes the Program. + +"Licensed Patents" mean patent claims licensable by a Contributor which are +necessarily infringed by the use or sale of its Contribution alone or when +combined with the Program. + +"Program" means the Contributions distributed in accordance with this Agreement. + +"Recipient" means anyone who receives the Program under this Agreement, +including all Contributors. + +2. GRANT OF RIGHTS + a) Subject to the terms of this Agreement, each Contributor hereby grants + Recipient a non-exclusive, worldwide, royalty-free copyright license to + reproduce, prepare derivative works of, publicly display, publicly perform, + distribute and sublicense the Contribution of such Contributor, if any, and + such derivative works, in source code and object code form. + b) Subject to the terms of this Agreement, each Contributor hereby grants + Recipient a non-exclusive, worldwide, royalty-free patent license under + Licensed Patents to make, use, sell, offer to sell, import and otherwise + transfer the Contribution of such Contributor, if any, in source code and + object code form. This patent license shall apply to the combination of the + Contribution and the Program if, at the time the Contribution is added by + the Contributor, such addition of the Contribution causes such combination + to be covered by the Licensed Patents. The patent license shall not apply + to any other combinations which include the Contribution. No hardware per + se is licensed hereunder. + c) Recipient understands that although each Contributor grants the licenses to + its Contributions set forth herein, no assurances are provided by any + Contributor that the Program does not infringe the patent or other + intellectual property rights of any other entity. Each Contributor + disclaims any liability to Recipient for claims brought by any other entity + based on infringement of intellectual property rights or otherwise. As a + condition to exercising the rights and licenses granted hereunder, each + Recipient hereby assumes sole responsibility to secure any other + intellectual property rights needed, if any. For example, if a third party + patent license is required to allow Recipient to distribute the Program, it + is Recipient's responsibility to acquire that license before distributing + the Program. + d) Each Contributor represents that to its knowledge it has sufficient + copyright rights in its Contribution, if any, to grant the copyright + license set forth in this Agreement. + +3. REQUIREMENTS + +A Contributor may choose to distribute the Program in object code form under its +own license agreement, provided that: + + a) it complies with the terms and conditions of this Agreement; and + b) its license agreement: + i) effectively disclaims on behalf of all Contributors all warranties and + conditions, express and implied, including warranties or conditions of + title and non-infringement, and implied warranties or conditions of + merchantability and fitness for a particular purpose; + ii) effectively excludes on behalf of all Contributors all liability for + damages, including direct, indirect, special, incidental and + consequential damages, such as lost profits; + iii) states that any provisions which differ from this Agreement are offered + by that Contributor alone and not by any other party; and + iv) states that source code for the Program is available from such + Contributor, and informs licensees how to obtain it in a reasonable + manner on or through a medium customarily used for software exchange. + +When the Program is made available in source code form: + + a) it must be made available under this Agreement; and + b) a copy of this Agreement must be included with each copy of the Program. + Contributors may not remove or alter any copyright notices contained within + the Program. + +Each Contributor must identify itself as the originator of its Contribution, if +any, in a manner that reasonably allows subsequent Recipients to identify the +originator of the Contribution. + +4. COMMERCIAL DISTRIBUTION + +Commercial distributors of software may accept certain responsibilities with +respect to end users, business partners and the like. While this license is +intended to facilitate the commercial use of the Program, the Contributor who +includes the Program in a commercial product offering should do so in a manner +which does not create potential liability for other Contributors. Therefore, if +a Contributor includes the Program in a commercial product offering, such +Contributor ("Commercial Contributor") hereby agrees to defend and indemnify +every other Contributor ("Indemnified Contributor") against any losses, damages +and costs (collectively "Losses") arising from claims, lawsuits and other legal +actions brought by a third party against the Indemnified Contributor to the +extent caused by the acts or omissions of such Commercial Contributor in +connection with its distribution of the Program in a commercial product +offering. The obligations in this section do not apply to any claims or Losses +relating to any actual or alleged intellectual property infringement. In order +to qualify, an Indemnified Contributor must: a) promptly notify the Commercial +Contributor in writing of such claim, and b) allow the Commercial Contributor to +control, and cooperate with the Commercial Contributor in, the defense and any +related settlement negotiations. The Indemnified Contributor may participate in +any such claim at its own expense. + +For example, a Contributor might include the Program in a commercial product +offering, Product X. That Contributor is then a Commercial Contributor. If that +Commercial Contributor then makes performance claims, or offers warranties +related to Product X, those performance claims and warranties are such +Commercial Contributor's responsibility alone. Under this section, the +Commercial Contributor would have to defend claims against the other +Contributors related to those performance claims and warranties, and if a court +requires any other Contributor to pay any damages as a result, the Commercial +Contributor must pay those damages. + +5. NO WARRANTY + +EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, THE PROGRAM IS PROVIDED ON AN +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, EITHER EXPRESS OR +IMPLIED INCLUDING, WITHOUT LIMITATION, ANY WARRANTIES OR CONDITIONS OF TITLE, +NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE. Each +Recipient is solely responsible for determining the appropriateness of using and +distributing the Program and assumes all risks associated with its exercise of +rights under this Agreement , including but not limited to the risks and costs +of program errors, compliance with applicable laws, damage to or loss of data, +programs or equipment, and unavailability or interruption of operations. + +6. DISCLAIMER OF LIABILITY + +EXCEPT AS EXPRESSLY SET FORTH IN THIS AGREEMENT, NEITHER RECIPIENT NOR ANY +CONTRIBUTORS SHALL HAVE ANY LIABILITY FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING WITHOUT LIMITATION LOST +PROFITS), HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, +STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +OUT OF THE USE OR DISTRIBUTION OF THE PROGRAM OR THE EXERCISE OF ANY RIGHTS +GRANTED HEREUNDER, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGES. + +7. GENERAL + +If any provision of this Agreement is invalid or unenforceable under applicable +law, it shall not affect the validity or enforceability of the remainder of the +terms of this Agreement, and without further action by the parties hereto, such +provision shall be reformed to the minimum extent necessary to make such +provision valid and enforceable. + +If Recipient institutes patent litigation against any entity (including a +cross-claim or counterclaim in a lawsuit) alleging that the Program itself +(excluding combinations of the Program with other software or hardware) +infringes such Recipient's patent(s), then such Recipient's rights granted under +Section 2(b) shall terminate as of the date such litigation is filed. + +All Recipient's rights under this Agreement shall terminate if it fails to +comply with any of the material terms or conditions of this Agreement and does +not cure such failure in a reasonable period of time after becoming aware of +such noncompliance. If all Recipient's rights under this Agreement terminate, +Recipient agrees to cease use and distribution of the Program as soon as +reasonably practicable. However, Recipient's obligations under this Agreement +and any licenses granted by Recipient relating to the Program shall continue and +survive. + +Everyone is permitted to copy and distribute copies of this Agreement, but in +order to avoid inconsistency the Agreement is copyrighted and may only be +modified in the following manner. The Agreement Steward reserves the right to +publish new versions (including revisions) of this Agreement from time to time. +No one other than the Agreement Steward has the right to modify this Agreement. +The Eclipse Foundation is the initial Agreement Steward. The Eclipse Foundation +may assign the responsibility to serve as the Agreement Steward to a suitable +separate entity. Each new version of the Agreement will be given a +distinguishing version number. The Program (including Contributions) may always +be distributed subject to the version of the Agreement under which it was +received. In addition, after a new version of the Agreement is published, +Contributor may elect to distribute the Program (including its Contributions) +under the new version. Except as expressly stated in Sections 2(a) and 2(b) +above, Recipient receives no rights or licenses to the intellectual property of +any Contributor under this Agreement, whether expressly, by implication, +estoppel or otherwise. All rights in the Program not expressly granted under +this Agreement are reserved. + +This Agreement is governed by the laws of the State of New York and the +intellectual property laws of the United States of America. No party to this +Agreement will bring a legal action under this Agreement more than one year +after the cause of action arose. Each party waives its rights to a jury trial in +any resulting litigation.
diff --git a/README.md b/README.md new file mode 100644 index 0000000..f5cec4a --- /dev/null +++ b/README.md
@@ -0,0 +1,10 @@ +Smidge +====== + +An [MQTT-SN](http://mqtt.org/new/wp-content/uploads/2009/06/MQTT-SN_spec_v1.2.pdf) implementation in Go + +Including +--------- +- MQTT-SN Client Library (in progress) +- MQTT-SN Aggregating and Transparent Gateway +
diff --git a/client/client.go b/client/client.go new file mode 100644 index 0000000..8148a7b --- /dev/null +++ b/client/client.go
@@ -0,0 +1,243 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package smidge + +import ( + "errors" + "net" + "net/url" + . "smidge/packets" + "sync" +) + +type MessageHandler func(client *SNClient, message *PublishMessage) + +type Will struct { + Topic string + Data []byte + Qos byte + Retain bool +} + +type Client interface { + Connect() +} + +const ( + UNCONNECTED byte = iota + CONNECTING + CONNECTED +) + +type SNClient struct { + sync.RWMutex + ClientId string + OutstandingMessages map[uint16]Message + RegisteredTopics map[string]uint16 + MessageHandlers map[uint16]MessageHandler + PredefinedTopics map[string]uint16 + PredefinedMessageHandlers map[uint16]MessageHandler + DefaultMessageHandler MessageHandler + MessageIds mids + will Will + suTokens map[int]Token + conn net.Conn + incoming chan Message + outgoing chan *MessageAndToken + stop chan struct{} + state byte +} + +func NewClient(server string, clientid string) (*SNClient, error) { + c := &SNClient{} + c.ClientId = clientid + serverURI, err := url.Parse(server) + if err != nil { + ERROR.Println(err.Error()) + return nil, err + } + c.conn, err = net.Dial("udp", serverURI.Host) + if err != nil { + ERROR.Println(err.Error()) + return nil, err + } + c.RegisteredTopics = make(map[string]uint16) + c.MessageHandlers = make(map[uint16]MessageHandler) + c.PredefinedTopics = make(map[string]uint16) + c.PredefinedMessageHandlers = make(map[uint16]MessageHandler) + c.suTokens = make(map[int]Token) + c.incoming = make(chan Message) + c.outgoing = make(chan *MessageAndToken) + c.stop = make(chan struct{}) + c.MessageIds.index = make(map[uint16]Token) + go c.receive() + go c.send() + go c.handle() + + return c, nil +} + +func (c *SNClient) setState(s byte) { + c.Lock() + defer c.Unlock() + c.state = s +} + +func (c *SNClient) Connect() *ConnectToken { + c.setState(CONNECTING) + ct := newToken(CONNECT).(*ConnectToken) + c.suTokens[CONNECT] = ct + cp := NewMessage(CONNECT).(*ConnectMessage) + cp.CleanSession = true + cp.ClientId = []byte(c.ClientId) + cp.Duration = 30 + if c.will.Topic != "" { + cp.Will = true + } + cp.Write(c.conn) + return ct +} + +func (c *SNClient) Predefine(topic string, topicid uint16) error { + for t, id := range c.RegisteredTopics { + if t == topic || id == topicid { + return errors.New("Topic or TopicId already in use") + } + } + c.RegisteredTopics[topic] = topicid + return nil +} + +func (c *SNClient) Register(topic string) *RegisterToken { + t := newToken(REGISTER).(*RegisterToken) + t.TopicName = topic + r := NewMessage(REGISTER).(*RegisterMessage) + r.TopicName = []byte(topic) + c.outgoing <- &MessageAndToken{m: r, t: t} + return t +} + +func (c *SNClient) Subscribe(topic string, qos byte, mh MessageHandler) *SubscribeToken { + t := newToken(SUBSCRIBE).(*SubscribeToken) + t.handler = mh + t.TopicName = topic + s := NewMessage(SUBSCRIBE).(*SubscribeMessage) + if len(topic) > 2 { + s.TopicIdType = 0x00 + t.topicType = 0x00 + } else { + s.TopicIdType = 0x02 + t.topicType = 0x02 + } + s.TopicName = []byte(topic) + s.Qos = qos + c.outgoing <- &MessageAndToken{m: s, t: t} + return t +} + +func (c *SNClient) SubscribePredefined(topicid uint16, qos byte, mh MessageHandler) *SubscribeToken { + t := newToken(SUBSCRIBE).(*SubscribeToken) + t.handler = mh + t.topicType = 0x01 + s := NewMessage(SUBSCRIBE).(*SubscribeMessage) + s.TopicIdType = 0x01 + s.TopicId = topicid + s.Qos = qos + c.outgoing <- &MessageAndToken{m: s, t: t} + return t +} + +func (c *SNClient) Publish(topic string, qos byte, retain bool, data []byte) *PublishToken { + t := newToken(PUBLISH).(*PublishToken) + p := NewMessage(PUBLISH).(*PublishMessage) + p.TopicId = c.RegisteredTopics[topic] + p.Qos = qos + p.Retain = retain + p.Data = data + c.outgoing <- &MessageAndToken{m: p, t: t} + return t +} + +func (c *SNClient) PublishPredefined(topicid uint16, qos byte, retain bool, data []byte) *PublishToken { + t := newToken(PUBLISH).(*PublishToken) + p := NewMessage(PUBLISH).(*PublishMessage) + p.TopicIdType = 0x01 + p.TopicId = topicid + p.Qos = qos + p.Retain = retain + p.Data = data + c.outgoing <- &MessageAndToken{m: p, t: t} + return t +} + +func (c *SNClient) SetWill(topic string, qos byte, retain bool, data []byte) { + c.will.Topic = topic + c.will.Qos = qos + c.will.Retain = retain + c.will.Data = make([]byte, len(data)) + copy(c.will.Data, data) +} + +func (c *SNClient) SetWillTopic(t string) *WillToken { + c.will.Topic = t + wt := newToken(WILLTOPIC).(*WillToken) + if c.state != UNCONNECTED { + wm := NewMessage(WILLTOPICUPD).(*WillTopicUpdateMessage) + wm.Qos = c.will.Qos + wm.Retain = c.will.Retain + c.outgoing <- &MessageAndToken{m: wm, t: wt} + return wt + } + wt.flowComplete() + return wt +} + +func (c *SNClient) SetWillQos(q byte) *WillToken { + c.will.Qos = q + wt := newToken(WILLTOPIC).(*WillToken) + if c.state != UNCONNECTED { + wm := NewMessage(WILLTOPICUPD).(*WillTopicUpdateMessage) + wm.Qos = c.will.Qos + wm.Retain = c.will.Retain + c.outgoing <- &MessageAndToken{m: wm, t: wt} + return wt + } + wt.flowComplete() + return wt +} + +func (c *SNClient) SetWillRetain(r bool) *WillToken { + c.will.Retain = r + wt := newToken(WILLTOPIC).(*WillToken) + if c.state != UNCONNECTED { + wm := NewMessage(WILLTOPICUPD).(*WillTopicUpdateMessage) + wm.Qos = c.will.Qos + wm.Retain = c.will.Retain + c.outgoing <- &MessageAndToken{m: wm, t: wt} + return wt + } + wt.flowComplete() + return wt +} + +func (c *SNClient) SetWillData(d []byte) *WillToken { + c.will.Data = make([]byte, len(d)) + copy(c.will.Data, d) + wt := newToken(WILLMSGUPD).(*WillToken) + if c.state != UNCONNECTED { + wm := NewMessage(WILLMSGUPD).(*WillMsgUpdateMessage) + c.outgoing <- &MessageAndToken{m: wm, t: wt} + return wt + } + wt.flowComplete() + return wt +}
diff --git a/client/messageids.go b/client/messageids.go new file mode 100644 index 0000000..746f740 --- /dev/null +++ b/client/messageids.go
@@ -0,0 +1,48 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package smidge + +import ( + "sync" +) + +type mids struct { + sync.RWMutex + index map[uint16]Token +} + +func (m *mids) freeId(id uint16) { + m.Lock() + defer m.Unlock() + delete(m.index, id) +} + +func (m *mids) getId(t Token) uint16 { + m.Lock() + defer m.Unlock() + for i := uint16(1); i < uint16(65535); i++ { + if _, ok := m.index[i]; !ok { + m.index[i] = t + return i + } + } + return 0 +} + +func (m *mids) getToken(id uint16) Token { + m.RLock() + defer m.RUnlock() + if token, ok := m.index[id]; ok { + return token + } + return nil +}
diff --git a/client/net.go b/client/net.go new file mode 100644 index 0000000..c45cbbf --- /dev/null +++ b/client/net.go
@@ -0,0 +1,154 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package smidge + +import ( + . "smidge/packets" +) + +func (c *SNClient) receive() { + var err error + var m Message + + DEBUG.Println(NET, "started receive()") + for { + if m, err = ReadPacket(c.conn); err != nil { + break + } + DEBUG.Println(NET, "Received", MessageNames[m.MessageType()]) + c.incoming <- m + } + + select { + case <-c.stop: + DEBUG.Println(NET, "stopped receive()") + default: + ERROR.Println(NET, "stopped receive() due to error") + } + return +} + +func (c *SNClient) send() { + DEBUG.Println(NET, "started send()") + for { + mt := <-c.outgoing + DEBUG.Println(NET, "sending message", MessageNames[mt.m.MessageType()]) + switch mt.m.MessageType() { + case REGISTER: + mt.m.(*RegisterMessage).MessageId = c.MessageIds.getId(mt.t) + case PUBLISH: + mt.m.(*PublishMessage).MessageId = c.MessageIds.getId(mt.t) + case SUBSCRIBE: + mt.m.(*SubscribeMessage).MessageId = c.MessageIds.getId(mt.t) + case UNSUBSCRIBE: + mt.m.(*UnsubscribeMessage).MessageId = c.MessageIds.getId(mt.t) + } + mt.m.Write(c.conn) + if mt.m.MessageType() == DISCONNECT { + DEBUG.Println(NET, "Sent DISCONNECT, closing connection") + c.conn.Close() + return + } + } +} + +func (c *SNClient) handle() { + DEBUG.Println(NET, "started handle()") + for { + select { + case m := <-c.incoming: + DEBUG.Println(NET, "got message off <-incoming", MessageNames[m.MessageType()]) + switch m.MessageType() { + case CONNACK: + ca := m.(*ConnackMessage) + c.setState(CONNECTED) + ct := c.suTokens[CONNECT].(*ConnectToken) + ct.ReturnCode = ca.ReturnCode + ct.flowComplete() + case REGISTER: + r := m.(*RegisterMessage) + c.RegisteredTopics[string(r.TopicName)] = r.TopicId + ra := NewMessage(REGACK).(*RegackMessage) + ra.MessageId = r.MessageId + ra.TopicId = r.TopicId + ra.ReturnCode = ACCEPTED + c.outgoing <- &MessageAndToken{m: ra, t: nil} + case REGACK: + ra := m.(*RegackMessage) + t := c.MessageIds.getToken(ra.MessageId).(*RegisterToken) + t.ReturnCode = ra.ReturnCode + switch ra.ReturnCode { + case ACCEPTED: + DEBUG.Println(NET, t.TopicName, "registered as", ra.TopicId) + c.RegisteredTopics[t.TopicName] = ra.TopicId + t.TopicId = ra.TopicId + default: + ERROR.Println(NET, ra.ReturnCode, "for REGISTER for", string(t.TopicName)) + } + t.flowComplete() + case SUBACK: + sa := m.(*SubackMessage) + t := c.MessageIds.getToken(sa.MessageId).(*SubscribeToken) + t.ReturnCode = sa.ReturnCode + switch sa.ReturnCode { + case ACCEPTED: + t.Qos = sa.Qos + if sa.TopicId > 0x00 { + t.TopicId = sa.TopicId + switch t.topicType { + case 0x01: + c.PredefinedMessageHandlers[sa.TopicId] = t.handler + case 0x00, 0x02: + c.RegisteredTopics[t.TopicName] = sa.TopicId + c.MessageHandlers[sa.TopicId] = t.handler + } + } + default: + ERROR.Println(NET, sa.ReturnCode, "for SUBSCRIBE to", t.TopicName) + } + t.flowComplete() + case PUBLISH: + p := m.(*PublishMessage) + switch p.TopicIdType { + case 0x00: + if handler, ok := c.MessageHandlers[p.TopicId]; ok { + go handler(c, p) + } + default: + if c.DefaultMessageHandler != nil { + go c.DefaultMessageHandler(c, p) + } + } + case PUBACK: + pa := m.(*PubackMessage) + t := c.MessageIds.getToken(pa.MessageId).(*PublishToken) + t.ReturnCode = pa.ReturnCode + t.TopicId = pa.TopicId + t.flowComplete() + case DISCONNECT: + c.setState(UNCONNECTED) + DEBUG.Println(NET, "Received DISCONNECT, closing socket") + c.conn.Close() + case WILLTOPICREQ: + wm := NewMessage(WILLTOPIC).(*WillTopicMessage) + wm.WillTopic = []byte(c.will.Topic) + wm.Qos = c.will.Qos + wm.Retain = c.will.Retain + c.outgoing <- &MessageAndToken{m: wm, t: nil} + case WILLMSGREQ: + wm := NewMessage(WILLMSG).(*WillMsgMessage) + wm.WillMsg = c.will.Data + c.outgoing <- &MessageAndToken{m: wm, t: nil} + } + } + } +}
diff --git a/client/samples/simple.go b/client/samples/simple.go new file mode 100644 index 0000000..542d71c --- /dev/null +++ b/client/samples/simple.go
@@ -0,0 +1,45 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package main + +import ( + "fmt" + "log" + "os" + "smidge/client" + "smidge/packets" + "time" +) + +func main() { + smidge.DEBUG = log.New(os.Stdout, "", 0) + c, err := smidge.NewClient("udp://127.0.0.1:1884", "gotestclient") + if err != nil { + panic(err) + } + c.SetWill("testwill", 1, true, []byte("Will test")) + ct := c.Connect() + ct.Wait() + fmt.Println(ct.ReturnCode) + rt := c.Register("gotest") + rt.WaitTimeout(10 * time.Second) + fmt.Println(rt.TopicName, rt.TopicId, rt.ReturnCode) + st := c.Subscribe("gotest", 0, func(c *smidge.SNClient, m *packets.PublishMessage) { + fmt.Println(string(m.Data)) + }) + st.WaitTimeout(10 * time.Second) + fmt.Println(st.TopicName, st.Qos, st.TopicId, st.ReturnCode) + pt := c.Publish("gotest", 1, false, []byte("Hello smidge")) + pt.WaitTimeout(10 * time.Second) + fmt.Println(pt.TopicId, pt.ReturnCode) + time.Sleep(3 * time.Second) +}
diff --git a/client/samples/simple2.go b/client/samples/simple2.go new file mode 100644 index 0000000..77f8b4d --- /dev/null +++ b/client/samples/simple2.go
@@ -0,0 +1,46 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package main + +import ( + "fmt" + "log" + "os" + "smidge/client" + "smidge/packets" + "time" +) + +func main() { + smidge.DEBUG = log.New(os.Stdout, "", 0) + c, err := smidge.NewClient("udp://127.0.0.1:1884", "gotestclient2") + if err != nil { + panic(err) + } + //c.SetWill("testwill", 1, true, []byte("Will test")) + ct := c.Connect() + ct.Wait() + fmt.Println(ct.ReturnCode) + c.DefaultMessageHandler = func(c *smidge.SNClient, m *packets.PublishMessage) { + fmt.Println(string(m.Data)) + } + // rt := c.Register("gotest") + // rt.WaitTimeout(10 * time.Second) + // fmt.Println(rt.TopicName, rt.TopicId, rt.ReturnCode) + st := c.Subscribe("#", 0, nil) + st.WaitTimeout(10 * time.Second) + fmt.Println(st.TopicName, st.Qos, st.TopicId, st.ReturnCode) + // pt := c.Publish("gotest", 1, false, []byte("Hello smidge")) + // pt.WaitTimeout(10 * time.Second) + // fmt.Println(pt.TopicId, pt.ReturnCode) + time.Sleep(3 * time.Second) +}
diff --git a/client/token.go b/client/token.go new file mode 100644 index 0000000..0747ba3 --- /dev/null +++ b/client/token.go
@@ -0,0 +1,112 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package smidge + +import ( + . "smidge/packets" + "time" +) + +type MessageAndToken struct { + m Message + t Token +} + +type Token interface { + Wait() + WaitTimeout(time.Duration) + flowComplete() +} + +type baseToken struct { + complete chan struct{} + ready bool + err error +} + +// Wait will wait indefinitely for the Token to complete, ie the Publish +// to be sent and confirmed receipt from the broker +func (b *baseToken) Wait() { + if !b.ready { + <-b.complete + b.ready = true + } +} + +// WaitTimeout takes a time in ms +func (b *baseToken) WaitTimeout(d time.Duration) { + if !b.ready { + select { + case <-b.complete: + b.ready = true + case <-time.After(d): + } + } +} + +func (b *baseToken) flowComplete() { + close(b.complete) +} + +func newToken(tType byte) Token { + switch tType { + case CONNECT: + return &ConnectToken{baseToken: baseToken{complete: make(chan struct{})}} + case REGISTER: + return &RegisterToken{baseToken: baseToken{complete: make(chan struct{})}} + case SUBSCRIBE: + return &SubscribeToken{baseToken: baseToken{complete: make(chan struct{})}} + case PUBLISH: + return &PublishToken{baseToken: baseToken{complete: make(chan struct{})}} + case UNSUBSCRIBE: + return &UnsubscribeToken{baseToken: baseToken{complete: make(chan struct{})}} + case WILLTOPIC, WILLMSG, WILLMSGUPD, WILLTOPICUPD: + return &WillToken{baseToken: baseToken{complete: make(chan struct{})}} + } + return nil +} + +type ConnectToken struct { + baseToken + ReturnCode byte +} + +type RegisterToken struct { + baseToken + TopicName string + TopicId uint16 + ReturnCode byte +} + +type PublishToken struct { + baseToken + TopicId uint16 + ReturnCode byte +} + +type SubscribeToken struct { + baseToken + handler MessageHandler + topicType byte + Qos byte + TopicName string + TopicId uint16 + ReturnCode byte +} + +type UnsubscribeToken struct { + baseToken +} + +type WillToken struct { + baseToken +}
diff --git a/client/trace.go b/client/trace.go new file mode 100644 index 0000000..a2673d3 --- /dev/null +++ b/client/trace.go
@@ -0,0 +1,44 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package smidge + +import ( + "io/ioutil" + "log" +) + +var ( + ERROR *log.Logger + CRITICAL *log.Logger + WARN *log.Logger + DEBUG *log.Logger +) + +const ( + NET = "[net] " + PNG = "[pinger] " + CLI = "[client] " + DEC = "[decode] " + MES = "[message] " + STR = "[store] " + MID = "[msgids] " + TST = "[test] " + STA = "[state] " + ERR = "[error] " +) + +func init() { + ERROR = log.New(ioutil.Discard, "", 0) + CRITICAL = log.New(ioutil.Discard, "", 0) + WARN = log.New(ioutil.Discard, "", 0) + DEBUG = log.New(ioutil.Discard, "", 0) +}
diff --git a/gateway/README.md b/gateway/README.md new file mode 100644 index 0000000..98ff491 --- /dev/null +++ b/gateway/README.md
@@ -0,0 +1,5 @@ +Smidge-Gateway +===== + +An [MQTT-SN](http://mqtt.org/new/wp-content/uploads/2009/06/MQTT-SN_spec_v1.2.pdf) Gateway implementation in Go +
diff --git a/gateway/config b/gateway/config new file mode 100644 index 0000000..df1f96d --- /dev/null +++ b/gateway/config
@@ -0,0 +1,4 @@ +mode transparent +port 1884 +mqtt-broker tcp://iot.eclipse.org:1883 +mqtt-timeout 90 \ No newline at end of file
diff --git a/gateway/gate/aggregatinggateway.go b/gateway/gate/aggregatinggateway.go new file mode 100644 index 0000000..9020bc0 --- /dev/null +++ b/gateway/gate/aggregatinggateway.go
@@ -0,0 +1,449 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "bytes" + "log" + "net" + "os" + "sync" + "time" + + MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git" + + . "smidge/packets" +) + +type AGateway struct { + mqttclient *MQTT.Client + stopsig chan os.Signal + port int + tIndex topicNames + tTree *TopicTree + clients Clients + handler MQTT.MessageHandler +} + +func NewAGateway(gc *GatewayConfig, stopsig chan os.Signal) *AGateway { + MQTT.WARN = log.New(os.Stdout, "", 0) + MQTT.DEBUG = log.New(os.Stdout, "", 0) + MQTT.CRITICAL = log.New(os.Stdout, "", 0) + MQTT.ERROR = log.New(os.Stdout, "", 0) + opts := MQTT.NewClientOptions() + opts.AddBroker(gc.mqttbroker) + if gc.mqttuser != "" { + opts.SetUsername(gc.mqttuser) + } + if gc.mqttpassword != "" { + opts.SetPassword(gc.mqttpassword) + } + if gc.mqttclientid != "" { + opts.SetClientID(gc.mqttclientid) + } + if gc.mqtttimeout > 0 { + opts.SetKeepAlive(time.Duration(gc.mqtttimeout)) + } + client := MQTT.NewClient(opts) + ag := &AGateway{ + client, + stopsig, + gc.port, + topicNames{ + sync.RWMutex{}, + make(map[uint16]string), + 0, + }, + NewTopicTree(), + Clients{ + sync.RWMutex{}, + make(map[string]SNClient), + }, + nil, + } + + ag.handler = func(client *MQTT.Client, msg MQTT.Message) { + ag.distribute(msg) + } + + return ag +} + +func (ag *AGateway) Port() int { + return ag.port +} + +func (ag *AGateway) Start() { + go ag.awaitStop() + INFO.Println("Aggregating Gateway is starting") + if token := ag.mqttclient.Connect(); token.Wait() && token.Error() != nil { + ERROR.Println(token.Error()) + return + } + INFO.Println("Aggregating Gateway is started") + listen(ag) +} + +// This does NOT WORK on Windows using Cygwin, however +// it does work using cmd.exe +func (ag *AGateway) awaitStop() { + <-ag.stopsig + INFO.Println("Aggregating Gateway is stopping") + ag.mqttclient.Disconnect(500) + time.Sleep(500) //give broker some time to process DISCONNECT + INFO.Println("Aggregating Gateway is stopped") + + // TODO: cleanly close down other goroutines + + os.Exit(0) +} + +func (ag *AGateway) distribute(msg MQTT.Message) { + topic := msg.Topic() + INFO.Printf("AG distributing a msg for topic \"%s\"\n", topic) + + // collect a list of clients to which msg should be + // published + // then publish msg to those clients (async) + + if clients, e := ag.tTree.SubscribersOf(topic); e != nil { + ERROR.Println(e) + } else { + for _, client := range clients { + go ag.publish(msg, client) + } + } +} + +func (ag *AGateway) publish(msg MQTT.Message, client *Client) { + INFO.Printf("publish to client \"%s\"... ", client.ClientId) + topicid := ag.tIndex.getId(msg.Topic()) + // topicidtype := byte(0x00) // todo: pre-defined (1) and shortname (2) + // msgid := uint16(0x00) // todo: what should this be?? + pm := NewPublishMessage(topicid, 0x00, msg.Payload(), msg.Qos(), 0x00, msg.Retained(), msg.Duplicate()) + + if client.Registered(topicid) { + INFO.Printf("client \"%s\" already registered to %d, publish ahoy!\n", client, topicid) + if err := client.Write(pm); err != nil { + ERROR.Println(err) + } else { + INFO.Printf("published a message to \"%s\"\n", client) + } + } else { + INFO.Printf("client \"%s\" is not registered to %d, must REGISTER first\n", client, topicid) + rm := NewRegisterMessage(topicid, 0x00, []byte(msg.Topic())) + client.AddPendingMessage(pm) + if err := client.Write(rm); err != nil { + ERROR.Printf("error writing REGISTER to \"%s\"\n", client) + } else { + INFO.Printf("sent REGISTER to \"%s\" for %d (%d bytes)\n", client, topicid, rm.Length) + } + } +} + +func (ag *AGateway) OnPacket(nbytes int, buffer []byte, con *net.UDPConn, addr *net.UDPAddr) { + INFO.Printf("OnPacket! - bytes: %s\n", string(buffer[0:nbytes])) + + buf := bytes.NewBuffer(buffer) + rawmsg, _ := ReadPacket(buf) + INFO.Printf("rawmsg.MessageType(): %s\n", rawmsg.MessageType()) + + switch msg := rawmsg.(type) { + case *AdvertiseMessage: + ag.handle_ADVERTISE(msg, addr) + case *SearchGwMessage: + ag.handle_SEARCHGW(msg, addr) + case *GwInfoMessage: + ag.handle_GWINFO(msg, addr) + case *ConnectMessage: + ag.handle_CONNECT(msg, con, addr) + case *ConnackMessage: + ag.handle_CONNACK(msg, addr) + case *WillTopicReqMessage: + ag.handle_WILLTOPICREQ(msg, addr) + case *WillTopicMessage: + ag.handle_WILLTOPIC(msg, addr) + case *WillMsgReqMessage: + ag.handle_WILLMSGREQ(msg, addr) + case *WillMsgMessage: + ag.handle_WILLMSG(msg, addr) + case *RegisterMessage: + ag.handle_REGISTER(msg, con, addr) + case *RegackMessage: + ag.handle_REGACK(msg, addr) + case *PublishMessage: + ag.handle_PUBLISH(msg, addr) + case *PubackMessage: + ag.handle_PUBACK(msg, addr) + case *PubcompMessage: + ag.handle_PUBCOMP(msg, addr) + case *PubrecMessage: + ag.handle_PUBREC(msg, addr) + case *PubrelMessage: + ag.handle_PUBREL(msg, addr) + case *SubscribeMessage: + ag.handle_SUBSCRIBE(msg, con, addr) + case *SubackMessage: + ag.handle_SUBACK(msg, addr) + case *UnsubscribeMessage: + ag.handle_UNSUBSCRIBE(msg, addr) + case *UnsubackMessage: + ag.handle_UNSUBACK(msg, addr) + case *PingreqMessage: + ag.handle_PINGREQ(msg, con, addr) + case *PingrespMessage: + ag.handle_PINGRESP(msg, addr) + case *DisconnectMessage: + ag.handle_DISCONNECT(msg, addr) + case *WillTopicUpdateMessage: + ag.handle_WILLTOPICUPD(msg, addr) + case *WillTopicRespMessage: + ag.handle_WILLTOPICRESP(msg, addr) + case *WillMsgUpdateMessage: + ag.handle_WILLMSGUPD(msg, addr) + case *WillMsgRespMessage: + ag.handle_WILLMSGRESP(msg, addr) + default: + ERROR.Printf("Unknown Message Type %T\n", msg) + } +} + +func (ag *AGateway) handle_ADVERTISE(m *AdvertiseMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_SEARCHGW(m *SearchGwMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_GWINFO(m *GwInfoMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_CONNECT(m *ConnectMessage, c *net.UDPConn, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + + if clientid, e := validateClientId(m.ClientId); e != nil { + ERROR.Println(e) + } else { + INFO.Printf("clientid: %s\n", clientid) + INFO.Printf("remoteaddr: %s\n", r) + INFO.Printf("will: %v\n", m.Will) + + if m.Will { + // todo: do something about that + } + + client := NewClient(clientid, c, r) + ag.clients.AddClient(client) + + ca := NewMessage(CONNACK).(*ConnackMessage) // todo: 0 ? + ca.ReturnCode = 0 + if ioerr := client.Write(ca); ioerr != nil { + ERROR.Println(ioerr) + } else { + INFO.Println("CONNACK was sent") + } + } +} + +func (ag *AGateway) handle_CONNACK(m *ConnackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLTOPICREQ(m *WillTopicReqMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLTOPIC(m *WillTopicMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLMSGREQ(m *WillMsgReqMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLMSG(m *WillMsgMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_REGISTER(m *RegisterMessage, c *net.UDPConn, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + topic := string(m.TopicName) + INFO.Printf("msg id: %d\n", m.MessageId) + INFO.Printf("topic name: %s\n", topic) + + var topicid uint16 + if !ag.tIndex.containsTopic(topic) { + topicid = ag.tIndex.putTopic(topic) + } else { + topicid = ag.tIndex.getId(topic) + } + + client := ag.clients.GetClient(r).(*Client) + client.Register(topicid, topic) + + INFO.Printf("ag topicid: %d\n", topicid) + + ra := NewRegackMessage(topicid, m.MessageId, 0) + INFO.Printf("ra.MsgId: %d\n", ra.MessageId) + + if err := client.Write(ra); err != nil { + ERROR.Println(err) + } else { + INFO.Println("REGACK sent") + } +} + +func (ag *AGateway) handle_REGACK(m *RegackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + // the gateway sends a register when there is a message + // that needs to be published, so we do that now + topicid := m.TopicId + client := ag.clients.GetClient(r).(*Client) + pm := client.FetchPendingMessage(topicid) + if pm == nil { + ERROR.Printf("no pending message for %s id %d\n", client, topicid) + } else { + if err := client.Write(pm); err != nil { + ERROR.Println(err) + } else { + INFO.Printf("published a pending message to \"%s\"\n", client) + } + } +} + +func (ag *AGateway) handle_PUBLISH(m *PublishMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + + INFO.Printf("m.TopicId: %d\n", m.TopicId) + INFO.Printf("m.Data: %s\n", string(m.Data)) + + topic := ag.tIndex.getTopic(m.TopicId) + + // TODO: what should the MQTT-QoS be set as? In case of MQTTSN-QoS -1 ? + if token := ag.mqttclient.Publish(topic, m.Qos, m.Retain, m.Data); token.WaitTimeout(2000) && token.Error() != nil { + ERROR.Println("Error publishing message", token.Error()) + } + INFO.Println("Message Published") +} + +func (ag *AGateway) handle_PUBACK(m *PubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_PUBCOMP(m *PubcompMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_PUBREC(m *PubrecMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_PUBREL(m *PubrelMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_SUBSCRIBE(m *SubscribeMessage, c *net.UDPConn, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + INFO.Printf("m.TopicIdType: %d\n", m.TopicIdType) + topic := string(m.TopicName) + var topicid uint16 + if m.TopicIdType == 0 { + INFO.Printf("m.TopicName: %s\n", topic) + if !ContainsWildcard(topic) { + topicid = ag.tIndex.getId(topic) + if topicid == 0 { + topicid = ag.tIndex.putTopic(topic) + } + } else { + // todo: if topic contains wildcard, something about REGISTER + // at a later time, but send topic id 0x0000 for now + } + } // todo: other topic id types + + client := ag.clients.GetClient(r).(*Client) + if first, err := ag.tTree.AddSubscription(client, topic); err != nil { + INFO.Println("error adding subscription: %v\n", err) + // todo: suback an error message? + } else { + if first { + INFO.Println("first subscriber of subscription, subscribbing via MQTT") + if token := ag.mqttclient.Subscribe(topic, 2, ag.handler); token.WaitTimeout(2000) && token.Error() != nil { + ERROR.Println("Error subscribing,", token.Error()) + } + } + // AG is subscribed at this point + client.Register(topicid, topic) + suba := NewSubackMessage(topicid, m.MessageId, m.Qos, 0) + var buf bytes.Buffer + suba.Write(&buf) + if nbytes, err := c.WriteToUDP(buf.Bytes(), r); err != nil { + ERROR.Println(err) + } else { + INFO.Printf("SUBACK sent %d bytes\n", nbytes) + } + } +} + +func (ag *AGateway) handle_SUBACK(m *SubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_UNSUBSCRIBE(m *UnsubscribeMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_UNSUBACK(m *UnsubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_PINGREQ(m *PingreqMessage, c *net.UDPConn, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + resp := NewMessage(PINGRESP) + + var buf bytes.Buffer + resp.Write(&buf) + + if nbytes, err := c.WriteToUDP(buf.Bytes(), r); err != nil { + ERROR.Println(err) + } else { + INFO.Printf("PINGRESP sent %d bytes\n", nbytes) + } +} + +func (ag *AGateway) handle_PINGRESP(m *PingrespMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_DISCONNECT(m *DisconnectMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) + INFO.Printf("duration: %d\n", m.Duration) + // todo: cleanup the client +} + +func (ag *AGateway) handle_WILLTOPICUPD(m *WillTopicUpdateMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLTOPICRESP(m *WillTopicRespMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLMSGUPD(m *WillMsgUpdateMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +} + +func (ag *AGateway) handle_WILLMSGRESP(m *WillMsgRespMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", m.MessageType(), r) +}
diff --git a/gateway/gate/client.go b/gateway/gate/client.go new file mode 100644 index 0000000..827daa0 --- /dev/null +++ b/gateway/gate/client.go
@@ -0,0 +1,89 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "bytes" + "net" + "sync" + + . "smidge/packets" +) + +type SNClient interface { + AddrString() string +} + +type Client struct { + sync.RWMutex + ClientId string + Conn *net.UDPConn + Address *net.UDPAddr + registeredTopics map[uint16]string + pendingMessages map[uint16]*PublishMessage +} + +func NewClient(ClientId string, Conn *net.UDPConn, Address *net.UDPAddr) *Client { + INFO.Printf("NewClient, id: \"%s\"\n", ClientId) + return &Client{ + sync.RWMutex{}, + ClientId, + Conn, + Address, + make(map[uint16]string), + make(map[uint16]*PublishMessage), + } +} + +func (c *Client) Write(m Message) error { + var buf bytes.Buffer + m.Write(&buf) + _, e := c.Conn.WriteToUDP(buf.Bytes(), c.Address) + return e +} + +func (c *Client) Register(topicId uint16, topic string) { + defer c.Unlock() + c.Lock() + INFO.Printf("client %s registered topicId %d\n", c.ClientId, topicId) + c.registeredTopics[topicId] = topic +} + +func (c *Client) Registered(topicId uint16) bool { + defer c.RUnlock() + c.RLock() + _, ok := c.registeredTopics[topicId] + return ok +} + +func (c *Client) AddPendingMessage(p *PublishMessage) { + defer c.Unlock() + c.Lock() + c.pendingMessages[p.TopicId] = p +} + +func (c *Client) FetchPendingMessage(topicId uint16) *PublishMessage { + defer c.Unlock() + c.Lock() + pm := c.pendingMessages[topicId] + delete(c.pendingMessages, topicId) + return pm +} + +func (c *Client) AddrString() string { + return c.Address.String() +} + +func (c *Client) String() string { + return c.ClientId +}
diff --git a/gateway/gate/clients.go b/gateway/gate/clients.go new file mode 100644 index 0000000..00dbeb2 --- /dev/null +++ b/gateway/gate/clients.go
@@ -0,0 +1,56 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "net" + "sync" +) + +type Clients struct { + sync.RWMutex + // indexed by "address:port" => StorableClient + clients map[string]SNClient +} + +func (c *Clients) GetClient(addr *net.UDPAddr) SNClient { + defer c.RUnlock() + c.RLock() + return c.clients[addr.String()] +} + +// Return true if this is a new client, false otherwise +// Clients are indexed by their address:port b/c +// that's the only indentifying information we have +// outside of a CONNECT packet +func (c *Clients) AddClient(client SNClient) bool { + defer c.Unlock() + c.Lock() + addr := client.AddrString() + INFO.Println("AddClient(%s - %s)", client, addr) + isNew := false + if c.clients[addr] == nil { + isNew = true + } + //todo: what to do if clientid is in use? + // is there some cleanup involved in topictree? + c.clients[addr] = client + return isNew +} + +func (c *Clients) RemoveClient(id string) { + defer c.Unlock() + c.Lock() + INFO.Println("RemoveClient(%s)", id) + delete(c.clients, id) +}
diff --git a/gateway/gate/config.go b/gateway/gate/config.go new file mode 100644 index 0000000..2284059 --- /dev/null +++ b/gateway/gate/config.go
@@ -0,0 +1,145 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "bufio" + "bytes" + "io/ioutil" + "strconv" + "strings" +) + +type GatewayConfig struct { + aggregating bool + port int + mqttbroker string + mqttuser string + mqttpassword string + mqttclientid string + mqtttimeout int +} + +func (gc *GatewayConfig) IsAggregating() bool { + return gc.aggregating +} + +func ParseConfigFile(file string) (*GatewayConfig, error) { + gc := &GatewayConfig{} + if bytes, rerr := ioutil.ReadFile(file); rerr != nil { + return nil, rerr + } else { + if perr := gc.parseConfig(string(bytes)); perr != nil { + return nil, perr + } + } + return gc, nil +} + +func (gc *GatewayConfig) parseConfig(config string) error { + scanner := bufio.NewScanner(bytes.NewReader([]byte(config))) + scanner.Split(bufio.ScanLines) + + var lineno int + for scanner.Scan() { + line := scanner.Text() + lineno++ + if k, v, e := gc.parseLine(line); e != nil { + return e + } else if k == "" && v == "" { + // skipping comment or blank line + } else { + if e = gc.setOption(k, v); e != nil { + ERROR.Printf("Error in configuration on line %d\n", lineno) + return e + } + } + } + return nil +} + +func (gc *GatewayConfig) parseLine(line string) (string, string, error) { + line = strings.TrimSpace(line) + if len(line) == 0 || line[0] == '#' { + return "", "", nil + } + fields := strings.Fields(line) + if len(fields) == 1 { + ERROR.Printf("Missing value for config option: \"%s\"", fields[0]) + return "", "", ErrMissingValueForConfigOption + } else if len(fields) > 2 { + ERROR.Printf("Too many values supplied for config option: \"%s\"", fields[0]) + return "", "", ErrTooManyValuesForConfigOption + } + return fields[0], fields[1], nil +} + +func (gc *GatewayConfig) setOption(key, value string) error { + var e error + switch key { + case "mode": + gc.aggregating, e = checkMode(value) + case "port": + gc.port, e = checkNum("port", value) + case "mqtt-broker": + gc.mqttbroker, e = checkURI(value) + case "mqtt-user": + gc.mqttuser = value + case "mqtt-password": + gc.mqttpassword = value + case "mqtt-clientid": + gc.mqttclientid = value + case "mqtt-timeout": + gc.mqtttimeout, e = checkNum("mqtt-timeout", value) + default: + ERROR.Printf("Unknown config option: \"%s\"", key) + return ErrUnknownConfigOption + } + return e +} + +func checkURI(value string) (string, error) { + if value[0:6] != "tcp://" && + value[0:6] != "ssl://" && + value[0:6] != "tls://" && + value[0:7] != "tcps://" { + ERROR.Printf("Invalid URI, must specify transport (ex: \"tcp://\"): \"%s\"", value) + return "", ErrNoTransportSpecified + } + // todo: check that a port is provided + // also there is probably a library way to verify a URI + return value, nil +} + +func checkMode(value string) (bool, error) { + var isAggregating bool + switch value { + case "aggregating": + isAggregating = true + case "transparent": + isAggregating = false + default: + ERROR.Printf("Invalid value specified for \"mode\": \"%s\"", value) + return false, ErrInvalidModeSpecified + } + return isAggregating, nil +} + +func checkNum(label, value string) (int, error) { + if p, e := strconv.Atoi(value); e != nil { + ERROR.Printf("Invalid value specified for \"%s\" (not a number): \"%s\"", label, value) + return 0, ErrNotANumber + } else { + return p, nil + } +}
diff --git a/gateway/gate/err.go b/gateway/gate/err.go new file mode 100644 index 0000000..067f003 --- /dev/null +++ b/gateway/gate/err.go
@@ -0,0 +1,48 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "errors" +) + +var ( + /* Config Errors */ + ErrMissingValueForConfigOption = errors.New("Missing value for config option") + ErrTooManyValuesForConfigOption = errors.New("Too many values for config option") + ErrUnknownConfigOption = errors.New("Unknown config option") + ErrNoTransportSpecified = errors.New("Missing transport") + ErrInvalidModeSpecified = errors.New("Invalid mode") + ErrNotANumber = errors.New("Not a number") + + /* Protocol Errors */ + ErrZeroLengthClientID = errors.New("Zero-length clientID is invalid") + ErrClientIDTooLong = errors.New("ClientID too long") + + /* Topic Errors */ + ErrTopicFilterEmptyString = errors.New("TopicFilter cannot be empty string") + ErrTopicFilterInvalidWildcard = errors.New("TopicFilter contains invalid wildcard") + ErrTopicNameEmptyString = errors.New("TopicName cannot be empty string") + ErrTopicNameContainsWildcard = errors.New("TopicName cannot contain wildcard") + + /* Topic Tree Errors */ + ErrNoSuchSubscriptionExists = errors.New("Subscription does not exist") + ErrNoSubscribers = errors.New("No subscribers") + ErrClientNotSubscribed = errors.New("Client not subscribed") +) + +func chkerr(e error) { + if e != nil { + panic(e) + } +}
diff --git a/gateway/gate/gateway.go b/gateway/gate/gateway.go new file mode 100644 index 0000000..410d583 --- /dev/null +++ b/gateway/gate/gateway.go
@@ -0,0 +1,23 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "net" +) + +type Gateway interface { + Start() + Port() int + OnPacket(int, []byte, *net.UDPConn, *net.UDPAddr) +}
diff --git a/gateway/gate/log.go b/gateway/gate/log.go new file mode 100644 index 0000000..372bb70 --- /dev/null +++ b/gateway/gate/log.go
@@ -0,0 +1,28 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "io" + "log" +) + +var ( + INFO *log.Logger + ERROR *log.Logger +) + +func InitLogger(infoHandle, errorHandle io.Writer) { + INFO = log.New(infoHandle, "INFO: ", log.Ldate|log.Ltime) + ERROR = log.New(errorHandle, "ERROR: ", log.Ldate|log.Ltime) +}
diff --git a/gateway/gate/shared.go b/gateway/gate/shared.go new file mode 100644 index 0000000..fce3e3d --- /dev/null +++ b/gateway/gate/shared.go
@@ -0,0 +1,25 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +func validateClientId(clientid []byte) (string, error) { + if len(clientid) == 0 { + ERROR.Println("zero length client id not allowed") + return "", ErrZeroLengthClientID + } + if len(clientid) > 23 { + ERROR.Println("client id longer than 23 characters") + return "", ErrClientIDTooLong + } + return string(clientid), nil +}
diff --git a/gateway/gate/topics.go b/gateway/gate/topics.go new file mode 100644 index 0000000..9829dba --- /dev/null +++ b/gateway/gate/topics.go
@@ -0,0 +1,127 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "strings" + "sync" +) + +// Topic Names and Topic Filters +// The MQTT v3.1.1 spec clarifies a number of ambiguities with regard +// to the validity of Topic strings. +// - A Topic must be between 1 and 65535 bytes. +// - A Topic is case sensitive. +// - A Topic may contain whitespace. +// - A Topic containing a leading forward slash is different than a Topic without. +// - A Topic may be "/" (two levels, both empty string). +// - A Topic must be UTF-8 encoded. +// - A Topic may contain any number of levels. +// - A Topic may contain an empty level (two forward slashes in a row). +// - A TopicName may not contain a wildcard. +// - A TopicFilter may only have a # (multi-level) wildcard as the last level. +// - A TopicFilter may contain any number of + (single-level) wildcards. +// - A TopicFilter with a # will match the absense of a level +// Example: a subscription to "foo/#" will match messages published to "foo". + +func ContainsWildcard(topic string) bool { + if len(topic) == 1 && (topic == "+" || topic == "#") { + return true + } + if len(topic) > 1 && (topic[len(topic)-2:] == "/#" || topic[len(topic)-2:] == "/+") { + return true + } + return strings.Contains(topic, "/+/") +} + +func ValidateTopicFilter(topic string) ([]string, error) { + if len(topic) == 0 { + return nil, ErrTopicFilterEmptyString + } + + levels := strings.Split(topic, "/") + for i, level := range levels { + if level == "#" && i != len(levels)-1 { + return nil, ErrTopicFilterInvalidWildcard + } + } + return levels, nil +} + +func ValidateTopicName(topic string) ([]string, error) { + if len(topic) == 0 { + return nil, ErrTopicNameEmptyString + } + + levels := strings.Split(topic, "/") + for _, level := range levels { + if level == "#" || level == "+" { + return nil, ErrTopicNameContainsWildcard + } + } + return levels, nil +} + +// This needs to be efficient for indexing by topicId. +// However, it is necessary when adding a new topic to index +// by topic name (to check if it already exists). We optimze +// for the former case. +type topicNames struct { + sync.RWMutex + contents map[uint16]string + next uint16 +} + +// O(n) +func (repo *topicNames) containsTopic(topic string) bool { + return repo.getId(topic) != 0 +} + +// O(1) +func (repo *topicNames) containsId(id uint16) bool { + return repo.getTopic(id) != "" +} + +// O(n) +func (repo *topicNames) getId(topic string) uint16 { + defer repo.RUnlock() + repo.RLock() + var topicid uint16 + for id, topicVal := range repo.contents { + if topicVal == topic { + topicid = id + break + } + } + INFO.Printf("get[%s] -> %d\n", topic, topicid) + return topicid +} + +// O(1) +func (repo *topicNames) getTopic(id uint16) string { + defer repo.RUnlock() + repo.RLock() + topic := repo.contents[id] + INFO.Printf("getTopic[%d] -> %s\n", id, topic) + return topic +} + +// O(1) +func (repo *topicNames) putTopic(topic string) uint16 { + defer repo.Unlock() + repo.Lock() + repo.next++ + repo.contents[repo.next] = topic + INFO.Printf("put[%d] -> %s\n", repo.next, topic) + return repo.next +}
diff --git a/gateway/gate/topictree.go b/gateway/gate/topictree.go new file mode 100644 index 0000000..79e3c71 --- /dev/null +++ b/gateway/gate/topictree.go
@@ -0,0 +1,152 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "sync" +) + +type TopicTree struct { + sync.RWMutex + root *node +} + +type node struct { + clients []*Client + children map[string]*node +} + +func newNode() *node { + return &node{ + make([]*Client, 0), + make(map[string]*node), + } +} + +// return true if level needed to be created, false otherwise +func (n *node) goTo(level string) (*node, bool) { + created := false + if n.children[level] == nil { + n.children[level] = newNode() + created = true + } + return n.children[level], created +} + +// return true if this is the first client to be added +// to this node (representing a subscription) +func (n *node) addClient(client *Client) bool { + isFirst := len(n.clients) == 0 + n.clients = append(n.clients, client) + return isFirst +} + +func NewTopicTree() *TopicTree { + var t TopicTree + t.root = newNode() + return &t +} + +// return false if there is already a subscriber of this topic, +// true if this is the first subscriber +func (tt *TopicTree) AddSubscription(client *Client, topic string) (bool, error) { + defer tt.Unlock() + tt.Lock() + INFO.Printf("AddSubscription(\"%s\", \"%s\")\n", client.ClientId, topic) + if levels, e := ValidateTopicFilter(topic); e != nil { + return false, e + } else { + n := tt.root + // walk the tree following the path of topic, creating new + // nodes as necessary + for _, level := range levels { + n, _ = n.goTo(level) + } + return n.addClient(client), nil + } +} + +// topic could contain wild cards, however we do only consider the literal +// topic string - (wilds are not evaluated for this) +func (tt *TopicTree) RemoveSubscription(s *Client, topic string) error { + defer tt.Unlock() + tt.Lock() + if levels, e := ValidateTopicFilter(topic); e != nil { + return e + } else { + n := tt.root + for _, level := range levels { + if n = n.children[level]; n == nil { + ERROR.Printf("no subscription exists \"%s\"\n", topic) + return ErrNoSuchSubscriptionExists + } + } + if len(n.clients) < 1 { + ERROR.Printf("no clients of subscription \"%s\"\n", topic) + return ErrNoSubscribers + } + for i := 0; i < len(n.clients); i++ { + if n.clients[i].ClientId == s.ClientId { + // inexpensive way of removing from a slice + n.clients[i] = n.clients[len(n.clients)-1] + n.clients = n.clients[0 : len(n.clients)-1] + INFO.Printf("deleted subscription of client \"%s\"\n", s.ClientId) + return nil + } + } + ERROR.Printf("client \"%s\" was not subscribed to \"%s\"\n", s.ClientId, topic) + return ErrClientNotSubscribed + } +} + +// topic MUST be valid (ie no wild cards, no empty level, no ending slash) +func (tt *TopicTree) SubscribersOf(topic string) ([]*Client, error) { + defer tt.RUnlock() + tt.RLock() + + clients := make([]*Client, 0) + + n := tt.root + if levels, e := ValidateTopicName(topic); e != nil { + return nil, e + } else { + subscribers(n, levels, &clients) + return clients, nil + } +} + +func subscribers(n *node, levels []string, clients *[]*Client) { + if len(levels) == 0 { + return + } + + if hash := n.children["#"]; hash != nil { + *clients = append(*clients, hash.clients...) + } + + if plus := n.children["+"]; plus != nil { + if len(levels) == 1 { + *clients = append(*clients, plus.clients...) + } else { + subscribers(plus, levels[1:], clients) + } + } + + if match := n.children[levels[0]]; match != nil { + if len(levels) == 1 { + *clients = append(*clients, match.clients...) + } else { + subscribers(match, levels[1:], clients) + } + } +}
diff --git a/gateway/gate/transparentclient.go b/gateway/gate/transparentclient.go new file mode 100644 index 0000000..07195c4 --- /dev/null +++ b/gateway/gate/transparentclient.go
@@ -0,0 +1,97 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "net" + "sync" + + . "smidge/packets" + + MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git" +) + +type TClient struct { + Client + mqttClient *MQTT.Client + mqttBroker string + username string + password string +} + +// Do not allow the creation of an MQTT-SN client if +// a connection to the MQTT broker cannot be established +func NewTClient(ClientId, Broker string, Connection *net.UDPConn, Address *net.UDPAddr) (*TClient, error) { + INFO.Println("NewTClient, id: %s", ClientId) + t := &TClient{ + Client{ + sync.RWMutex{}, + ClientId, + Connection, + Address, + make(map[uint16]string), + make(map[uint16]*PublishMessage), + }, + nil, + Broker, + "", + "", + } + if err := t.connectMQTT(ClientId, Broker); err != nil { + return nil, err + } + return t, nil +} + +func (t *TClient) connectMQTT(ClientId, Broker string) error { + opts := MQTT.NewClientOptions() + opts.AddBroker(Broker) + opts.SetClientID(ClientId) + if t.username != "" { + opts.SetUsername(t.username) + opts.SetPassword(t.password) + } + t.mqttClient = MQTT.NewClient(opts) + + if token := t.mqttClient.Connect(); token.Wait() && token.Error() != nil { + return token.Error() + } + INFO.Println("TClient connected to mqtt broker") + return nil +} + +func (t *TClient) disconnectMQTT() { + t.mqttClient.Disconnect(100) +} + +func (t *TClient) subscribeMQTT(qos byte, topic string, tIndex *topicNames) { + var handler MQTT.MessageHandler = func(client *MQTT.Client, msg MQTT.Message) { + INFO.Println("publish handler") + + tid := tIndex.getId(msg.Topic()) + // is topicid type always 0 coming out of tIndex ? + // todo: msgid is not always 0 + pm := NewPublishMessage(tid, 0x00, msg.Payload(), msg.Qos(), 0x00, msg.Retained(), msg.Duplicate()) + + if err := t.Write(pm); err != nil { + ERROR.Println(err) + } else { + INFO.Println("incoming mqtt published to mqtt-sn") + } + } + + if token := t.mqttClient.Subscribe(topic, qos, handler); token.WaitTimeout(2000) && token.Error() != nil { + ERROR.Println("Error subscribing,", token.Error()) + } + INFO.Println(t.ClientId, "subscribed to", topic) +}
diff --git a/gateway/gate/transparentgateway.go b/gateway/gate/transparentgateway.go new file mode 100644 index 0000000..6bedb3b --- /dev/null +++ b/gateway/gate/transparentgateway.go
@@ -0,0 +1,326 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "bytes" + "net" + "os" + "sync" + + . "smidge/packets" + + //MQTT "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git" +) + +type TGateway struct { + stopsig chan os.Signal + port int + mqttBroker string + clients Clients + tIndex topicNames +} + +func NewTGateway(gc *GatewayConfig, stopsig chan os.Signal) *TGateway { + t := &TGateway{ + stopsig, + gc.port, + gc.mqttbroker, + Clients{ + sync.RWMutex{}, + make(map[string]SNClient), + }, + topicNames{ + sync.RWMutex{}, + make(map[uint16]string), + 0, + }, + } + return t +} + +func (t *TGateway) Port() int { + return t.port +} + +func (t *TGateway) Start() { + go t.awaitStop() + INFO.Println("Transparent Gataway is started") + listen(t) +} + +func (t *TGateway) awaitStop() { + <-t.stopsig + INFO.Println("Transparent Gateway is stopped") + os.Exit(0) +} + +func (t *TGateway) OnPacket(nbytes int, buffer []byte, con *net.UDPConn, addr *net.UDPAddr) { + INFO.Println("TG OnPacket!") + INFO.Printf("bytes: %s\n", string(buffer[0:nbytes])) + + buf := bytes.NewBuffer(buffer) + rawmsg, _ := ReadPacket(buf) + + INFO.Printf("rawmsg.MessageType(): %s\n", MessageNames[rawmsg.MessageType()]) + + switch msg := rawmsg.(type) { + case *AdvertiseMessage: + t.handle_ADVERTISE(msg, addr) + case *SearchGwMessage: + t.handle_SEARCHGW(msg, addr) + case *GwInfoMessage: + t.handle_GWINFO(msg, addr) + case *ConnectMessage: + t.handle_CONNECT(msg, con, addr) + case *ConnackMessage: + t.handle_CONNACK(msg, addr) + case *WillTopicReqMessage: + t.handle_WILLTOPICREQ(msg, addr) + case *WillTopicMessage: + t.handle_WILLTOPIC(msg, addr) + case *WillMsgReqMessage: + t.handle_WILLMSGREQ(msg, addr) + case *WillMsgMessage: + t.handle_WILLMSG(msg, addr) + case *RegisterMessage: + t.handle_REGISTER(msg, con, addr) + case *RegackMessage: + t.handle_REGACK(msg, addr) + case *PublishMessage: + t.handle_PUBLISH(msg, addr) + case *PubackMessage: + t.handle_PUBACK(msg, addr) + case *PubcompMessage: + t.handle_PUBCOMP(msg, addr) + case *PubrecMessage: + t.handle_PUBREC(msg, addr) + case *PubrelMessage: + t.handle_PUBREL(msg, addr) + case *SubscribeMessage: + t.handle_SUBSCRIBE(msg, addr) + case *SubackMessage: + t.handle_SUBACK(msg, addr) + case *UnsubackMessage: + t.handle_UNSUBACK(msg, addr) + case *PingreqMessage: + t.handle_PINGREQ(msg, con, addr) + case *DisconnectMessage: + t.handle_DISCONNECT(msg, addr) + case *WillTopicUpdateMessage: + t.handle_WILLTOPICUPD(msg, addr) + case *WillTopicRespMessage: + t.handle_WILLTOPICRESP(msg, addr) + case *WillMsgUpdateMessage: + t.handle_WILLMSGUPD(msg, addr) + case *WillMsgRespMessage: + t.handle_WILLMSGRESP(msg, addr) + default: + ERROR.Printf("Unknown Message Type %T\n", msg) + } +} + +func (t *TGateway) handle_ADVERTISE(m *AdvertiseMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) +} + +func (t *TGateway) handle_SEARCHGW(m *SearchGwMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) +} + +func (t *TGateway) handle_GWINFO(m *GwInfoMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) +} + +func (t *TGateway) handle_CONNECT(m *ConnectMessage, c *net.UDPConn, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) + INFO.Println(m.ProtocolId, m.Duration, m.ClientId) + if clientid, err := validateClientId(m.ClientId); err != nil { + ERROR.Println(err) + } else { + INFO.Printf("clientid: %s\n", clientid) + INFO.Printf("remoteaddr: %s\n", a) + INFO.Printf("will: %v\n", m.Will) + if m.Will { + // todo: will msg + } + if tClient, err := NewTClient(string(clientid), t.mqttBroker, c, a); err != nil { + ERROR.Println(err) + } else { + t.clients.AddClient(tClient) + + // establish connection to mqtt broker + + ca := NewMessage(CONNACK).(*ConnackMessage) + ca.ReturnCode = 0 + if err = tClient.Write(ca); err != nil { + ERROR.Println(err) + } else { + INFO.Println("CONNACK was sent") + } + } + } +} + +func (t *TGateway) handle_CONNACK(m *ConnackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLTOPICREQ(m *WillTopicReqMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLTOPIC(m *WillTopicMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLMSGREQ(m *WillMsgReqMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLMSG(m *WillMsgMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_REGISTER(m *RegisterMessage, c *net.UDPConn, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) + topic := string(m.TopicName) + var topicid uint16 + if !t.tIndex.containsTopic(topic) { + topicid = t.tIndex.putTopic(topic) + } else { + topicid = t.tIndex.getId(topic) + } + + INFO.Printf("t topicid: %d\n", topicid) + + tclient := t.clients.GetClient(r).(*TClient) + tclient.Register(topicid, topic) + + ra := NewRegackMessage(topicid, m.MessageId, 0) + INFO.Printf("ra.Msgid: %d\n", ra.MessageId) + + if err := tclient.Write(ra); err != nil { + ERROR.Println(err) + } else { + INFO.Println("REGACK sent") + } +} + +func (t *TGateway) handle_REGACK(m *RegackMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) +} + +func (t *TGateway) handle_PUBLISH(m *PublishMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) + tclient := t.clients.GetClient(a).(*TClient) + + topic := t.tIndex.getTopic(m.TopicId) + + INFO.Println(topic, m.Qos, m.Retain, m.Data) + if token := tclient.mqttClient.Publish(topic, m.Qos, m.Retain, m.Data); token.WaitTimeout(2000) && token.Error() != nil { + ERROR.Println("Error publishing message", token.Error()) + return + } + INFO.Println("PUBLISH published") +} + +func (t *TGateway) handle_PUBACK(m *PubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_PUBCOMP(m *PubcompMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_PUBREC(m *PubrecMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_PUBREL(m *PubrelMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_SUBSCRIBE(m *SubscribeMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) + topic := "" + if m.TopicIdType == 0 { // todo: other topic id types, also use enum + topic = string(m.TopicName) + } else { + ERROR.Println("other topic id types not supported yet") + topic = "not_implemented" + } + tclient := t.clients.GetClient(r).(*TClient) + INFO.Printf("subscribe, qos: %d, topic: %s\n", m.Qos, topic) + tclient.subscribeMQTT(m.Qos, topic, &t.tIndex) + + suba := NewSubackMessage(0, m.MessageId, m.Qos, 0) + + if err := tclient.Write(suba); err != nil { + ERROR.Println(err) + } else { + INFO.Println("SUBACK sent") + } +} + +func (t *TGateway) handle_SUBACK(m *SubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_UNSUBSCRIBE(m *UnsubscribeMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_UNSUBACK(m *UnsubackMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_PINGREQ(m *PingreqMessage, c *net.UDPConn, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) + tclient := t.clients.GetClient(a).(*TClient) + + resp := NewMessage(PINGRESP) + + if err := tclient.Write(resp); err != nil { + ERROR.Println(err) + } else { + INFO.Println("PINGRESP sent") + } +} + +func (t *TGateway) handle_PINGRESP(m *PingrespMessage, a *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], a) +} + +func (t *TGateway) handle_DISCONNECT(m *DisconnectMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) + tclient := t.clients.GetClient(r).(*TClient) + tclient.disconnectMQTT() + t.clients.RemoveClient(tclient.ClientId) +} + +func (t *TGateway) handle_WILLTOPICUPD(m *WillTopicUpdateMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLTOPICRESP(m *WillTopicRespMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLMSGUPD(m *WillMsgUpdateMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +} + +func (t *TGateway) handle_WILLMSGRESP(m *WillMsgRespMessage, r *net.UDPAddr) { + INFO.Printf("handle_%s from %v\n", MessageNames[m.MessageType()], r) +}
diff --git a/gateway/gate/udp.go b/gateway/gate/udp.go new file mode 100644 index 0000000..608bc02 --- /dev/null +++ b/gateway/gate/udp.go
@@ -0,0 +1,35 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "fmt" + "net" +) + +func port2str(port int) string { + return fmt.Sprintf(":%d", port) +} + +func listen(g Gateway) { + address, err := net.ResolveUDPAddr("udp", port2str(g.Port())) + chkerr(err) + udpconn, err := net.ListenUDP("udp", address) + chkerr(err) + for { + buffer := make([]byte, 1024) + n, remote, err := udpconn.ReadFromUDP(buffer) + chkerr(err) + go g.OnPacket(n, buffer, udpconn, remote) + } +}
diff --git a/gateway/gate/unit_gate_test.go b/gateway/gate/unit_gate_test.go new file mode 100644 index 0000000..81ff43c --- /dev/null +++ b/gateway/gate/unit_gate_test.go
@@ -0,0 +1,105 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "sync" + "testing" +) + +func new_topicNames() *topicNames { + t := &topicNames{ + sync.RWMutex{}, + make(map[uint16]string), + 0, + } + return t +} + +func Test_topicName_contains(t *testing.T) { + topics := new_topicNames() + topics.containsTopic("notinthere") + + if topics.containsTopic("notinthere") != false { + t.Errorf("new topicNames contains something") + } + + topics.putTopic("alpha") + if topics.containsTopic("beta") != false { + t.Errorf("topicNames contains invalid topic") + } + + if topics.containsTopic("alpha") == false { + t.Errorf("topicNames missing topic") + } + + if topics.containsId(333) { + t.Errorf("topicNames contains rediculous topicId") + } + + if !topics.containsId(1) { + t.Errorf("topicNames is missing a topicId") + } + + if topics.containsId(0) { + t.Errorf("topicNames contains impossible topicId") + } +} + +func Test_topicName_putTopic(t *testing.T) { + topics := new_topicNames() + + i := topics.putTopic("foo") + if !topics.containsTopic("foo") { + t.Errorf("topicNames missing topic") + } + if i != 1 { + t.Errorf("topicNames put unexpected topicId") + } + + i = topics.putTopic("bar") + if !topics.containsTopic("bar") { + t.Errorf("topicNames missing 2nd topic") + } + if i != 2 { + t.Errorf("topicNames put unexpected topicId") + } + + if !topics.containsTopic("foo") { + t.Errorf("topicNames lost topic") + } +} + +func Test_topicName_get(t *testing.T) { + topics := new_topicNames() + + a := topics.putTopic("/a/b") + b := topics.getId("/a/b") + if a != b { + t.Errorf("topicNames did not get the same topic id as assigned by put") + } + b2 := topics.getTopic(1) + if b2 != "/a/b" { + t.Errorf("getTopic got wrong topic") + } + + c := topics.putTopic("d/c") + d := topics.getId("d/c") + e := topics.getId("/a/b") + if c != d { + t.Errorf("topicNames did not get the same topic id as assigned by put") + } + if c == e { + t.Errorf("topicName assigned same topic id to different topics") + } +}
diff --git a/gateway/gate/unit_topics_test.go b/gateway/gate/unit_topics_test.go new file mode 100644 index 0000000..af52e9a --- /dev/null +++ b/gateway/gate/unit_topics_test.go
@@ -0,0 +1,259 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +// Remember benchmarks can be run with the command +// go test -bench=".*" ./... + +import ( + "testing" +) + +func Test_ContainsWildcard(t *testing.T) { + topics := map[string]bool{ + "a": false, + "a/b": false, + "/a/b": false, + "+": true, + "/+": true, + "a/+": true, + "a/b/c/+": true, + "a/+/+/d": true, + "a/++/+d": false, + "#": true, + "/#": true, + "a/b/c/#": true, + "a/##/b": false, + } + + for topic, exp := range topics { + res := ContainsWildcard(topic) + if res != exp { + t.Errorf("ContainsWildcard expected \"%s\", got \"%s\"", exp, res) + } + } +} + +func Benchmark_ContainsWildcard(b *testing.B) { + topics := map[string]bool{ + "a": false, + "a/b": false, + "/a/b": false, + "+": true, + "/+": true, + "a/+": true, + "a/b/c/+": true, + "a/+/+/d": true, + "a/++/+d": false, + "#": true, + "/#": true, + "a/b/c/#": true, + "a/##/b": false, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z": false, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/#": true, + "+/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/+/s/+/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/+/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/+/y/+": true, + "a/b/+/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/+/w/x/+/z": true, + "+/+/+/d/+/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/+/t/u/v/w/x/+/#": true, + "a/b/c/d/e/f/g/h/i/j/k/+/+/n/+/p/+/r/+/t/u/v/w/x/+/z": true, + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/++/##/z": false, + } + + res := false + for i := 0; i < b.N; i++ { + for topic, _ := range topics { + res = res && ContainsWildcard(topic) + } + } +} + +func Test_ValidateTopicFilter(t *testing.T) { + topics := map[string]bool{ + "": false, + "a": true, + "#": true, + "+": true, + "/": true, + "/a": true, + "/#": true, + "a/": true, + "+/": true, + "#/": false, + "a/b": true, + "a/#": true, + "a/+": true, + "b/a": true, + "#/b": false, + "+/b": true, + "a/b/c/": true, + "/a/b/c": true, + "/a/++": true, + "a/++/b": true, + "/a/##": true, + "a/##/b": true, + "a/#/b": false, + "a/+/b": true, + "a/+/b/+": true, + "/a/+/b/#": true, + "/a/+#/b": true, + "/a/#+/b": true, + "a//b": true, + "//a": true, + "a//": true, + } + + for topic, expectError := range topics { + _, e := ValidateTopicFilter(topic) + if expectError == (e != nil) { + t.Errorf("error validating topic \"%s\"", topic) + } + } +} + +func Benchmark_ValidateTopicFilter(b *testing.B) { + topics := map[string]bool{ + "": false, + "a": true, + "#": true, + "+": true, + "/": true, + "/a": true, + "/#": true, + "a/": true, + "+/": true, + "#/": false, + "a/b": true, + "a/#": true, + "a/+": true, + "b/a": true, + "#/b": false, + "+/b": true, + "a/b/c/": false, + "/a/b/c": true, + "/a/++": true, + "a/++/b": true, + "/a/##": true, + "a/##/b": true, + "a/#/b": false, + "a/+/b": true, + "a/+/b/+": true, + "/a/+/b/#": true, + "/a/+#/b": true, + "/a/#+/b": true, + "a//b": true, + "//a": true, + "a//": true, + } + + for i := 0; i < b.N; i++ { + sum := 0 + for topic, _ := range topics { + toks, _ := ValidateTopicFilter(topic) + sum += len(toks) + } + } +} + +func Test_ValidateTopicName(t *testing.T) { + topics := map[string]bool{ + "": false, + "a": true, + "#": false, + "+": false, + "/": true, + "/a": true, + "/#": false, + "a/": true, + "+/": false, + "#/": false, + "a/b": true, + "a/#": false, + "a/+": false, + "b/a": true, + "#/b": false, + "+/b": false, + "a/b/c/": true, + "/a/b/c": true, + "/a/++": true, + "a/++/b": true, + "/a/##": true, + "a/##/b": true, + "a/#/b": false, + "a/+/b": false, + "a/+/b/+": false, + "/a/+/b/#": false, + "/a/+#/b": true, + "/a/#+/b": true, + "//": true, + "a//": true, + "//a": true, + "///a///": true, + } + + for topic, expectError := range topics { + _, e := ValidateTopicName(topic) + if expectError == (e != nil) { + t.Errorf("error validating topic \"%s\"", topic) + } + } +} + +func Benchmark_ValidateTopicName(b *testing.B) { + topics := map[string]bool{ + "": false, + "a": true, + "#": false, + "+": false, + "/": true, + "/a": true, + "/#": false, + "a/": true, + "+/": false, + "#/": false, + "a/b": true, + "a/#": false, + "a/+": false, + "b/a": true, + "#/b": false, + "+/b": false, + "a/b/c/": true, + "/a/b/c": true, + "/a/++": true, + "a/++/b": true, + "/a/##": true, + "a/##/b": true, + "a/#/b": false, + "a/+/b": false, + "a/+/b/+": false, + "/a/+/b/#": false, + "/a/+#/b": true, + "/a/#+/b": true, + "//": true, + "a//": true, + "//a": true, + "///a///": true, + } + + for i := 0; i < b.N; i++ { + sum := 0 + for topic, _ := range topics { + toks, _ := ValidateTopicName(topic) + sum += len(toks) + } + } +}
diff --git a/gateway/gate/unit_topictree_test.go b/gateway/gate/unit_topictree_test.go new file mode 100644 index 0000000..7f56c84 --- /dev/null +++ b/gateway/gate/unit_topictree_test.go
@@ -0,0 +1,648 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package gateway + +import ( + "testing" +) + +func eok(e error, t *testing.T) { + if e != nil { + t.Fatalf("ERROR %s\n", e) + } +} + +func enok(e error, t *testing.T) { + if e == nil { + t.Fatalf("ERROR (NO ERROR)") + } +} + +func chkb(b bool, e bool, t *testing.T) { + if b != e { + t.Fatalf("ERROR bool\n") + } +} + +func Test_NewTopicTree(t *testing.T) { + tt := NewTopicTree() + if tt == nil { + t.Fatalf("NewTopicTree was nil") + } + if tt.root == nil { + t.Fatalf("NewTopicTree.Root was nil") + } +} + +func Test_AddSubscription_alpha(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c1", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "alpha") + eok(e, t) + chkb(b, true, t) + if len(tt.root.clients) != 0 { + t.Fatalf("AddSub root had client") + } + if len(tt.root.children["alpha"].clients) != 1 { + t.Fatalf("AddSub alpha != 1") + } +} + +func Test_AddSubscription_Salpha(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c2", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "/alpha") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_alphaS(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c2.5", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "alpha/") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_aSbScSd(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c3", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "a/b/c/d") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_aSSb(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c4", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "a//b") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_H(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c5", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "#") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_SH(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c6", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "/#") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_aSHSb(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c7", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "a/#/b") + enok(e, t) + chkb(b, false, t) +} + +func Test_AddSubscription_aSHS(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c8", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "a/#/") + enok(e, t) + chkb(b, false, t) +} + +func Test_AddSubscription_P(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c9", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "+") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription_SPSbSP(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c10", conn, addr) + tt := NewTopicTree() + b, e := tt.AddSubscription(c, "/+/b/+") + eok(e, t) + chkb(b, true, t) +} + +func Test_AddSubscription(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("c", conn, addr) + tt := NewTopicTree() + + topics := []string{ + "a", + "a/b", + "a/b/c", + "a/b/d", + "a/b/e", + "a/b/f", + "a/b/c/d/e/f/g/h/i/j/k", + "a/b/c/d/e/f/g/h/i/j/+", + "a/b/c/d/e/f/g/h/i/j/k/+", + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z", + "a/b/c/d/e/f/g/h/i/j/k", + "+/b/c", + "a/+/b", + "a/b/+", + "a/b/#", + "#", + } + + for _, topic := range topics { + _, e := tt.AddSubscription(c, topic) + eok(e, t) + } +} + +func Benchmark_AddSubscription(b *testing.B) { + var conn uConn + var addr uAddr + c := NewClient("b", conn, addr) + tt := NewTopicTree() + + topics := []string{ + "a", + "a/b", + "a/b/c", + "a/b/d", + "a/b/e", + "a/b/f", + "a/b/c/d/e/f/g/h/i/j/k", + "a/b/c/d/e/f/g/h/i/j/+", + "a/b/c/d/e/f/g/h/i/j/k/+", + "a/b/c/d/e/f/g/h/i/j/k/l/m/n/o/p/q/r/s/t/u/v/w/x/y/z", + "a/b/c/d/e/f/g/h/i/j/k", + "+/b/c", + "a/+/b", + "a/b/+", + "a/b/#", + "#", + } + + for i := 0; i < b.N; i++ { + for _, topic := range topics { + tt.AddSubscription(c, topic) + } + } +} + +func elen(clients []*Client, e error) int { + if clients == nil { + return -1 + } + return len(clients) +} + +func Test_SubscribersOf_none(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_A", conn, addr) + tt := NewTopicTree() + + sum := 0 + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "kappa") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "/kappa") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "kappa/#") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "kappa/+") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "a/b") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "b/a") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + tt.AddSubscription(c, "/a/#") + sum += elen(tt.SubscribersOf("a")) + sum += elen(tt.SubscribersOf("/a")) + + if sum != 0 { + t.Fatalf("SubscribersOf_none had subscriber") + } +} + +func alen(exp, act, i int, t *testing.T) { + if exp != act { + t.Errorf("assert #%d expected %d, got %d", i, exp, act) + } +} + +func Test_SubscribersOf_1(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "a") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) +} + +func Test_SubscribersOf_1h(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1h", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "#") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(1, elen(tt.SubscribersOf("b")), 2, t) + alen(1, elen(tt.SubscribersOf("/a")), 3, t) + alen(1, elen(tt.SubscribersOf("/b")), 4, t) +} + +func Test_SubscribersOf_1sh(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1h", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "/#") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(1, elen(tt.SubscribersOf("/a")), 3, t) + alen(1, elen(tt.SubscribersOf("/b")), 4, t) +} + +func Test_SubscribersOf_1p(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1h", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "+") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(1, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) +} + +func Test_SubscribersOf_1sp(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1h", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "/+") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(1, elen(tt.SubscribersOf("/a")), 3, t) + alen(1, elen(tt.SubscribersOf("/b")), 4, t) +} + +func Test_SubscribersOf_1fooHS(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1fooHS", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "foo/#") + alen(1, elen(tt.SubscribersOf("foo/")), 1, t) +} + +// func Test_SubscribersOf_1fooH(t *testing.T) { +// var conn uConn +// var addr uAddr +// c := NewClient("so_1fooH", conn, addr) +// tt := NewTopicTree() + +// tt.AddSubscription(c, "foo/#") +// alen(1, elen(tt.SubscribersOf("foo")), 1, t) +// } + +func Test_SubscribersOf_2(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "a/b") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(1, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("a/b/c")), 8, t) +} + +func Test_SubscribersOf_2ash(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "a/#") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(1, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(1, elen(tt.SubscribersOf("a/a")), 7, t) + alen(1, elen(tt.SubscribersOf("a/b/c")), 8, t) +} + +func Test_SubscribersOf_2h(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "#") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(1, elen(tt.SubscribersOf("b")), 2, t) + alen(1, elen(tt.SubscribersOf("/a")), 3, t) + alen(1, elen(tt.SubscribersOf("/b")), 4, t) + alen(1, elen(tt.SubscribersOf("a/b")), 5, t) + alen(1, elen(tt.SubscribersOf("b/a")), 6, t) + alen(1, elen(tt.SubscribersOf("a/a")), 7, t) + alen(1, elen(tt.SubscribersOf("a/b/c")), 8, t) +} + +func Test_SubscribersOf_2p(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "+") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(1, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(0, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("a/b/c")), 8, t) +} + +func Test_SubscribersOf_2asp(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "a/+") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(1, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(1, elen(tt.SubscribersOf("a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("a/b/c")), 8, t) +} + +func Test_SubscribersOf_3aspsc(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "a/+/c") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(0, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("a/a")), 7, t) + alen(1, elen(tt.SubscribersOf("a/b/c")), 8, t) + alen(0, elen(tt.SubscribersOf("a/b/z")), 8, t) +} + +func Test_SubscribersOf_3saspsc(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "/a/+/c") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(0, elen(tt.SubscribersOf("a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("a/b/c")), 8, t) + alen(0, elen(tt.SubscribersOf("a/b/z")), 9, t) +} + +func Test_SubscribersOf_mix(t *testing.T) { + var conn uConn + var addr uAddr + c := NewClient("so_1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c, "/a/+/c/#") + + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(0, elen(tt.SubscribersOf("/a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("/b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("/a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("/a/b/c")), 8, t) + alen(0, elen(tt.SubscribersOf("/a/b/z")), 9, t) + alen(1, elen(tt.SubscribersOf("/a/b/c/d")), 10, t) + alen(1, elen(tt.SubscribersOf("/a/b/c/d/e")), 11, t) + alen(1, elen(tt.SubscribersOf("/a/b/c/d/e/f")), 12, t) + alen(0, elen(tt.SubscribersOf("/a/b/z/d/e/f")), 13, t) + alen(0, elen(tt.SubscribersOf("/a/b/b/c/d/e/f")), 14, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d")), 15, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d/e")), 16, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d/e/f")), 17, t) +} + +func Test_SubscribersOf_multi(t *testing.T) { + var conn uConn + var addr uAddr + c1 := NewClient("c1", conn, addr) + c2 := NewClient("c2", conn, addr) + c3 := NewClient("c3", conn, addr) + c4 := NewClient("c4", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c1, "a") + tt.AddSubscription(c2, "/a/+/c/d") + tt.AddSubscription(c3, "/a/+/c/#") + tt.AddSubscription(c4, "/a/+/c/d/+") + + alen(1, elen(tt.SubscribersOf("a")), 1, t) + alen(0, elen(tt.SubscribersOf("b")), 2, t) + alen(0, elen(tt.SubscribersOf("/a")), 3, t) + alen(0, elen(tt.SubscribersOf("/b")), 4, t) + alen(0, elen(tt.SubscribersOf("/a/b")), 5, t) + alen(0, elen(tt.SubscribersOf("/b/a")), 6, t) + alen(0, elen(tt.SubscribersOf("/a/a")), 7, t) + alen(0, elen(tt.SubscribersOf("/a/b/c")), 8, t) + alen(0, elen(tt.SubscribersOf("/a/b/z")), 9, t) + alen(2, elen(tt.SubscribersOf("/a/b/c/d")), 10, t) + alen(2, elen(tt.SubscribersOf("/a/b/c/d/e")), 11, t) + alen(1, elen(tt.SubscribersOf("/a/b/c/d/e/f")), 12, t) + alen(0, elen(tt.SubscribersOf("/a/b/z/d/e/f")), 13, t) + alen(0, elen(tt.SubscribersOf("/a/b/b/c/d/e/f")), 14, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d")), 15, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d/e")), 16, t) + alen(0, elen(tt.SubscribersOf("a/b/c/d/e/f")), 17, t) +} + +func Test_SubscribersOf_SS(t *testing.T) { + var conn uConn + var addr uAddr + c1 := NewClient("c1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c1, "/") + alen(0, elen(tt.SubscribersOf("a")), 1, t) + alen(1, elen(tt.SubscribersOf("/")), 2, t) + alen(0, elen(tt.SubscribersOf("//")), 3, t) + + tt.AddSubscription(c1, "//") + alen(0, elen(tt.SubscribersOf("a")), 4, t) + alen(1, elen(tt.SubscribersOf("/")), 5, t) + alen(1, elen(tt.SubscribersOf("//")), 6, t) +} + +func Test_RemoveSubscription_rs1(t *testing.T) { + var conn uConn + var addr uAddr + c1 := NewClient("c1", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c1, "a") + alen(1, elen(tt.SubscribersOf("a")), 1, t) + + tt.RemoveSubscription(c1, "a") + alen(0, elen(tt.SubscribersOf("a")), 2, t) +} + +func Test_RemoveSubscription_rs2(t *testing.T) { + var conn uConn + var addr uAddr + c1 := NewClient("c1", conn, addr) + c2 := NewClient("c2", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c1, "/alpha/beta") + tt.AddSubscription(c2, "/alpha/beta") + alen(2, elen(tt.SubscribersOf("/alpha/beta")), 1, t) + + tt.RemoveSubscription(c2, "/alpha/beta") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 2, t) + + tt.RemoveSubscription(c2, "/alpha/beta") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 3, t) + + tt.RemoveSubscription(c1, "/alpha/beta") + alen(0, elen(tt.SubscribersOf("/alpha/beta")), 4, t) +} + +func Test_RemoveSubscription_rs3(t *testing.T) { + var conn uConn + var addr uAddr + c1 := NewClient("c1", conn, addr) + c2 := NewClient("c2", conn, addr) + c3 := NewClient("c3", conn, addr) + tt := NewTopicTree() + + tt.AddSubscription(c1, "/alpha/+/gamma") + tt.AddSubscription(c2, "/alpha/beta") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 1, t) + alen(1, elen(tt.SubscribersOf("/alpha/beta/gamma")), 2, t) + + tt.RemoveSubscription(c1, "/alpha/+/gamma") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 3, t) + alen(0, elen(tt.SubscribersOf("/alpha/beta/gamma")), 4, t) + + tt.AddSubscription(c3, "/alpha/#") + alen(2, elen(tt.SubscribersOf("/alpha/beta")), 5, t) + alen(1, elen(tt.SubscribersOf("/alpha/foo")), 6, t) + + tt.RemoveSubscription(c2, "/alpha/beta") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 7, t) + alen(1, elen(tt.SubscribersOf("/alpha/zoo")), 8, t) + alen(1, elen(tt.SubscribersOf("/alpha/beta/gamma")), 9, t) + alen(1, elen(tt.SubscribersOf("/alpha/beta/gamma/zeta")), 10, t) + alen(0, elen(tt.SubscribersOf("/alpha")), 11, t) + + tt.RemoveSubscription(c3, "/alpha/beta") + alen(1, elen(tt.SubscribersOf("/alpha/beta")), 12, t) + alen(0, elen(tt.SubscribersOf("/alpha")), 13, t) + + tt.RemoveSubscription(c3, "/alpha/#") + alen(0, elen(tt.SubscribersOf("/alpha/beta")), 14, t) + alen(0, elen(tt.SubscribersOf("/alpha/gamma")), 15, t) + alen(0, elen(tt.SubscribersOf("/alpha/beta/gamma")), 16, t) + alen(0, elen(tt.SubscribersOf("/alpha")), 17, t) +}
diff --git a/gateway/gateway.go b/gateway/gateway.go new file mode 100644 index 0000000..fb85c06 --- /dev/null +++ b/gateway/gateway.go
@@ -0,0 +1,75 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander + * Seth Hoenig + *******************************************************************************/ + +package main + +import ( + "flag" + "os" + "os/signal" + + G "smidge/gateway/gate" +) + +func main() { + var gateway G.Gateway + stopsig := registerSignals() + gatewayconf := setup() + + G.InitLogger(os.Stdout, os.Stderr) // todo: configurable + + if gatewayconf.IsAggregating() { + G.INFO.Println("SMIDGE Gateway starting in aggregating mode") + gateway = initAggregating(gatewayconf, stopsig) + } else { + G.INFO.Println("SMIDGE Gateway starting in transparent mode") + gateway = initTransparent(gatewayconf, stopsig) + } + + gateway.Start() +} + +func setup() *G.GatewayConfig { + var configFile string + var port int + + flag.StringVar(&configFile, "c", "", "Configuration File") + flag.IntVar(&port, "port", 0, "MQTT-G UDP Listening Port") + flag.Parse() + + if configFile != "" { + if gc, err := G.ParseConfigFile(configFile); err != nil { + G.ERROR.Fatal(err) + } else { + return gc + } + } + + G.ERROR.Fatal("-configuration <file> must be specified") + return nil +} + +func initAggregating(c *G.GatewayConfig, stopsig chan os.Signal) *G.AGateway { + a := G.NewAGateway(c, stopsig) + return a +} + +func initTransparent(c *G.GatewayConfig, stopsig chan os.Signal) *G.TGateway { + t := G.NewTGateway(c, stopsig) + return t +} + +func registerSignals() chan os.Signal { + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt, os.Kill) + return c +}
diff --git a/gateway/samples/aggregating.cfg b/gateway/samples/aggregating.cfg new file mode 100644 index 0000000..fed305a --- /dev/null +++ b/gateway/samples/aggregating.cfg
@@ -0,0 +1,7 @@ +mode aggregating +port 1883 +mqtt-broker tcp://localhost:1883 +mqtt-user agateway +mqtt-password wasspord +mqtt-clientid AGGW +mqtt-timeout 300
diff --git a/gateway/samples/transparent.cfg b/gateway/samples/transparent.cfg new file mode 100644 index 0000000..24ae8b9 --- /dev/null +++ b/gateway/samples/transparent.cfg
@@ -0,0 +1,3 @@ +mode transparent +port 1883 +mqtt-broker tcp://localhost:1883
diff --git a/packets/advertise.go b/packets/advertise.go new file mode 100644 index 0000000..d973217 --- /dev/null +++ b/packets/advertise.go
@@ -0,0 +1,41 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type AdvertiseMessage struct { + Header + GatewayId byte + Duration uint16 +} + +func (a *AdvertiseMessage) MessageType() byte { + return ADVERTISE +} + +func (a *AdvertiseMessage) Write(w io.Writer) error { + packet := a.Header.pack() + packet.WriteByte(ADVERTISE) + packet.WriteByte(a.GatewayId) + packet.Write(encodeUint16(a.Duration)) + _, err := packet.WriteTo(w) + + return err +} + +func (a *AdvertiseMessage) Unpack(b io.Reader) { + a.GatewayId = readByte(b) + a.Duration = readUint16(b) +}
diff --git a/packets/advertise_test.go b/packets/advertise_test.go new file mode 100644 index 0000000..c36d62a --- /dev/null +++ b/packets/advertise_test.go
@@ -0,0 +1,31 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestAdvertiseStruct(t *testing.T) { + msg := NewMessage(ADVERTISE).(*AdvertiseMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.AdvertiseMessage", reflect.TypeOf(msg).String(), "Type should be AdvertiseMessage") + assert.Equal(t, 0, msg.GatewayId, "Default GatewayId should be 0") + assert.Equal(t, 0, msg.Duration, "Default Duration should be 0") + assert.Equal(t, 5, msg.Length, "Length should be 5") + + assert.Equal(t, ADVERTISE, msg.MessageType(), "MessageType() should return ADVERTISE") + } +}
diff --git a/packets/connack.go b/packets/connack.go new file mode 100644 index 0000000..8c27b41 --- /dev/null +++ b/packets/connack.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type ConnackMessage struct { + Header + ReturnCode byte +} + +func (c *ConnackMessage) MessageType() byte { + return CONNACK +} + +func (c *ConnackMessage) Write(w io.Writer) error { + packet := c.Header.pack() + packet.WriteByte(CONNACK) + packet.WriteByte(c.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (c *ConnackMessage) Unpack(b io.Reader) { + c.ReturnCode = readByte(b) +}
diff --git a/packets/connack_test.go b/packets/connack_test.go new file mode 100644 index 0000000..70e39a8 --- /dev/null +++ b/packets/connack_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestConnackStruct(t *testing.T) { + msg := NewMessage(CONNACK).(*ConnackMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.ConnackMessage", reflect.TypeOf(msg).String(), "Type should be ConnackMessage") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + assert.Equal(t, 3, msg.Length, "Length should be 3") + + assert.Equal(t, CONNACK, msg.MessageType(), "MessageType() should return CONNACK") + } +}
diff --git a/packets/connect.go b/packets/connect.go new file mode 100644 index 0000000..0ebdedc --- /dev/null +++ b/packets/connect.go
@@ -0,0 +1,66 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type ConnectMessage struct { + Header + Will bool + CleanSession bool + ProtocolId byte + Duration uint16 + ClientId []byte +} + +func (c *ConnectMessage) MessageType() byte { + return CONNECT +} + +func (c *ConnectMessage) decodeFlags(b byte) { + c.Will = (b & WILLFLAG) == WILLFLAG + c.CleanSession = (b & CLEANSESSION) == CLEANSESSION +} + +func (c *ConnectMessage) encodeFlags() byte { + var b byte + if c.Will { + b |= WILLFLAG + } + if c.CleanSession { + b |= CLEANSESSION + } + return b +} + +func (c *ConnectMessage) Write(w io.Writer) error { + c.Header.Length = uint16(len(c.ClientId) + 6) + packet := c.Header.pack() + packet.WriteByte(CONNECT) + packet.WriteByte(c.encodeFlags()) + packet.WriteByte(c.ProtocolId) + packet.Write(encodeUint16(c.Duration)) + packet.Write([]byte(c.ClientId)) + _, err := packet.WriteTo(w) + + return err +} + +func (c *ConnectMessage) Unpack(b io.Reader) { + c.decodeFlags(readByte(b)) + c.ProtocolId = readByte(b) + c.Duration = readUint16(b) + c.ClientId = make([]byte, c.Header.Length-6) + b.Read(c.ClientId) +}
diff --git a/packets/connect_test.go b/packets/connect_test.go new file mode 100644 index 0000000..c1187f3 --- /dev/null +++ b/packets/connect_test.go
@@ -0,0 +1,33 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestConnectMessage(t *testing.T) { + msg := NewMessage(CONNECT).(*ConnectMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.ConnectMessage", reflect.TypeOf(msg).String(), "Type should be ConnectMessage") + assert.Equal(t, false, msg.Will, "Default Will should be false") + assert.Equal(t, false, msg.CleanSession, "Default CleanSession should be false") + assert.Equal(t, 0x01, msg.ProtocolId, "Default ProtocolId should be 1") + assert.Equal(t, 0, msg.Duration, "Default Duration should be 0") + assert.Equal(t, []byte(nil), msg.ClientId, "Default ClientId should be blank") + + assert.Equal(t, CONNECT, msg.MessageType(), "MessageType() should reurn CONNECT") + } +}
diff --git a/packets/disconnect.go b/packets/disconnect.go new file mode 100644 index 0000000..050985c --- /dev/null +++ b/packets/disconnect.go
@@ -0,0 +1,50 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "bytes" + "io" +) + +type DisconnectMessage struct { + Header + Duration uint16 +} + +func (d *DisconnectMessage) MessageType() byte { + return DISCONNECT +} + +func (d *DisconnectMessage) Write(w io.Writer) error { + var packet bytes.Buffer + + if d.Duration == 0 { + d.Header.Length = 2 + packet = d.Header.pack() + packet.WriteByte(DISCONNECT) + } else { + d.Header.Length = 4 + packet = d.Header.pack() + packet.WriteByte(DISCONNECT) + packet.Write(encodeUint16(d.Duration)) + } + _, err := packet.WriteTo(w) + + return err +} + +func (d *DisconnectMessage) Unpack(b io.Reader) { + if d.Header.Length == 4 { + d.Duration = readUint16(b) + } +}
diff --git a/packets/disconnect_test.go b/packets/disconnect_test.go new file mode 100644 index 0000000..9e1d3f9 --- /dev/null +++ b/packets/disconnect_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestDisconnectStruct(t *testing.T) { + msg := NewMessage(DISCONNECT).(*DisconnectMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.DisconnectMessage", reflect.TypeOf(msg).String(), "Type should be DisconnectMessage") + assert.Equal(t, 0, msg.Duration, "Default Duration should be 0") + + assert.Equal(t, DISCONNECT, msg.MessageType(), "MessageType() should return DISCONNECT") + } +}
diff --git a/packets/gwinfo.go b/packets/gwinfo.go new file mode 100644 index 0000000..e316d06 --- /dev/null +++ b/packets/gwinfo.go
@@ -0,0 +1,44 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type GwInfoMessage struct { + Header + GatewayId byte + GatewayAddress []byte +} + +func (g *GwInfoMessage) MessageType() byte { + return GWINFO +} + +func (g *GwInfoMessage) Write(w io.Writer) error { + g.Header.Length = uint16(len(g.GatewayAddress) + 3) + packet := g.Header.pack() + packet.WriteByte(GWINFO) + packet.WriteByte(g.GatewayId) + packet.Write(g.GatewayAddress) + _, err := packet.WriteTo(w) + + return err +} + +func (g *GwInfoMessage) Unpack(b io.Reader) { + g.GatewayId = readByte(b) + if g.Header.Length > 3 { + b.Read(g.GatewayAddress) + } +}
diff --git a/packets/gwinfo_test.go b/packets/gwinfo_test.go new file mode 100644 index 0000000..fe1ebbb --- /dev/null +++ b/packets/gwinfo_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestGwInfoStruct(t *testing.T) { + msg := NewMessage(GWINFO).(*GwInfoMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.GwInfoMessage", reflect.TypeOf(msg).String(), "Type should be GwInfoMessage") + assert.Equal(t, 0, msg.GatewayId, "Default GatewayId should be 0") + assert.Equal(t, []byte(nil), msg.GatewayAddress, "Default GatewayAddress should be blank") + + assert.Equal(t, GWINFO, msg.MessageType(), "MessageType() should return GWINFO") + } +}
diff --git a/packets/packets.go b/packets/packets.go new file mode 100644 index 0000000..1ebde99 --- /dev/null +++ b/packets/packets.go
@@ -0,0 +1,296 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +/* +Package smidge/packets provides de/serialisation of MQTTSN packets +*/ + +package packets + +import ( + "bytes" + "encoding/binary" + "errors" + "io" +) + +type Message interface { + MessageType() byte + Write(io.Writer) error + Unpack(io.Reader) + //String() string + //Details() Details + //UUID() uuid.UUID +} + +type Header struct { + Length uint16 + MessageType byte +} + +func ReadPacket(r io.Reader) (m Message, err error) { + var h Header + packet := make([]byte, 1500) + r.Read(packet) + packetBuf := bytes.NewBuffer(packet) + h.unpack(packetBuf) + m = NewMessageWithHeader(h) + if m == nil { + return nil, errors.New("Bad data from client") + } + m.Unpack(packetBuf) + return m, nil +} + +func (h *Header) unpack(b io.Reader) { + lengthCheck := readByte(b) + if lengthCheck == 0x01 { + h.Length = readUint16(b) + } else { + h.Length = uint16(lengthCheck) + } + h.MessageType = readByte(b) +} + +func (h *Header) pack() bytes.Buffer { + var header bytes.Buffer + if h.Length > 256 { + h.Length += 2 + header.WriteByte(0x01) + header.Write(encodeUint16(h.Length)) + } else { + header.WriteByte(byte(h.Length)) + } + return header +} + +func NewMessage(msgType byte) (m Message) { + switch msgType { + case ADVERTISE: + m = &AdvertiseMessage{Header: Header{MessageType: ADVERTISE, Length: 5}} + case SEARCHGW: + m = &SearchGwMessage{Header: Header{MessageType: SEARCHGW, Length: 3}} + case GWINFO: + m = &GwInfoMessage{Header: Header{MessageType: GWINFO}} + case CONNECT: + m = &ConnectMessage{Header: Header{MessageType: CONNECT}, ProtocolId: 0x01} + case CONNACK: + m = &ConnackMessage{Header: Header{MessageType: CONNACK, Length: 3}} + case WILLTOPICREQ: + m = &WillTopicReqMessage{Header: Header{MessageType: WILLTOPICREQ, Length: 2}} + case WILLTOPIC: + m = &WillTopicMessage{Header: Header{MessageType: WILLTOPIC}} + case WILLMSGREQ: + m = &WillMsgReqMessage{Header: Header{MessageType: WILLMSGREQ, Length: 2}} + case WILLMSG: + m = &WillMsgMessage{Header: Header{MessageType: WILLMSG}} + case REGISTER: + m = &RegisterMessage{Header: Header{MessageType: REGISTER}} + case REGACK: + m = &RegackMessage{Header: Header{MessageType: REGACK, Length: 7}} + case PUBLISH: + m = &PublishMessage{Header: Header{MessageType: PUBLISH}} + case PUBACK: + m = &PubackMessage{Header: Header{MessageType: PUBACK, Length: 7}} + case PUBCOMP: + m = &PubcompMessage{Header: Header{MessageType: PUBCOMP, Length: 4}} + case PUBREC: + m = &PubrecMessage{Header: Header{MessageType: PUBREC, Length: 4}} + case PUBREL: + m = &PubrelMessage{Header: Header{MessageType: PUBREL, Length: 4}} + case SUBSCRIBE: + m = &SubscribeMessage{Header: Header{MessageType: SUBSCRIBE}} + case SUBACK: + m = &SubackMessage{Header: Header{MessageType: SUBACK, Length: 8}} + case UNSUBSCRIBE: + m = &UnsubscribeMessage{Header: Header{MessageType: UNSUBSCRIBE}} + case UNSUBACK: + m = &UnsubackMessage{Header: Header{MessageType: UNSUBACK, Length: 4}} + case PINGREQ: + m = &PingreqMessage{Header: Header{MessageType: PINGREQ}} + case PINGRESP: + m = &PingrespMessage{Header: Header{MessageType: PINGRESP, Length: 2}} + case DISCONNECT: + m = &DisconnectMessage{Header: Header{MessageType: DISCONNECT}} + case WILLTOPICUPD: + m = &WillTopicUpdateMessage{Header: Header{MessageType: WILLTOPICUPD}} + case WILLTOPICRESP: + m = &WillTopicRespMessage{Header: Header{MessageType: WILLTOPICRESP, Length: 3}} + case WILLMSGUPD: + m = &WillMsgUpdateMessage{Header: Header{MessageType: WILLMSGUPD}} + case WILLMSGRESP: + m = &WillMsgRespMessage{Header: Header{MessageType: WILLMSGRESP, Length: 3}} + } + return +} + +func NewMessageWithHeader(h Header) (m Message) { + switch h.MessageType { + case ADVERTISE: + m = &AdvertiseMessage{Header: h} + case SEARCHGW: + m = &SearchGwMessage{Header: h} + case GWINFO: + m = &GwInfoMessage{Header: h} + case CONNECT: + m = &ConnectMessage{Header: h} + case CONNACK: + m = &ConnackMessage{Header: h} + case WILLTOPICREQ: + m = &WillTopicReqMessage{Header: h} + case WILLTOPIC: + m = &WillTopicMessage{Header: h} + case WILLMSGREQ: + m = &WillMsgReqMessage{Header: h} + case WILLMSG: + m = &WillMsgMessage{Header: h} + case REGISTER: + m = &RegisterMessage{Header: h} + case REGACK: + m = &RegackMessage{Header: h} + case PUBLISH: + m = &PublishMessage{Header: h} + case PUBACK: + m = &PubackMessage{Header: h} + case PUBCOMP: + m = &PubcompMessage{Header: h} + case PUBREC: + m = &PubrecMessage{Header: h} + case PUBREL: + m = &PubrelMessage{Header: h} + case SUBSCRIBE: + m = &SubscribeMessage{Header: h} + case SUBACK: + m = &SubackMessage{Header: h} + case UNSUBSCRIBE: + m = &UnsubscribeMessage{Header: h} + case UNSUBACK: + m = &UnsubackMessage{Header: h} + case PINGREQ: + m = &PingreqMessage{Header: h} + case PINGRESP: + m = &PingrespMessage{Header: h} + case DISCONNECT: + m = &DisconnectMessage{Header: h} + case WILLTOPICUPD: + m = &WillTopicUpdateMessage{Header: h} + case WILLTOPICRESP: + m = &WillTopicRespMessage{Header: h} + case WILLMSGUPD: + m = &WillMsgUpdateMessage{Header: h} + case WILLMSGRESP: + m = &WillMsgRespMessage{Header: h} + } + return +} + +func readByte(b io.Reader) byte { + num := make([]byte, 1) + b.Read(num) + return num[0] +} + +func readUint16(b io.Reader) uint16 { + num := make([]byte, 2) + b.Read(num) + return binary.BigEndian.Uint16(num) +} + +func encodeUint16(num uint16) []byte { + bytes := make([]byte, 2) + binary.BigEndian.PutUint16(bytes, num) + return bytes +} + +// Flags +const ( + TOPICIDTYPE = 0x03 + CLEANSESSION = 0x04 + WILLFLAG = 0x08 + RETAINFLAG = 0x10 + QOSBITS = 0x60 + DUPFLAG = 0x80 +) + +// Errors +const ( + ACCEPTED = 0x00 + REJ_CONGESTION = 0x01 + REJ_INVALID_TID = 0x02 + REJ_NOT_SUPORTED = 0x03 +) + +// Message Types +const ( + ADVERTISE = 0x00 + SEARCHGW = 0x01 + GWINFO = 0x02 + CONNECT = 0x04 + CONNACK = 0x05 + WILLTOPICREQ = 0x06 + WILLTOPIC = 0x07 + WILLMSGREQ = 0x08 + WILLMSG = 0x09 + REGISTER = 0x0A + REGACK = 0x0B + PUBLISH = 0x0C + PUBACK = 0x0D + PUBCOMP = 0x0E + PUBREC = 0x0F + PUBREL = 0x10 + SUBSCRIBE = 0x12 + SUBACK = 0x13 + UNSUBSCRIBE = 0x14 + UNSUBACK = 0x15 + PINGREQ = 0x16 + PINGRESP = 0x17 + DISCONNECT = 0x18 + WILLTOPICUPD = 0x1A + WILLTOPICRESP = 0x1B + WILLMSGUPD = 0x1C + WILLMSGRESP = 0x1D + // 0x03 is reserved + // 0x11 is reserved + // 0x19 is reserved + // 0x1E - 0xFD is reserved + // 0xFE - Encapsulated message + // 0xFF is reserved +) + +var MessageNames = map[byte]string{ + ADVERTISE: "ADVERTISE", + SEARCHGW: "SEARCHGW", + GWINFO: "GWINFO", + CONNECT: "CONNECT", + CONNACK: "CONNACK", + WILLTOPICREQ: "WILLTOPICREQ", + WILLTOPIC: "WILLTOPIC", + WILLMSGREQ: "WILLMSGREQ", + WILLMSG: "WILLMSG", + REGISTER: "REGISTER", + REGACK: "REGACK", + PUBLISH: "PUBLISH", + PUBACK: "PUBACK", + PUBCOMP: "PUBCOMP", + PUBREC: "PUBREC", + PUBREL: "PUBREL", + SUBSCRIBE: "SUBSCRIBE", + SUBACK: "SUBACK", + UNSUBSCRIBE: "UNSUBSCRIBE", + UNSUBACK: "UNSUBACK", + PINGREQ: "PINGREQ", + PINGRESP: "PINGRESP", + DISCONNECT: "DISCONNECT", + WILLTOPICUPD: "WILLTOPICUPD", + WILLTOPICRESP: "WILLTOPICRESP", + WILLMSGUPD: "WILLMSGUPD", + WILLMSGRESP: "WILLMSGRESP", +}
diff --git a/packets/packets_test.go b/packets/packets_test.go new file mode 100644 index 0000000..597185d --- /dev/null +++ b/packets/packets_test.go
@@ -0,0 +1,47 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "testing" +) + +func TestMessageTypeValues(t *testing.T) { + assert.Equal(t, 0x00, ADVERTISE, "ADVERTISE should be 0x00") + assert.Equal(t, 0x01, SEARCHGW, "SEARCHGW should be 0x01") + assert.Equal(t, 0x02, GWINFO, "GWINFO should be 0x02") + assert.Equal(t, 0x04, CONNECT, "CONNECT should be 0x04") + assert.Equal(t, 0x05, CONNACK, "CONNACK should be 0x05") + assert.Equal(t, 0x06, WILLTOPICREQ, "WILLTOPICREQ should be 0x06") + assert.Equal(t, 0x07, WILLTOPIC, "WILLTOPIC should be 0x07") + assert.Equal(t, 0x08, WILLMSGREQ, "WILLMSGREQ should be 0x08") + assert.Equal(t, 0x09, WILLMSG, "WILLMSG should be 0x09") + assert.Equal(t, 0x0A, REGISTER, "REGISTER should be 0x0A") + assert.Equal(t, 0x0B, REGACK, "REGACK should be 0x0B") + assert.Equal(t, 0x0C, PUBLISH, "PUBLISH should be 0x0C") + assert.Equal(t, 0x0D, PUBACK, "PUBACK should be 0x0D") + assert.Equal(t, 0x0E, PUBCOMP, "PUBCOMP should be 0x0E") + assert.Equal(t, 0x0F, PUBREC, "PUBREC should be 0x0F") + assert.Equal(t, 0x10, PUBREL, "PUBREL should be 0x10") + assert.Equal(t, 0x12, SUBSCRIBE, "SUBSCRIBE should be 0x12") + assert.Equal(t, 0x13, SUBACK, "SUBACK should be 0x13") + assert.Equal(t, 0x14, UNSUBSCRIBE, "UNSUBSCRIBE should be 0x14") + assert.Equal(t, 0x15, UNSUBACK, "UNSUBACK should be 0x15") + assert.Equal(t, 0x16, PINGREQ, "PINGREQ should be 0x16") + assert.Equal(t, 0x17, PINGRESP, "PINGRESPshould be 0x17") + assert.Equal(t, 0x18, DISCONNECT, "DISCONNECTshould be 0x18") + assert.Equal(t, 0x1A, WILLTOPICUPD, "WILLTOPICUPDshould be 0x1A") + assert.Equal(t, 0x1B, WILLTOPICRESP, "WILLTOPICRESshould be 0x1B") + assert.Equal(t, 0x1C, WILLMSGUPD, "WILLMSGUPDshould be 0x1C") + assert.Equal(t, 0x1D, WILLMSGRESP, "WILLMSGRESPshould be 0x1D") +}
diff --git a/packets/pingreq.go b/packets/pingreq.go new file mode 100644 index 0000000..6461292 --- /dev/null +++ b/packets/pingreq.go
@@ -0,0 +1,43 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PingreqMessage struct { + Header + ClientId []byte +} + +func (p *PingreqMessage) MessageType() byte { + return PINGREQ +} + +func (p *PingreqMessage) Write(w io.Writer) error { + p.Header.Length = uint16(len(p.ClientId) + 2) + packet := p.Header.pack() + packet.WriteByte(PINGREQ) + if len(p.ClientId) > 0 { + packet.Write(p.ClientId) + } + _, err := packet.WriteTo(w) + + return err +} + +func (p *PingreqMessage) Unpack(b io.Reader) { + if p.Header.Length > 2 { + b.Read(p.ClientId) + } +}
diff --git a/packets/pingreq_test.go b/packets/pingreq_test.go new file mode 100644 index 0000000..0a9b13d --- /dev/null +++ b/packets/pingreq_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPingreqStruct(t *testing.T) { + msg := NewMessage(PINGREQ).(*PingreqMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PingreqMessage", reflect.TypeOf(msg).String(), "Type should be PingreqMessage") + assert.Equal(t, []byte(nil), msg.ClientId, "Default ClientId should be blank") + + assert.Equal(t, PINGREQ, msg.MessageType(), "MessageType() should return PINGREQ") + } +}
diff --git a/packets/pingresp.go b/packets/pingresp.go new file mode 100644 index 0000000..0c32039 --- /dev/null +++ b/packets/pingresp.go
@@ -0,0 +1,35 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PingrespMessage struct { + Header +} + +func (p *PingrespMessage) MessageType() byte { + return PINGRESP +} + +func (p *PingrespMessage) Write(w io.Writer) error { + packet := p.Header.pack() + packet.WriteByte(PINGRESP) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PingrespMessage) Unpack(b io.Reader) { +}
diff --git a/packets/pingresp_test.go b/packets/pingresp_test.go new file mode 100644 index 0000000..644f4b0 --- /dev/null +++ b/packets/pingresp_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPingrespStruct(t *testing.T) { + msg := NewMessage(PINGRESP).(*PingrespMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PingrespMessage", reflect.TypeOf(msg).String(), "Type should be PingrespMessage") + assert.Equal(t, 2, msg.Length, "Default Length should be 2") + + assert.Equal(t, PINGRESP, msg.MessageType(), "MessageType() should return PINGRESP") + } +}
diff --git a/packets/puback.go b/packets/puback.go new file mode 100644 index 0000000..282e322 --- /dev/null +++ b/packets/puback.go
@@ -0,0 +1,44 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PubackMessage struct { + Header + TopicId uint16 + MessageId uint16 + ReturnCode byte +} + +func (p *PubackMessage) MessageType() byte { + return PUBACK +} + +func (p *PubackMessage) Write(w io.Writer) error { + packet := p.Header.pack() + packet.WriteByte(PUBACK) + packet.Write(encodeUint16(p.TopicId)) + packet.Write(encodeUint16(p.MessageId)) + packet.WriteByte(p.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PubackMessage) Unpack(b io.Reader) { + p.TopicId = readUint16(b) + p.MessageId = readUint16(b) + p.ReturnCode = readByte(b) +}
diff --git a/packets/puback_test.go b/packets/puback_test.go new file mode 100644 index 0000000..7e4b101 --- /dev/null +++ b/packets/puback_test.go
@@ -0,0 +1,32 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPubackStruct(t *testing.T) { + msg := NewMessage(PUBACK).(*PubackMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PubackMessage", reflect.TypeOf(msg).String(), "Type should be PubackMessage") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + assert.Equal(t, 7, msg.Length, "Default Length should be 2") + + assert.Equal(t, PUBACK, msg.MessageType(), "MessageType() should return PUBACK") + } +}
diff --git a/packets/pubcomp.go b/packets/pubcomp.go new file mode 100644 index 0000000..1a4dae7 --- /dev/null +++ b/packets/pubcomp.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PubcompMessage struct { + Header + MessageId uint16 +} + +func (p *PubcompMessage) MessageType() byte { + return PUBCOMP +} + +func (p *PubcompMessage) Write(w io.Writer) error { + packet := p.Header.pack() + packet.WriteByte(PUBCOMP) + packet.Write(encodeUint16(p.MessageId)) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PubcompMessage) Unpack(b io.Reader) { + p.MessageId = readUint16(b) +}
diff --git a/packets/pubcomp_test.go b/packets/pubcomp_test.go new file mode 100644 index 0000000..a164eb2 --- /dev/null +++ b/packets/pubcomp_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPubcompStruct(t *testing.T) { + msg := NewMessage(PUBCOMP).(*PubcompMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PubcompMessage", reflect.TypeOf(msg).String(), "Type should be PubcompMessage") + assert.Equal(t, 4, msg.Length, "Default Length should be 4") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + + assert.Equal(t, PUBCOMP, msg.MessageType(), "MessageType() should return PUBCOMP") + } +}
diff --git a/packets/publish.go b/packets/publish.go new file mode 100644 index 0000000..0f66072 --- /dev/null +++ b/packets/publish.go
@@ -0,0 +1,84 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PublishMessage struct { + Header + Dup bool + Retain bool + Qos byte + TopicIdType byte + TopicId uint16 + MessageId uint16 + Data []byte +} + +func NewPublishMessage(TopicId uint16, TopicIdType byte, Data []byte, Qos byte, MessageId uint16, Retain bool, Dup bool) *PublishMessage { + return &PublishMessage{ + TopicId: TopicId, + TopicIdType: TopicIdType, + Data: Data, + Qos: Qos, + MessageId: MessageId, + Retain: Retain, + Dup: Dup, + } +} + +func (p *PublishMessage) MessageType() byte { + return PUBLISH +} + +func (p *PublishMessage) encodeFlags() byte { + var b byte + if p.Dup { + b |= DUPFLAG + } + b |= (p.Qos << 5) & QOSBITS + if p.Retain { + b |= RETAINFLAG + } + b |= p.TopicIdType & TOPICIDTYPE + return b +} + +func (p *PublishMessage) decodeFlags(b byte) { + p.Dup = (b & DUPFLAG) == DUPFLAG + p.Qos = (b & QOSBITS) >> 5 + p.Retain = (b & RETAINFLAG) == RETAINFLAG + p.TopicIdType = b & TOPICIDTYPE +} + +func (p *PublishMessage) Write(w io.Writer) error { + p.Header.Length = uint16(len(p.Data) + 7) + packet := p.Header.pack() + packet.WriteByte(PUBLISH) + packet.WriteByte(p.encodeFlags()) + packet.Write(encodeUint16(p.TopicId)) + packet.Write(encodeUint16(p.MessageId)) + packet.Write(p.Data) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PublishMessage) Unpack(b io.Reader) { + p.decodeFlags(readByte(b)) + p.TopicId = readUint16(b) + p.MessageId = readUint16(b) + p.Data = make([]byte, p.Header.Length-7) + b.Read(p.Data) +}
diff --git a/packets/publish_test.go b/packets/publish_test.go new file mode 100644 index 0000000..c90db29 --- /dev/null +++ b/packets/publish_test.go
@@ -0,0 +1,35 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPublishStruct(t *testing.T) { + msg := NewMessage(PUBLISH).(*PublishMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PublishMessage", reflect.TypeOf(msg).String(), "Type should be PublishMessage") + assert.Equal(t, false, msg.Dup, "Default Dup flag should be false") + assert.Equal(t, false, msg.Retain, "Default Retain flag should be false") + assert.Equal(t, 0, msg.Qos, "Default Qos should be 0") + assert.Equal(t, 0, msg.TopicIdType, "Default TopicIdType should be 0") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, []byte(nil), msg.Data, "Default Data should be blank") + + assert.Equal(t, PUBLISH, msg.MessageType(), "MessageType() should return PUBLISH") + } +}
diff --git a/packets/pubrec.go b/packets/pubrec.go new file mode 100644 index 0000000..eaf40d9 --- /dev/null +++ b/packets/pubrec.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PubrecMessage struct { + Header + MessageId uint16 +} + +func (p *PubrecMessage) MessageType() byte { + return PUBREC +} + +func (p *PubrecMessage) Write(w io.Writer) error { + packet := p.Header.pack() + packet.WriteByte(PUBREC) + packet.Write(encodeUint16(p.MessageId)) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PubrecMessage) Unpack(b io.Reader) { + p.MessageId = readUint16(b) +}
diff --git a/packets/pubrec_test.go b/packets/pubrec_test.go new file mode 100644 index 0000000..d81805d --- /dev/null +++ b/packets/pubrec_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPubrecStruct(t *testing.T) { + msg := NewMessage(PUBREC).(*PubrecMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PubrecMessage", reflect.TypeOf(msg).String(), "Type should be PubrecMessage") + assert.Equal(t, 4, msg.Length, "Default Length should be 4") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + + assert.Equal(t, PUBREC, msg.MessageType(), "MessageType() should return PUBREC") + } +}
diff --git a/packets/pubrel.go b/packets/pubrel.go new file mode 100644 index 0000000..9b93191 --- /dev/null +++ b/packets/pubrel.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type PubrelMessage struct { + Header + MessageId uint16 +} + +func (p *PubrelMessage) MessageType() byte { + return PUBREL +} + +func (p *PubrelMessage) Write(w io.Writer) error { + packet := p.Header.pack() + packet.WriteByte(PUBREL) + packet.Write(encodeUint16(p.MessageId)) + _, err := packet.WriteTo(w) + + return err +} + +func (p *PubrelMessage) Unpack(b io.Reader) { + p.MessageId = readUint16(b) +}
diff --git a/packets/pubrel_test.go b/packets/pubrel_test.go new file mode 100644 index 0000000..c72b43b --- /dev/null +++ b/packets/pubrel_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestPubrelStruct(t *testing.T) { + msg := NewMessage(PUBREL).(*PubrelMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.PubrelMessage", reflect.TypeOf(msg).String(), "Type should be PubrelMessage") + assert.Equal(t, 4, msg.Length, "Default Length should be 4") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + + assert.Equal(t, PUBREL, msg.MessageType(), "MessageType() should return PUBREL") + } +}
diff --git a/packets/regack.go b/packets/regack.go new file mode 100644 index 0000000..d22671a --- /dev/null +++ b/packets/regack.go
@@ -0,0 +1,52 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type RegackMessage struct { + Header + TopicId uint16 + MessageId uint16 + ReturnCode byte +} + +func NewRegackMessage(TopicId uint16, MessageId uint16, rc byte) *RegackMessage { + return &RegackMessage{ + TopicId: TopicId, + MessageId: MessageId, + ReturnCode: rc, + } +} + +func (r *RegackMessage) MessageType() byte { + return REGACK +} + +func (r *RegackMessage) Write(w io.Writer) error { + packet := r.Header.pack() + packet.WriteByte(REGACK) + packet.Write(encodeUint16(r.TopicId)) + packet.Write(encodeUint16(r.MessageId)) + packet.WriteByte(r.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (r *RegackMessage) Unpack(b io.Reader) { + r.TopicId = readUint16(b) + r.MessageId = readUint16(b) + r.ReturnCode = readByte(b) +}
diff --git a/packets/regack_test.go b/packets/regack_test.go new file mode 100644 index 0000000..6163154 --- /dev/null +++ b/packets/regack_test.go
@@ -0,0 +1,32 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestRegackStruct(t *testing.T) { + msg := NewMessage(REGACK).(*RegackMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.RegackMessage", reflect.TypeOf(msg).String(), "Type should be RegackMessage") + assert.Equal(t, 7, msg.Length, "Default Length should be 7") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + + assert.Equal(t, REGACK, msg.MessageType(), "MessageType() should return REGACK") + } +}
diff --git a/packets/register.go b/packets/register.go new file mode 100644 index 0000000..2e87e8d --- /dev/null +++ b/packets/register.go
@@ -0,0 +1,54 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type RegisterMessage struct { + Header + TopicId uint16 + MessageId uint16 + TopicName []byte +} + +func NewRegisterMessage(TopicId, MessageId uint16, TopicName []byte) *RegisterMessage { + return &RegisterMessage{ + TopicId: TopicId, + MessageId: MessageId, + TopicName: TopicName, + } +} + +func (r *RegisterMessage) MessageType() byte { + return REGISTER +} + +func (r *RegisterMessage) Write(w io.Writer) error { + r.Header.Length = uint16(len(r.TopicName) + 6) + packet := r.Header.pack() + packet.WriteByte(REGISTER) + packet.Write(encodeUint16(r.TopicId)) + packet.Write(encodeUint16(r.MessageId)) + packet.Write(r.TopicName) + _, err := packet.WriteTo(w) + + return err +} + +func (r *RegisterMessage) Unpack(b io.Reader) { + r.TopicId = readUint16(b) + r.MessageId = readUint16(b) + r.TopicName = make([]byte, r.Header.Length-6) + b.Read(r.TopicName) +}
diff --git a/packets/register_test.go b/packets/register_test.go new file mode 100644 index 0000000..2cc10cd --- /dev/null +++ b/packets/register_test.go
@@ -0,0 +1,31 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestRegisterStruct(t *testing.T) { + msg := NewMessage(REGISTER).(*RegisterMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.RegisterMessage", reflect.TypeOf(msg).String(), "Type should be RegisterMessage") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, []byte(nil), msg.TopicName, "Default TopicName should be blank") + + assert.Equal(t, REGISTER, msg.MessageType(), "MessageType() should return REGISTER") + } +}
diff --git a/packets/searchgw.go b/packets/searchgw.go new file mode 100644 index 0000000..019d73a --- /dev/null +++ b/packets/searchgw.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type SearchGwMessage struct { + Header + Radius byte +} + +func (s *SearchGwMessage) MessageType() byte { + return SEARCHGW +} + +func (s *SearchGwMessage) Write(w io.Writer) error { + packet := s.Header.pack() + packet.WriteByte(SEARCHGW) + packet.WriteByte(s.Radius) + _, err := packet.WriteTo(w) + + return err +} + +func (s *SearchGwMessage) Unpack(b io.Reader) { + s.Radius = readByte(b) +}
diff --git a/packets/searchgw_test.go b/packets/searchgw_test.go new file mode 100644 index 0000000..b46224b --- /dev/null +++ b/packets/searchgw_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestSearchGwStruct(t *testing.T) { + msg := NewMessage(SEARCHGW).(*SearchGwMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.SearchGwMessage", reflect.TypeOf(msg).String(), "Type should be SearchGwMessage") + assert.Equal(t, 3, msg.Length, "Default Length should be 3") + assert.Equal(t, 0, msg.Radius, "Default Radius should be 0") + + assert.Equal(t, SEARCHGW, msg.MessageType(), "MessageType() should return SEARCHGW") + } +}
diff --git a/packets/suback.go b/packets/suback.go new file mode 100644 index 0000000..56270e1 --- /dev/null +++ b/packets/suback.go
@@ -0,0 +1,66 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type SubackMessage struct { + Header + Qos byte + ReturnCode byte + TopicId uint16 + MessageId uint16 +} + +func NewSubackMessage(TopicId uint16, MessageId uint16, Qos byte, rc byte) *SubackMessage { + return &SubackMessage{ + Qos: Qos, + ReturnCode: rc, + TopicId: TopicId, + MessageId: MessageId, + } +} + +func (s *SubackMessage) MessageType() byte { + return SUBACK +} + +func (s *SubackMessage) encodeFlags() byte { + var b byte + b |= (s.Qos << 5) & QOSBITS + return b +} + +func (s *SubackMessage) decodeFlags(b byte) { + s.Qos = (b & QOSBITS) >> 5 +} + +func (s *SubackMessage) Write(w io.Writer) error { + packet := s.Header.pack() + packet.WriteByte(SUBACK) + packet.WriteByte(s.encodeFlags()) + packet.Write(encodeUint16(s.TopicId)) + packet.Write(encodeUint16(s.MessageId)) + packet.WriteByte(s.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (s *SubackMessage) Unpack(b io.Reader) { + s.decodeFlags(readByte(b)) + s.TopicId = readUint16(b) + s.MessageId = readUint16(b) + s.ReturnCode = readByte(b) +}
diff --git a/packets/suback_test.go b/packets/suback_test.go new file mode 100644 index 0000000..32a5424 --- /dev/null +++ b/packets/suback_test.go
@@ -0,0 +1,33 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestSubackStruct(t *testing.T) { + msg := NewMessage(SUBACK).(*SubackMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.SubackMessage", reflect.TypeOf(msg).String(), "Type should be SubackMessage") + assert.Equal(t, 8, msg.Length, "Default Length should be 8") + assert.Equal(t, 0, msg.Qos, "Default Qos should be 0") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + + assert.Equal(t, SUBACK, msg.MessageType(), "MessageType() should return SUBACK") + } +}
diff --git a/packets/subscribe.go b/packets/subscribe.go new file mode 100644 index 0000000..066ae8e --- /dev/null +++ b/packets/subscribe.go
@@ -0,0 +1,80 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type SubscribeMessage struct { + Header + Dup bool + Qos byte + TopicIdType byte + MessageId uint16 + TopicId uint16 + TopicName []byte +} + +func (s *SubscribeMessage) MessageType() byte { + return SUBSCRIBE +} + +func (s *SubscribeMessage) encodeFlags() byte { + var b byte + if s.Dup { + b |= DUPFLAG + } + b |= (s.Qos << 5) & QOSBITS + b |= s.TopicIdType & TOPICIDTYPE + return b +} + +func (s *SubscribeMessage) decodeFlags(b byte) { + s.Dup = (b & DUPFLAG) == DUPFLAG + s.Qos = (b & QOSBITS) >> 5 + s.TopicIdType = b & TOPICIDTYPE +} + +func (s *SubscribeMessage) Write(w io.Writer) error { + switch s.TopicIdType { + case 0x00, 0x02: + s.Header.Length = uint16(len(s.TopicName) + 5) + case 0x01: + s.Header.Length = 7 + } + packet := s.Header.pack() + packet.WriteByte(SUBSCRIBE) + packet.WriteByte(s.encodeFlags()) + packet.Write(encodeUint16(s.MessageId)) + switch s.TopicIdType { + case 0x00, 0x02: + packet.Write(s.TopicName) + case 0x01: + packet.Write(encodeUint16(s.TopicId)) + } + _, err := packet.WriteTo(w) + + return err +} + +func (s *SubscribeMessage) Unpack(b io.Reader) { + s.decodeFlags(readByte(b)) + s.MessageId = readUint16(b) + switch s.TopicIdType { + case 0x00, 0x02: + s.TopicName = make([]byte, s.Header.Length-5) + b.Read(s.TopicName) + case 0x01: + s.TopicId = readUint16(b) + } +}
diff --git a/packets/subscribe_test.go b/packets/subscribe_test.go new file mode 100644 index 0000000..576f82d --- /dev/null +++ b/packets/subscribe_test.go
@@ -0,0 +1,34 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestSubscribeStruct(t *testing.T) { + msg := NewMessage(SUBSCRIBE).(*SubscribeMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.SubscribeMessage", reflect.TypeOf(msg).String(), "Type should be SubscribeMessage") + assert.Equal(t, false, msg.Dup, "Default Dup flag should be false") + assert.Equal(t, 0, msg.Qos, "Default Qos should be 0") + assert.Equal(t, 0, msg.TopicIdType, "Default TopicIdType should be 0") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, []byte(nil), msg.TopicName, "Default Topicname should be blank") + + assert.Equal(t, SUBSCRIBE, msg.MessageType(), "MessageType() should return SUBSCRIBE") + } +}
diff --git a/packets/unsuback.go b/packets/unsuback.go new file mode 100644 index 0000000..e0e7dba --- /dev/null +++ b/packets/unsuback.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type UnsubackMessage struct { + Header + MessageId uint16 +} + +func (u *UnsubackMessage) MessageType() byte { + return UNSUBACK +} + +func (u *UnsubackMessage) Write(w io.Writer) error { + packet := u.Header.pack() + packet.WriteByte(UNSUBACK) + packet.Write(encodeUint16(u.MessageId)) + _, err := packet.WriteTo(w) + + return err +} + +func (u *UnsubackMessage) Unpack(b io.Reader) { + u.MessageId = readUint16(b) +}
diff --git a/packets/unsuback_test.go b/packets/unsuback_test.go new file mode 100644 index 0000000..7f3e08b --- /dev/null +++ b/packets/unsuback_test.go
@@ -0,0 +1,30 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestUnsubackStruct(t *testing.T) { + msg := NewMessage(UNSUBACK).(*UnsubackMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.UnsubackMessage", reflect.TypeOf(msg).String(), "Type should be UnsubackMessage") + assert.Equal(t, 4, msg.Length, "Default Length should be 4") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + + assert.Equal(t, UNSUBACK, msg.MessageType(), "MessageType() should return UNSUBACK") + } +}
diff --git a/packets/unsubscribe.go b/packets/unsubscribe.go new file mode 100644 index 0000000..8ed0511 --- /dev/null +++ b/packets/unsubscribe.go
@@ -0,0 +1,71 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type UnsubscribeMessage struct { + Header + TopicIdType byte + MessageId uint16 + TopicId uint16 + TopicName []byte +} + +func (u *UnsubscribeMessage) MessageType() byte { + return UNSUBSCRIBE +} + +func (s *UnsubscribeMessage) encodeFlags() byte { + var b byte + b |= s.TopicIdType & TOPICIDTYPE + return b +} + +func (s *UnsubscribeMessage) decodeFlags(b byte) { + s.TopicIdType = b & TOPICIDTYPE +} + +func (u *UnsubscribeMessage) Write(w io.Writer) error { + switch u.TopicIdType { + case 0x00, 0x02: + u.Header.Length = uint16(len(u.TopicName) + 5) + case 0x01: + u.Header.Length = 7 + } + packet := u.Header.pack() + packet.WriteByte(UNSUBSCRIBE) + packet.WriteByte(u.encodeFlags()) + packet.Write(encodeUint16(u.MessageId)) + switch u.TopicIdType { + case 0x00, 0x02: + packet.Write(u.TopicName) + case 0x01: + packet.Write(encodeUint16(u.TopicId)) + } + _, err := packet.WriteTo(w) + + return err +} + +func (u *UnsubscribeMessage) Unpack(b io.Reader) { + u.decodeFlags(readByte(b)) + u.MessageId = readUint16(b) + switch u.TopicIdType { + case 0x00, 0x02: + b.Read(u.TopicName) + case 0x01: + u.TopicId = readUint16(b) + } +}
diff --git a/packets/unsubscribe_test.go b/packets/unsubscribe_test.go new file mode 100644 index 0000000..cc2cff9 --- /dev/null +++ b/packets/unsubscribe_test.go
@@ -0,0 +1,32 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestUnsubscribeStruct(t *testing.T) { + msg := NewMessage(UNSUBSCRIBE).(*UnsubscribeMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.UnsubscribeMessage", reflect.TypeOf(msg).String(), "Type should be UnsubscribeMessage") + assert.Equal(t, 0, msg.TopicIdType, "Default TopicIdType should be 0") + assert.Equal(t, 0, msg.TopicId, "Default TopicId should be 0") + assert.Equal(t, 0, msg.MessageId, "Default MessageId should be 0") + assert.Equal(t, []byte(nil), msg.TopicName, "Default TopicName should be blank") + + assert.Equal(t, UNSUBSCRIBE, msg.MessageType(), "MessageType() should return UNSUBSCRIBE") + } +}
diff --git a/packets/willmsg.go b/packets/willmsg.go new file mode 100644 index 0000000..45ab87d --- /dev/null +++ b/packets/willmsg.go
@@ -0,0 +1,39 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillMsgMessage struct { + Header + WillMsg []byte +} + +func (wm *WillMsgMessage) MessageType() byte { + return WILLMSG +} + +func (wm *WillMsgMessage) Write(w io.Writer) error { + wm.Header.Length = uint16(len(wm.WillMsg) + 2) + packet := wm.Header.pack() + packet.WriteByte(WILLMSG) + packet.Write(wm.WillMsg) + _, err := packet.WriteTo(w) + + return err +} + +func (wm *WillMsgMessage) Unpack(b io.Reader) { + b.Read(wm.WillMsg) +}
diff --git a/packets/willmsg_test.go b/packets/willmsg_test.go new file mode 100644 index 0000000..7d69906 --- /dev/null +++ b/packets/willmsg_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillMsgStruct(t *testing.T) { + msg := NewMessage(WILLMSG).(*WillMsgMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillMsgMessage", reflect.TypeOf(msg).String(), "Type should be WillMsgMessage") + assert.Equal(t, []byte(nil), msg.WillMsg, "Default WillMsg should be blank") + + assert.Equal(t, WILLMSG, msg.MessageType(), "MessageType() should return WILLMSG") + } +}
diff --git a/packets/willmsgreq.go b/packets/willmsgreq.go new file mode 100644 index 0000000..6e96a4d --- /dev/null +++ b/packets/willmsgreq.go
@@ -0,0 +1,36 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillMsgReqMessage struct { + Header +} + +func (wm *WillMsgReqMessage) MessageType() byte { + return WILLMSGREQ +} + +func (wm *WillMsgReqMessage) Write(w io.Writer) error { + packet := wm.Header.pack() + packet.WriteByte(wm.Header.MessageType) + _, err := packet.WriteTo(w) + + return err +} + +func (wm *WillMsgReqMessage) Unpack(b io.Reader) { + +}
diff --git a/packets/willmsgreq_test.go b/packets/willmsgreq_test.go new file mode 100644 index 0000000..fdc339d --- /dev/null +++ b/packets/willmsgreq_test.go
@@ -0,0 +1,28 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillMsgReqStruct(t *testing.T) { + msg := NewMessage(WILLMSGREQ).(*WillMsgReqMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillMsgReqMessage", reflect.TypeOf(msg).String(), "Type should be WillMsgReqMessage") + + assert.Equal(t, WILLMSGREQ, msg.MessageType(), "MessageType() should return WILLMSGREQ") + } +}
diff --git a/packets/willmsgresp.go b/packets/willmsgresp.go new file mode 100644 index 0000000..0f1e04b --- /dev/null +++ b/packets/willmsgresp.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillMsgRespMessage struct { + Header + ReturnCode byte +} + +func (wm *WillMsgRespMessage) MessageType() byte { + return WILLMSGRESP +} + +func (wm *WillMsgRespMessage) Write(w io.Writer) error { + packet := wm.Header.pack() + packet.WriteByte(WILLMSGRESP) + packet.WriteByte(wm.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (wm *WillMsgRespMessage) Unpack(b io.Reader) { + wm.ReturnCode = readByte(b) +}
diff --git a/packets/willmsgresp_test.go b/packets/willmsgresp_test.go new file mode 100644 index 0000000..1597fe0 --- /dev/null +++ b/packets/willmsgresp_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillMsgRespStruct(t *testing.T) { + msg := NewMessage(WILLMSGRESP).(*WillMsgRespMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillMsgRespMessage", reflect.TypeOf(msg).String(), "Type should be WillMsgRespMessage") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + + assert.Equal(t, WILLMSGRESP, msg.MessageType(), "MessageType() should return WILLMSGRESP") + } +}
diff --git a/packets/willmsgupd.go b/packets/willmsgupd.go new file mode 100644 index 0000000..e255c4f --- /dev/null +++ b/packets/willmsgupd.go
@@ -0,0 +1,39 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillMsgUpdateMessage struct { + Header + WillMsg []byte +} + +func (wm *WillMsgUpdateMessage) MessageType() byte { + return WILLMSGUPD +} + +func (wm *WillMsgUpdateMessage) Write(w io.Writer) error { + wm.Header.Length = uint16(len(wm.WillMsg) + 2) + packet := wm.Header.pack() + packet.WriteByte(WILLMSGUPD) + packet.Write(wm.WillMsg) + _, err := packet.WriteTo(w) + + return err +} + +func (wm *WillMsgUpdateMessage) Unpack(b io.Reader) { + b.Read(wm.WillMsg) +}
diff --git a/packets/willmsgupd_test.go b/packets/willmsgupd_test.go new file mode 100644 index 0000000..a706c11 --- /dev/null +++ b/packets/willmsgupd_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillMsgUpdateStruct(t *testing.T) { + msg := NewMessage(WILLMSGUPD).(*WillMsgUpdateMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillMsgUpdateMessage", reflect.TypeOf(msg).String(), "Type should be WillMsgUpdateMessage") + assert.Equal(t, []byte(nil), msg.WillMsg, "Default WillMsg should be blank") + + assert.Equal(t, WILLMSGUPD, msg.MessageType(), "MessageType() should return WILLMSGUPD") + } +}
diff --git a/packets/willtopic.go b/packets/willtopic.go new file mode 100644 index 0000000..9de8e98 --- /dev/null +++ b/packets/willtopic.go
@@ -0,0 +1,66 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillTopicMessage struct { + Header + Qos byte + Retain bool + WillTopic []byte +} + +func (wt *WillTopicMessage) MessageType() byte { + return wt.Header.MessageType +} + +func (wt *WillTopicMessage) encodeFlags() byte { + var b byte + + b |= (wt.Qos << 5) & QOSBITS + if wt.Retain { + b |= RETAINFLAG + } + return b +} + +func (wt *WillTopicMessage) decodeFlags(b byte) { + wt.Qos = (b & QOSBITS) >> 5 + wt.Retain = (b & RETAINFLAG) == RETAINFLAG +} + +func (wt *WillTopicMessage) Write(w io.Writer) error { + if len(wt.WillTopic) == 0 { + wt.Header.Length = 2 + } else { + wt.Header.Length = uint16(len(wt.WillTopic) + 3) + } + packet := wt.Header.pack() + packet.WriteByte(wt.Header.MessageType) + if wt.Header.Length > 2 { + packet.WriteByte(wt.encodeFlags()) + packet.Write(wt.WillTopic) + } + _, err := packet.WriteTo(w) + + return err +} + +func (wt *WillTopicMessage) Unpack(b io.Reader) { + if wt.Header.Length > 2 { + wt.decodeFlags(readByte(b)) + b.Read(wt.WillTopic) + } +}
diff --git a/packets/willtopic_test.go b/packets/willtopic_test.go new file mode 100644 index 0000000..0889212 --- /dev/null +++ b/packets/willtopic_test.go
@@ -0,0 +1,31 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillTopicStruct(t *testing.T) { + msg := NewMessage(WILLTOPIC).(*WillTopicMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillTopicMessage", reflect.TypeOf(msg).String(), "Type should be WillTopicMessage") + assert.Equal(t, 0, msg.Qos, "Default Qos should be 0") + assert.Equal(t, false, msg.Retain, "Default Retain flag should be false") + assert.Equal(t, []byte(nil), msg.WillTopic, "Default WillMsg should be blank") + + assert.Equal(t, WILLTOPIC, msg.MessageType(), "MessageType() should return WILLTOPIC") + } +}
diff --git a/packets/willtopicreq.go b/packets/willtopicreq.go new file mode 100644 index 0000000..e0aeb9f --- /dev/null +++ b/packets/willtopicreq.go
@@ -0,0 +1,36 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillTopicReqMessage struct { + Header +} + +func (wt *WillTopicReqMessage) MessageType() byte { + return WILLTOPICREQ +} + +func (wt *WillTopicReqMessage) Write(w io.Writer) error { + packet := wt.Header.pack() + packet.WriteByte(WILLTOPICREQ) + _, err := packet.WriteTo(w) + + return err +} + +func (wt *WillTopicReqMessage) Unpack(b io.Reader) { + +}
diff --git a/packets/willtopicreq_test.go b/packets/willtopicreq_test.go new file mode 100644 index 0000000..553d5e2 --- /dev/null +++ b/packets/willtopicreq_test.go
@@ -0,0 +1,28 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillTopicReqStruct(t *testing.T) { + msg := NewMessage(WILLTOPICREQ).(*WillTopicReqMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillTopicReqMessage", reflect.TypeOf(msg).String(), "Type should be WillTopicReqMessage") + + assert.Equal(t, WILLTOPICREQ, msg.MessageType(), "MessageType() should return WILLTOPICREQ") + } +}
diff --git a/packets/willtopicresp.go b/packets/willtopicresp.go new file mode 100644 index 0000000..b008a97 --- /dev/null +++ b/packets/willtopicresp.go
@@ -0,0 +1,38 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillTopicRespMessage struct { + Header + ReturnCode byte +} + +func (wt *WillTopicRespMessage) MessageType() byte { + return WILLTOPICRESP +} + +func (wt *WillTopicRespMessage) Write(w io.Writer) error { + packet := wt.Header.pack() + packet.WriteByte(WILLTOPICRESP) + packet.WriteByte(wt.ReturnCode) + _, err := packet.WriteTo(w) + + return err +} + +func (wt *WillTopicRespMessage) Unpack(b io.Reader) { + wt.ReturnCode = readByte(b) +}
diff --git a/packets/willtopicresp_test.go b/packets/willtopicresp_test.go new file mode 100644 index 0000000..e363a17 --- /dev/null +++ b/packets/willtopicresp_test.go
@@ -0,0 +1,29 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillTopicRespStruct(t *testing.T) { + msg := NewMessage(WILLTOPICRESP).(*WillTopicRespMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillTopicRespMessage", reflect.TypeOf(msg).String(), "Type should be WillTopicRespMessage") + assert.Equal(t, 0, msg.ReturnCode, "Default ReturnCode should be 0") + + assert.Equal(t, WILLTOPICRESP, msg.MessageType(), "MessageType() should return WILLTOPICRESP") + } +}
diff --git a/packets/willtopicupd.go b/packets/willtopicupd.go new file mode 100644 index 0000000..9772786 --- /dev/null +++ b/packets/willtopicupd.go
@@ -0,0 +1,57 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "io" +) + +type WillTopicUpdateMessage struct { + Header + Qos byte + Retain bool + WillTopic []byte +} + +func (wt *WillTopicUpdateMessage) MessageType() byte { + return WILLTOPICUPD +} + +func (wt *WillTopicUpdateMessage) encodeFlags() byte { + var b byte + b |= (wt.Qos << 5) & QOSBITS + if wt.Retain { + b |= RETAINFLAG + } + return b +} + +func (wt *WillTopicUpdateMessage) decodeFlags(b byte) { + wt.Qos = (b & QOSBITS) >> 5 + wt.Retain = (b & RETAINFLAG) == RETAINFLAG +} + +func (wt *WillTopicUpdateMessage) Write(w io.Writer) error { + wt.Header.Length = uint16(len(wt.WillTopic) + 3) + packet := wt.Header.pack() + packet.WriteByte(WILLTOPICUPD) + packet.WriteByte(wt.encodeFlags()) + packet.Write(wt.WillTopic) + _, err := packet.WriteTo(w) + + return err +} + +func (wt *WillTopicUpdateMessage) Unpack(b io.Reader) { + wt.decodeFlags(readByte(b)) + b.Read(wt.WillTopic) +}
diff --git a/packets/willtopicupd_test.go b/packets/willtopicupd_test.go new file mode 100644 index 0000000..934c33f --- /dev/null +++ b/packets/willtopicupd_test.go
@@ -0,0 +1,32 @@ +/******************************************************************************* + * Copyright (c) 2014-2015 IBM Corp. + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * which accompanies this distribution, and is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * Contributors: + * Allan Stockdill-Mander - initial API and implementation + *******************************************************************************/ + +package packets + +import ( + "github.com/stretchr/testify/assert" + "reflect" + "testing" +) + +func TestWillTopicUpdateStruct(t *testing.T) { + msg := NewMessage(WILLTOPICUPD).(*WillTopicUpdateMessage) + + if assert.NotNil(t, msg, "New message should not be nil") { + assert.Equal(t, "*packets.WillTopicUpdateMessage", reflect.TypeOf(msg).String(), "Type should be WillTopicUpdateMessage") + assert.Equal(t, 0, msg.Qos, "Default Qos should be 0") + assert.Equal(t, false, msg.Retain, "Default Retain flag should be false") + assert.Equal(t, []byte(nil), msg.WillTopic, "Default WillMsg should be blank") + + assert.Equal(t, WILLTOPICUPD, msg.MessageType(), "MessageType() should return WILLTOPICUPD") + } + +}