support qod downgrade

Author: chobie

encoding/mqtt/mqtt.go

@@ -163,6 +163,39 @@ func NewConnackMessage() *ConnackMessage {
 	return message
 }
 
+func CopyPublishMessage(msg *PublishMessage) (*PublishMessage, error) {
+	v, err := CopyMessage(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	return v.(*PublishMessage), nil
+}
+
+func CopyMessage(msg Message) (Message, error) {
+	var result Message
+
+	switch (msg.GetType()) {
+	case PACKET_TYPE_PUBLISH:
+		t := msg.(*PublishMessage)
+		c := NewPublishMessage()
+		c.Payload = t.Payload
+		c.TopicName = t.TopicName
+		c.PacketIdentifier = t.PacketIdentifier
+		c.Opaque = t.Opaque
+		c.FixedHeader.Type = t.FixedHeader.Type
+		c.FixedHeader.Dupe = t.FixedHeader.Dupe
+		c.FixedHeader.QosLevel = t.FixedHeader.QosLevel
+		c.FixedHeader.Retain = t.FixedHeader.Retain
+		c.FixedHeader.RemainingLength = t.FixedHeader.RemainingLength
+		result = c
+		break
+	default:
+		return nil, errors.New("hoge")
+	}
+	return result, nil
+}
+
 // TODO: このアホっぽい感じどうにかしたいなー
 // TODO: 読み込んだサイズ返す
 // TODO: サイズ超えてたらエラーなげるの

encoding/mqtt/subscribe.go

@@ -4,6 +4,8 @@ import (
 	"bytes"
 	"encoding/binary"
 	"io"
+
+	"fmt"
 )
 
 type SubscribePayload struct {
@@ -51,7 +53,9 @@ func (self *SubscribeMessage) decode(reader io.Reader) error {
 
 		_, _ = io.CopyN(buffer, reader, int64(length))
 		m.TopicPath = string(buffer.Bytes())
+		// TODO: これ実装まちがってね？
 		binary.Read(reader, binary.BigEndian, &m.RequestedQos)
+		fmt.Printf(" === requested qos %d\n", m.RequestedQos)
 		self.Payload = append(self.Payload, m)
 
 		buffer.Reset()

server/connection.go

@@ -54,8 +54,9 @@ type Connection interface {
 	GetWillMessage() *mqtt.WillMessage
 	HasWillMessage() bool
 	GetOutGoingTable() *util.MessageTable
-	GetSubscribedTopics() []string
-	AppendSubscribedTopic(string)
+	GetSubscribedTopicQos(string) int
+	GetSubscribedTopics() map[string]int
+	AppendSubscribedTopic(string, int)
 	RemoveSubscribedTopic(string)
 	SetKeepaliveInterval(int)
 	GetId() string

server/engine.go

@@ -215,20 +215,32 @@ func (self *Pidgey) Run() {
 				// で、Pubackが帰ってきたらrefcountを下げて0になったらMessageを消す
 				log.Debug("TopicName: %s", m.TopicName)
 				targets := self.Qlobber.Match(m.TopicName)
-				if m.QosLevel > 0 {
-					// TODO: ClientごとにInflightTableを持つ
-					// engineのOutGoingTableなのはとりあえず、ということ
-					id := self.OutGoingTable.NewId()
-					m.PacketIdentifier = id
-
-					if sender, ok := m.Opaque.(Connection); ok {
-						self.OutGoingTable.Register2(m.PacketIdentifier, m, len(targets), sender)
+				for i := range targets {
+					cn := targets[i].(Connection)
+					x, err := codec.CopyPublishMessage(m)
+					if err != nil {
+						continue
 					}
-				}
 
-				for i := range targets {
-					log.Debug("sending publish message to %s [%s %s %d %d]", targets[i].(Connection).GetId(), m.TopicName, m.Payload, m.PacketIdentifier, m.QosLevel)
-					targets[i].(Connection).WriteMessageQueue(m)
+					subscriberQos := cn.GetSubscribedTopicQos(m.TopicName)
+					// Downgrade QoS
+					if subscriberQos < x.QosLevel {
+						log.Debug("===Downgrade QoS %d > %d", x.QosLevel, subscriberQos)
+						x.QosLevel = subscriberQos
+						// client側はtopic filterで持ってるから単純にはqosわかんねーんだよな
+					}
+
+					if x.QosLevel > 0 {
+						// TODO: ClientごとにInflightTableを持つ
+						// engineのOutGoingTableなのはとりあえず、ということ
+						id := self.OutGoingTable.NewId()
+						x.PacketIdentifier = id
+						if sender, ok := x.Opaque.(Connection); ok {
+							self.OutGoingTable.Register2(x.PacketIdentifier, x, len(targets), sender)
+						}
+					}
+					log.Debug("sending publish message to %s [%s %s %d %d]", targets[i].(Connection).GetId(), x.TopicName, x.Payload, x.PacketIdentifier, x.QosLevel)
+					cn.WriteMessageQueue(x)
 				}
 				break
 			default:
@@ -243,7 +255,7 @@ func (self *Pidgey) Run() {
 }
 
 func (self *Pidgey) CleanHoge(conn Connection) {
-	for _, t := range conn.GetSubscribedTopics() {
+	for t, _ := range conn.GetSubscribedTopics() {
 		if conn.ShouldClearSession() {
 			self.Qlobber.Remove(t, conn)
 		}
@@ -388,7 +400,7 @@ func (self *Pidgey) handle(conn Connection) error {
 			self.SubscribeMap[conn.GetId()] = payload.TopicPath
 
 			self.Qlobber.Add(payload.TopicPath, conn)
-			conn.AppendSubscribedTopic(payload.TopicPath)
+			conn.AppendSubscribedTopic(payload.TopicPath, int(payload.RequestedQos))
 
 			// TODO: これはAtomicにさせたいなー、とおもったり。
 			// というかTopicは実装上もうつかってないので消していいや

server/mmux_connection.go

@@ -204,20 +204,29 @@ func (self *MmuxConnection) GetOutGoingTable() *util.MessageTable {
 //	return self.PrimaryConnection.GetOutGoingTable()
 }
 
-func (self *MmuxConnection) GetSubscribedTopics() []string {
+func (self *MmuxConnection) GetSubscribedTopicQos(topic string) int {
+	if self.PrimaryConnection == nil {
+		fmt.Printf("this is dummy connection\n")
+		return 0
+	}
+
+	return self.PrimaryConnection.GetSubscribedTopicQos(topic)
+}
+
+func (self *MmuxConnection) GetSubscribedTopics() map[string]int {
 	if self.PrimaryConnection == nil {
 		return nil
 	}
 
 	return self.PrimaryConnection.GetSubscribedTopics()
 }
 
-func (self *MmuxConnection) AppendSubscribedTopic(topic string) {
+func (self *MmuxConnection) AppendSubscribedTopic(topic string, qos int) {
 	if self.PrimaryConnection == nil {
 		return
 	}
 
-	self.PrimaryConnection.AppendSubscribedTopic(topic)
+	self.PrimaryConnection.AppendSubscribedTopic(topic, qos)
 }
 
 func (self *MmuxConnection) RemoveSubscribedTopic(topic string) {

server/tcp_connection.go

@@ -21,12 +21,13 @@ type TcpConnection struct {
 	yield func(conn Connection, time time.Time)
 	WillMessage *mqtt.WillMessage
 	OutGoingTable *util.MessageTable
-	SubscribedTopics []string
+	SubscribedTopics map[string]int
 	WriteQueue chan mqtt.Message
 	WriteQueueFlag chan bool
 	Last time.Time
 	KeepaliveInterval int
 	ClearSession bool
+	Qlobber *util.Qlobber
 }
 
 func (self *TcpConnection) SetWillMessage(will mqtt.WillMessage) {
@@ -77,6 +78,8 @@ func NewTcpConnection(socket net.Conn, server Server, retry chan *Retryable, yie
 		KeepaliveInterval: 0,
 		Last: time.Now(),
 		ClearSession: true,
+		SubscribedTopics: make(map[string]int),
+		Qlobber: util.NewQlobber(),
 	}
 
 	readMessage := make([]byte, 0, MAX_REQUEST_SIZE)
@@ -126,24 +129,33 @@ func NewTcpConnection(socket net.Conn, server Server, retry chan *Retryable, yie
 	return conn
 }
 
-func (self *TcpConnection) GetSubscribedTopics() []string {
+func (self *TcpConnection) GetSubscribedTopicQos(topic string) int {
+	v := self.Qlobber.Match(topic)
+	if r, ok := v[0].(int); ok {
+		return r;
+	}
+	return -1
+//	if qos, ok := self.SubscribedTopics[topic]; ok {
+//		return qos
+//	}
+//	return -1
+}
+
+func (self *TcpConnection) GetSubscribedTopics() map[string]int {
 	return self.SubscribedTopics
 }
 
-func (self *TcpConnection) AppendSubscribedTopic(topic string) {
-	self.SubscribedTopics = append(self.SubscribedTopics, topic)
+func (self *TcpConnection) AppendSubscribedTopic(topic string, qos int) {
+	self.SubscribedTopics[topic] = qos
+	self.Qlobber.Add(topic, qos)
 }
 
 func (self *TcpConnection) RemoveSubscribedTopic(topic string) {
-	offset := -1
-	for i, v := range self.SubscribedTopics {
-		if v == topic {
-			offset = i
-			break
-		}
-	}
+	self.Qlobber.Remove(topic, nil)
 
-	self.SubscribedTopics = append(self.SubscribedTopics[:offset], self.SubscribedTopics[:offset+1]...)
+	if _, ok := self.SubscribedTopics[topic]; ok {
+		delete(self.SubscribedTopics, topic)
+	}
 }
 
 func (self *TcpConnection) GetSocket() net.Conn {