packet

broker/broker.go

@@ -3,7 +3,6 @@ package broker
 import (
 	"crypto/tls"
 	"hmq/lib/acl"
-	"hmq/lib/message"
 	"hmq/packets"
 	"net"
 	"net/http"
@@ -294,34 +293,35 @@ func (b *Broker) CheckRemoteExist(remoteID, url string) bool {
 
 func (b *Broker) SendLocalSubsToRouter(c *client) {
 	clients := b.clients.Items()
-	subMsg := message.NewSubscribeMessage()
+	subInfo := packets.NewControlPacket(packets.Subscribe).(*packets.SubscribePacket)
 	for _, client := range clients {
 		subs := client.subs
 		for _, sub := range subs {
-			subMsg.AddTopic(sub.topic, sub.qos)
+			subInfo.Topics = append(subInfo.Topics, string(sub.topic))
+			subInfo.Qoss = append(subInfo.Qoss, sub.qos)
 		}
 	}
-	err := c.writeMessage(subMsg)
+	err := c.WriterPacket(subInfo)
 	if err != nil {
 		log.Error("Send localsubs To Router error :", err)
 	}
 }
 
-func (b *Broker) BroadcastInfoMessage(remoteID string, msg message.Message) {
+func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacket) {
 	remotes := b.remotes.Items()
 	for _, r := range remotes {
 		if r.route.remoteID == remoteID {
 			continue
 		}
-		r.writeMessage(msg)
+		r.WriterPacket(msg)
 	}
 	// log.Info("BroadcastInfoMessage success ")
 }
 
-func (b *Broker) BroadcastSubOrUnsubMessage(buf []byte) {
+func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
 	remotes := b.remotes.Items()
 	for _, r := range remotes {
-		r.writeBuffer(buf)
+		r.WriterPacket(packet)
 	}
 	// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
 }
@@ -340,12 +340,8 @@ func (b *Broker) removeClient(c *client) {
 	// log.Info("delete client ,", clientId)
 }
 
-func (b *Broker) ProcessPublishMessage(msg *message.PublishMessage) {
-	if b == nil {
-		return
-	}
-	topic := string(msg.Topic())
-
+func (b *Broker) ProcessPublishMessage(packet *packets.PublishPacket) {
+	topic := packet.TopicName
 	r := b.sl.Match(topic)
 	// log.Info("psubs num: ", len(r.psubs))
 	if len(r.qsubs) == 0 && len(r.psubs) == 0 {
@@ -354,7 +350,7 @@ func (b *Broker) ProcessPublishMessage(msg *message.PublishMessage) {
 
 	for _, sub := range r.psubs {
 		if sub != nil {
-			err := sub.client.writeMessage(msg)
+			err := sub.client.WriterPacket(packet)
 			if err != nil {
 				log.Error("process message for psub error,  ", err)
 			}
@@ -365,7 +361,7 @@ func (b *Broker) ProcessPublishMessage(msg *message.PublishMessage) {
 		// s.qmu.Lock()
 		if cnt, exist := b.queues[string(sub.topic)]; exist && i == cnt {
 			if sub != nil {
-				err := sub.client.writeMessage(msg)
+				err := sub.client.WriterPacket(packet)
 				if err != nil {
 					log.Error("process will message for qsub error,  ", err)
 				}

broker/client.go

@@ -2,7 +2,6 @@ package broker
 
 import (
 	"errors"
-	"hmq/lib/message"
 	"hmq/packets"
 	"net"
 	"strings"
@@ -44,7 +43,7 @@ type info struct {
 	username  string
 	password  []byte
 	keepalive uint16
-	willMsg   packets.ControlPacket
+	willMsg   *packets.PublishPacket
 	localIP   string
 	remoteIP  string
 }
@@ -67,7 +66,7 @@ func (c *client) readLoop(msgPool *MessagePool) {
 	}
 	msg := &Message{}
 	for {
-		packet, err := packets.ReadPacket(conn)
+		packet, err := packets.ReadPacket(nc)
 		// buf, err := ReadPacket(nc)
 		if err != nil {
 			log.Error("read packet error: ", err)
@@ -75,56 +74,65 @@ func (c *client) readLoop(msgPool *MessagePool) {
 			return
 		}
 		msg.client = c
-		msg.msg = packet
+		msg.packet = packet
 		msgPool.queue <- msg
 	}
 	msgPool.Reduce()
 }
 
 func ProcessMessage(msg *Message) {
-	buf := msg.msg
 	c := msg.client
-	if c == nil || buf == nil {
+	ca := msg.packet
+	if c == nil || ca == nil {
 		return
 	}
-	switch m := msg.(type) {
-	case *packets.Connack:
+	switch ca.(type) {
+	case *packets.ConnackPacket:
 		// log.Info("Recv conack message..........")
-		c.ProcessConnAck(buf)
-	case *packets.Connect:
+		packet := ca.(*packets.ConnackPacket)
+		c.ProcessConnAck(packet)
+	case *packets.ConnectPacket:
 		// log.Info("Recv connect message..........")
-		c.ProcessConnect(buf)
-	case *packets.Publish:
+		packet := ca.(*packets.ConnectPacket)
+		c.ProcessConnect(packet)
+	case *packets.PublishPacket:
 		// log.Info("Recv publish message..........")
-		c.ProcessPublish(buf)
-	case *packets.Puback:
+		packet := ca.(*packets.PublishPacket)
+		c.ProcessPublish(packet)
+	case *packets.PubackPacket:
 		//log.Info("Recv publish  ack message..........")
-		c.ProcessPubAck(buf)
-	case *packets.Pubrec::
+		packet := ca.(*packets.PubackPacket)
+		c.ProcessPubAck(packet)
+	case *packets.PubrecPacket:
 		//log.Info("Recv publish rec message..........")
-		c.ProcessPubREC(buf)
-	case *packets.Pubrel:
+		packet := ca.(*packets.PubrecPacket)
+		c.ProcessPubREC(packet)
+	case *packets.PubrelPacket:
 		//log.Info("Recv publish rel message..........")
-		c.ProcessPubREL(buf)
-	case *packets.Pubcomp:
+		packet := ca.(*packets.PubrelPacket)
+		c.ProcessPubREL(packet)
+	case *packets.PubcompPacket:
 		//log.Info("Recv publish  ack message..........")
-		c.ProcessPubComp(buf)
-	case *packets.Subscribe:
+		packet := ca.(*packets.PubcompPacket)
+		c.ProcessPubComp(packet)
+	case *packets.SubscribePacket:
 		// log.Info("Recv subscribe message.....")
-		c.ProcessSubscribe(buf)
-	case *packets.Suback:
+		packet := ca.(*packets.SubscribePacket)
+		c.ProcessSubscribe(packet)
+	case *packets.SubackPacket:
 		// log.Info("Recv suback message.....")
-	case *packets.Unsubscribe:
+	case *packets.UnsubscribePacket:
 		// log.Info("Recv unsubscribe message.....")
-		c.ProcessUnSubscribe(buf)
-	case *packets.Unsuback:
+		packet := ca.(*packets.UnsubscribePacket)
+		c.ProcessUnSubscribe(packet)
+	case *packets.UnsubackPacket:
 		//log.Info("Recv unsuback message.....")
-	case *packets.Pingreq:
+	case *packets.PingreqPacket:
 		// log.Info("Recv PINGREQ message..........")
-		c.ProcessPing(buf)
+		c.ProcessPing()
 	case *packets.PingrespPacket:
 		//log.Info("Recv PINGRESP message..........")
-	case *packets.Disconnect:
+	case *packets.DisconnectPacket:
 		// log.Info("Recv DISCONNECT message.......")
 		c.Close()
 	default:
@@ -132,31 +140,25 @@ func ProcessMessage(msg *Message) {
 	}
 }
 
-func (c *client) ProcessConnect(buf []byte) {
+func (c *client) ProcessConnect(packet *packets.ConnectPacket) {
 
 }
 
-func (c *client) ProcessConnAck(buf []byte) {
+func (c *client) ProcessConnAck(packet *packets.ConnackPacket) {
 
 }
 
-func (c *client) ProcessPublish(buf []byte) {
-	msg, err := DecodePublishMessage(buf)
-	if err != nil {
-		log.Error("Decode Publish Message error: ", err)
-		c.Close()
+func (c *client) ProcessPublish(packet *packets.PublishPacket) {
+	topic := packet.TopicName
+
+	if c.typ != CLIENT || !c.CheckTopicAuth(PUB, topic) {
 		return
 	}
-	topic := msg.Topic()
+	c.ProcessPublishMessage(packet)
 
-	if c.typ != CLIENT || !c.CheckTopicAuth(PUB, string(topic)) {
-		return
-	}
-	c.ProcessPublishMessage(buf, msg)
-
-	if msg.Retain() {
+	if packet.Retain {
 		if b := c.broker; b != nil {
-			err := b.rl.Insert(topic, buf)
+			err := b.rl.Insert([]byte(topic), packet)
 			if err != nil {
 				log.Error("Insert Retain Message error: ", err)
 			}
@@ -165,14 +167,14 @@ func (c *client) ProcessPublish(buf []byte) {
 
 }
 
-func (c *client) ProcessPublishMessage(buf []byte, msg *message.PublishMessage) {
+func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 
 	b := c.broker
 	if b == nil {
 		return
 	}
 	typ := c.typ
-	topic := string(msg.Topic())
+	topic := packet.TopicName
 
 	r := b.sl.Match(topic)
 	// log.Info("psubs num: ", len(r.psubs))
@@ -187,7 +189,7 @@ func (c *client) ProcessPublishMessage(buf []byte, msg *message.PublishMessage)
 			}
 		}
 		if sub != nil {
-			err := sub.client.writeBuffer(buf)
+			err := sub.client.WriterPacket(packet)
 			if err != nil {
 				log.Error("process message for psub error,  ", err)
 			}
@@ -203,7 +205,7 @@ func (c *client) ProcessPublishMessage(buf []byte, msg *message.PublishMessage)
 		// s.qmu.Lock()
 		if cnt, exist := b.queues[string(sub.topic)]; exist && i == cnt {
 			if sub != nil {
-				err := sub.client.writeBuffer(buf)
+				err := sub.client.WriterPacket(packet)
 				if err != nil {
 					log.Error("process will message for qsub error,  ", err)
 				}
@@ -215,47 +217,41 @@ func (c *client) ProcessPublishMessage(buf []byte, msg *message.PublishMessage)
 	}
 }
 
-func (c *client) ProcessPubAck(buf []byte) {
+func (c *client) ProcessPubAck(packet *packets.PubackPacket) {
 
 }
 
-func (c *client) ProcessPubREC(buf []byte) {
+func (c *client) ProcessPubREC(packet *packets.PubrecPacket) {
 
 }
 
-func (c *client) ProcessPubREL(buf []byte) {
+func (c *client) ProcessPubREL(packet *packets.PubrelPacket) {
 
 }
 
-func (c *client) ProcessPubComp(buf []byte) {
+func (c *client) ProcessPubComp(packet *packets.PubcompPacket) {
 
 }
 
-func (c *client) ProcessSubscribe(buf []byte) {
+func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
 	b := c.broker
 	if b == nil {
 		return
 	}
-	msg, err := DecodeSubscribeMessage(buf)
-	if err != nil {
-		log.Error("Decode Subscribe Message error: ", err)
-		c.Close()
-		return
-	}
-	topics := msg.Topics()
-	qos := msg.Qos()
+	topics := packet.Topics
+	qoss := packet.Qoss
 
-	suback := message.NewSubackMessage()
-	suback.SetPacketId(msg.PacketId())
+	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
+	suback.MessageID = packet.MessageID
 	var retcodes []byte
 
-	for i, t := range topics {
-		topic := string(t)
+	for i, topic := range topics {
+		t := []byte(topic)
 		//check topic auth for client
 		if c.typ == CLIENT {
 			if !c.CheckTopicAuth(SUB, topic) {
 				log.Error("CheckSubAuth failed")
-				retcodes = append(retcodes, message.QosFailure)
+				retcodes = append(retcodes, QosFailure)
 				continue
 			}
 		}
@@ -271,13 +267,13 @@ func (c *client) ProcessSubscribe(buf []byte) {
 					}
 					// b.qmu.Unlock()
 				} else {
-					retcodes = append(retcodes, message.QosFailure)
+					retcodes = append(retcodes, QosFailure)
 					continue
 				}
 			}
 			sub := &subscription{
 				topic:  t,
-				qos:    qos[i],
+				qos:    qoss[i],
 				client: c,
 				queue:  queue,
 			}
@@ -289,82 +285,69 @@ func (c *client) ProcessSubscribe(buf []byte) {
 			err := b.sl.Insert(sub)
 			if err != nil {
 				log.Error("Insert subscription error: ", err)
-				retcodes = append(retcodes, message.QosFailure)
+				retcodes = append(retcodes, QosFailure)
 			}
-			retcodes = append(retcodes, qos[i])
+			retcodes = append(retcodes, qoss[i])
 		} else {
 			//if exist ,check whether qos change
-			c.subs[topic].qos = qos[i]
-			retcodes = append(retcodes, qos[i])
+			c.subs[topic].qos = qoss[i]
+			retcodes = append(retcodes, qoss[i])
 		}
 
 	}
 
-	if err := suback.AddReturnCodes(retcodes); err != nil {
-		log.Error("add return suback code error, ", err)
-		// if typ == CLIENT {
-		c.Close()
-		// }
-		return
-	}
+	suback.ReturnCodes = retcodes
 
-	err1 := c.writeMessage(suback)
-	if err1 != nil {
-		log.Error("send suback error, ", err1)
+	err := c.WriterPacket(suback)
+	if err != nil {
+		log.Error("send suback error, ", err)
 		return
 	}
 	//broadcast subscribe message
 	if c.typ == CLIENT {
-		go b.BroadcastSubOrUnsubMessage(buf)
+		go b.BroadcastSubOrUnsubMessage(packet)
 	}
 
 	//process retain message
 	for _, t := range topics {
-		bufs := b.rl.Match(t)
-		for _, buf := range bufs {
-			log.Info("process retain  message: ", string(buf))
-			if buf != nil && string(buf) != "" {
-				c.writeBuffer(buf)
+		packets := b.rl.Match([]byte(t))
+		for _, packet := range packets {
+			log.Info("process retain  message: ", packet)
+			if packet != nil {
+				c.WriterPacket(packet)
 			}
 		}
 	}
 }
 
-func (c *client) ProcessUnSubscribe(buf []byte) {
+func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
 	b := c.broker
 	if b == nil {
 		return
 	}
-
-	unsub, err := DecodeUnsubscribeMessage(buf)
-	if err != nil {
-		log.Error("Decode UnSubscribe Message error: ", err)
-		c.Close()
-		return
-	}
-	topics := unsub.Topics()
+	topics := packet.Topics
 
 	for _, t := range topics {
 		var sub *subscription
 		ok := false
 
-		if sub, ok = c.subs[string(t)]; ok {
+		if sub, ok = c.subs[t]; ok {
 			go c.unsubscribe(sub)
 		}
 
 	}
 
-	resp := message.NewUnsubackMessage()
-	resp.SetPacketId(unsub.PacketId())
+	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
+	unsuback.MessageID = packet.MessageID
 
-	err1 := c.writeMessage(resp)
-	if err1 != nil {
-		log.Error("send ubsuback error, ", err1)
+	err := c.WriterPacket(unsuback)
+	if err != nil {
+		log.Error("send unsuback error, ", err)
 		return
 	}
 	// //process ubsubscribe message
 	if c.typ == CLIENT {
-		b.BroadcastSubOrUnsubMessage(buf)
+		b.BroadcastSubOrUnsubMessage(packet)
 	}
 }
 
@@ -379,16 +362,9 @@ func (c *client) unsubscribe(sub *subscription) {
 	}
 }
 
-func (c *client) ProcessPing(buf []byte) {
-	_, err := DecodePingreqMessage(buf)
-	if err != nil {
-		log.Error("Decode PingRequest Message error: ", err)
-		c.Close()
-		return
-	}
-
-	pingRspMsg := message.NewPingrespMessage()
-	err = c.writeMessage(pingRspMsg)
+func (c *client) ProcessPing() {
+	resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
+	err := c.WriterPacket(resp)
 	if err != nil {
 		log.Error("send PingResponse error, ", err)
 		return
@@ -407,7 +383,7 @@ func (c *client) Close() {
 			}
 		}
 		if c.info.willMsg != nil {
-			// b.ProcessPublishMessage(c.info.willMsg)
+			b.ProcessPublishMessage(c.info.willMsg)
 		}
 	}
 	if c.conn != nil {
@@ -416,6 +392,13 @@ func (c *client) Close() {
 	}
 }
 
+func (c *client) WriterPacket(packet packets.ControlPacket) error {
+	c.mu.Lock()
+	err := packet.Write(c.conn)
+	c.mu.Unlock()
+	return err
+}
+
 func WriteBuffer(conn net.Conn, buf []byte) error {
 	if conn == nil {
 		return errors.New("conn is nul")
@@ -430,10 +413,10 @@ func (c *client) writeBuffer(buf []byte) error {
 	return err
 }
 
-func (c *client) writeMessage(msg message.Message) error {
-	buf, err := EncodeMessage(msg)
-	if err != nil {
-		return err
-	}
-	return c.writeBuffer(buf)
-}
+// func (c *client) writeMessage(msg message.Message) error {
+// 	buf, err := EncodeMessage(msg)
+// 	if err != nil {
+// 		return err
+// 	}
+// 	return c.writeBuffer(buf)
+// }

broker/comm.go

@@ -40,6 +40,12 @@ const (
 	PINGRESP
 	DISCONNECT
 )
+const (
+	QosAtMostOnce byte = iota
+	QosAtLeastOnce
+	QosExactlyOnce
+	QosFailure = 0x80
+)
 
 func SubscribeTopicCheckAndSpilt(subject []byte) ([]string, error) {
 

broker/info.go

@@ -2,7 +2,7 @@ package broker
 
 import (
 	"fmt"
-	"hmq/lib/message"
+	"hmq/packets"
 	"time"
 
 	simplejson "github.com/bitly/go-simplejson"
@@ -13,7 +13,7 @@ func (c *client) SendInfo() {
 	url := c.info.localIP + ":" + c.broker.config.Cluster.Port
 
 	infoMsg := NewInfo(c.broker.id, url, false)
-	err := c.writeMessage(infoMsg)
+	err := c.WriterPacket(infoMsg)
 	if err != nil {
 		log.Error("send info message error, ", err)
 		return
@@ -23,11 +23,11 @@ func (c *client) SendInfo() {
 
 func (c *client) StartPing() {
 	timeTicker := time.NewTicker(time.Second * 30)
-	ping := message.NewPingreqMessage()
+	ping := packets.NewControlPacket(packets.Pingreq).(*packets.PingreqPacket)
 	for {
 		select {
 		case <-timeTicker.C:
-			err := c.writeMessage(ping)
+			err := c.WriterPacket(ping)
 			if err != nil {
 				log.Error("ping error: ", err)
 			}
@@ -37,11 +37,12 @@ func (c *client) StartPing() {
 
 func (c *client) SendConnect() {
 
-	clientID := c.info.clientID
-	connMsg := message.NewConnectMessage()
-	connMsg.SetClientId([]byte(clientID))
-	connMsg.SetVersion(0x04)
-	err := c.writeMessage(connMsg)
+	m := packets.NewControlPacket(packets.Connect).(*packets.ConnectPacket)
+
+	m.CleanSession = true
+	m.ClientIdentifier = c.info.clientID
+	m.Keepalive = uint16(60)
+	err := c.WriterPacket(m)
 	if err != nil {
 		log.Error("send connect message error, ", err)
 		return
@@ -49,27 +50,27 @@ func (c *client) SendConnect() {
 	// log.Info("send connet success")
 }
 
-func NewInfo(sid, url string, isforword bool) *message.PublishMessage {
-	infoMsg := message.NewPublishMessage()
-	infoMsg.SetTopic([]byte(BrokerInfoTopic))
+func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
+	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
+	pub.Qos = 0
+	pub.TopicName = BrokerInfoTopic
+	pub.Retain = false
 	info := fmt.Sprintf(`{"remoteID":"%s","url":"%s","isForward":%t}`, sid, url, isforword)
 	// log.Info("new info", string(info))
-	infoMsg.SetPayload([]byte(info))
-	infoMsg.SetQoS(0)
-	infoMsg.SetRetain(false)
-	return infoMsg
+	pub.Payload = []byte(info)
+	return pub
 }
 
-func (c *client) ProcessInfo(msg *message.PublishMessage) {
+func (c *client) ProcessInfo(packet *packets.PublishPacket) {
 	nc := c.conn
 	b := c.broker
 	if nc == nil {
 		return
 	}
 
-	log.Info("recv remoteInfo: ", string(msg.Payload()))
+	log.Info("recv remoteInfo: ", string(packet.Payload))
 
-	js, e := simplejson.NewJson(msg.Payload())
+	js, e := simplejson.NewJson(packet.Payload)
 	if e != nil {
 		log.Warn("parse info message err", e)
 		return

broker/msgpool.go

@@ -14,7 +14,7 @@ const (
 
 type Message struct {
 	client *client
-	msg    packets.ControlPacket
+	packet packets.ControlPacket
 }
 
 var (

broker/retain.go

@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"hmq/packets"
 	"sync"
 )
 
@@ -13,14 +14,14 @@ type rlevel struct {
 }
 type rnode struct {
 	next *rlevel
-	msg  []byte
+	msg  *packets.PublishPacket
 }
 type RetainResult struct {
-	msg [][]byte
+	msg []*packets.PublishPacket
 }
 
 func newRNode() *rnode {
-	return &rnode{msg: make([]byte, 0, 4)}
+	return &rnode{}
 }
 
 func newRLevel() *rlevel {
@@ -31,7 +32,7 @@ func NewRetainList() *RetainList {
 	return &RetainList{root: newRLevel()}
 }
 
-func (r *RetainList) Insert(topic, buf []byte) error {
+func (r *RetainList) Insert(topic []byte, buf *packets.PublishPacket) error {
 
 	tokens, err := PublishTopicCheckAndSpilt(topic)
 	if err != nil {
@@ -58,7 +59,7 @@ func (r *RetainList) Insert(topic, buf []byte) error {
 	return nil
 }
 
-func (r *RetainList) Match(topic []byte) [][]byte {
+func (r *RetainList) Match(topic []byte) []*packets.PublishPacket {
 
 	tokens, err := SubscribeTopicCheckAndSpilt(topic)
 	if err != nil {
@@ -110,7 +111,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) {
 
 func (r *rnode) GetAll(results *RetainResult) {
 	// log.Info("node 's message: ", string(r.msg))
-	if r.msg != nil && string(r.msg) != "" {
+	if r.msg != nil {
 		results.msg = append(results.msg, r.msg)
 	}
 	l := r.next