| /******************************************************************************* |
| * 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" |
| P "git.eclipse.org/gitroot/paho.incubator/smidge.git/packets" |
| "net" |
| "net/url" |
| "sync" |
| ) |
| |
| type MessageHandler func(client *SNClient, message *P.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]P.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 P.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 P.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(P.CONNECT).(*ConnectToken) |
| c.suTokens[P.CONNECT] = ct |
| cp := P.NewMessage(P.CONNECT).(*P.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(P.REGISTER).(*RegisterToken) |
| t.TopicName = topic |
| r := P.NewMessage(P.REGISTER).(*P.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(P.SUBSCRIBE).(*SubscribeToken) |
| t.handler = mh |
| t.TopicName = topic |
| s := P.NewMessage(P.SUBSCRIBE).(*P.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(P.SUBSCRIBE).(*SubscribeToken) |
| t.handler = mh |
| t.topicType = 0x01 |
| s := P.NewMessage(P.SUBSCRIBE).(*P.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(P.PUBLISH).(*PublishToken) |
| p := P.NewMessage(P.PUBLISH).(*P.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(P.PUBLISH).(*PublishToken) |
| p := P.NewMessage(P.PUBLISH).(*P.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(P.WILLTOPIC).(*WillToken) |
| if c.state != UNCONNECTED { |
| wm := P.NewMessage(P.WILLTOPICUPD).(*P.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(P.WILLTOPIC).(*WillToken) |
| if c.state != UNCONNECTED { |
| wm := P.NewMessage(P.WILLTOPICUPD).(*P.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(P.WILLTOPIC).(*WillToken) |
| if c.state != UNCONNECTED { |
| wm := P.NewMessage(P.WILLTOPICUPD).(*P.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(P.WILLMSGUPD).(*WillToken) |
| if c.state != UNCONNECTED { |
| wm := P.NewMessage(P.WILLMSGUPD).(*P.WillMsgUpdateMessage) |
| c.outgoing <- &MessageAndToken{m: wm, t: wt} |
| return wt |
| } |
| wt.flowComplete() |
| return wt |
| } |
