141 lines
3.8 KiB
Go
141 lines
3.8 KiB
Go
/*
|
|
* Copyright (c) 2021 IBM Corp and others.
|
|
*
|
|
* All rights reserved. This program and the accompanying materials
|
|
* are made available under the terms of the Eclipse Public License v2.0
|
|
* and Eclipse Distribution License v1.0 which accompany this distribution.
|
|
*
|
|
* The Eclipse Public License is available at
|
|
* https://www.eclipse.org/legal/epl-2.0/
|
|
* and the Eclipse Distribution License is available at
|
|
* http://www.eclipse.org/org/documents/edl-v10.php.
|
|
*
|
|
* Contributors:
|
|
* Seth Hoenig
|
|
* Allan Stockdill-Mander
|
|
* Mike Robertson
|
|
*/
|
|
|
|
package mqtt
|
|
|
|
import (
|
|
"fmt"
|
|
"strconv"
|
|
|
|
"github.com/eclipse/paho.mqtt.golang/packets"
|
|
)
|
|
|
|
const (
|
|
inboundPrefix = "i."
|
|
outboundPrefix = "o."
|
|
)
|
|
|
|
// Store is an interface which can be used to provide implementations
|
|
// for message persistence.
|
|
// Because we may have to store distinct messages with the same
|
|
// message ID, we need a unique key for each message. This is
|
|
// possible by prepending "i." or "o." to each message id
|
|
type Store interface {
|
|
Open()
|
|
Put(key string, message packets.ControlPacket)
|
|
Get(key string) packets.ControlPacket
|
|
All() []string
|
|
Del(key string)
|
|
Close()
|
|
Reset()
|
|
}
|
|
|
|
// A key MUST have the form "X.[messageid]"
|
|
// where X is 'i' or 'o'
|
|
func mIDFromKey(key string) uint16 {
|
|
s := key[2:]
|
|
i, err := strconv.ParseUint(s, 10, 16)
|
|
chkerr(err)
|
|
return uint16(i)
|
|
}
|
|
|
|
// Return true if key prefix is outbound
|
|
func isKeyOutbound(key string) bool {
|
|
return key[:2] == outboundPrefix
|
|
}
|
|
|
|
// Return true if key prefix is inbound
|
|
func isKeyInbound(key string) bool {
|
|
return key[:2] == inboundPrefix
|
|
}
|
|
|
|
// Return a string of the form "i.[id]"
|
|
func inboundKeyFromMID(id uint16) string {
|
|
return fmt.Sprintf("%s%d", inboundPrefix, id)
|
|
}
|
|
|
|
// Return a string of the form "o.[id]"
|
|
func outboundKeyFromMID(id uint16) string {
|
|
return fmt.Sprintf("%s%d", outboundPrefix, id)
|
|
}
|
|
|
|
// govern which outgoing messages are persisted
|
|
func persistOutbound(s Store, m packets.ControlPacket) {
|
|
switch m.Details().Qos {
|
|
case 0:
|
|
switch m.(type) {
|
|
case *packets.PubackPacket, *packets.PubcompPacket:
|
|
// Sending puback. delete matching publish
|
|
// from ibound
|
|
s.Del(inboundKeyFromMID(m.Details().MessageID))
|
|
}
|
|
case 1:
|
|
switch m.(type) {
|
|
case *packets.PublishPacket, *packets.PubrelPacket, *packets.SubscribePacket, *packets.UnsubscribePacket:
|
|
// Sending publish. store in obound
|
|
// until puback received
|
|
s.Put(outboundKeyFromMID(m.Details().MessageID), m)
|
|
default:
|
|
ERROR.Println(STR, "Asked to persist an invalid message type")
|
|
}
|
|
case 2:
|
|
switch m.(type) {
|
|
case *packets.PublishPacket:
|
|
// Sending publish. store in obound
|
|
// until pubrel received
|
|
s.Put(outboundKeyFromMID(m.Details().MessageID), m)
|
|
default:
|
|
ERROR.Println(STR, "Asked to persist an invalid message type")
|
|
}
|
|
}
|
|
}
|
|
|
|
// govern which incoming messages are persisted
|
|
func persistInbound(s Store, m packets.ControlPacket) {
|
|
switch m.Details().Qos {
|
|
case 0:
|
|
switch m.(type) {
|
|
case *packets.PubackPacket, *packets.SubackPacket, *packets.UnsubackPacket, *packets.PubcompPacket:
|
|
// Received a puback. delete matching publish
|
|
// from obound
|
|
s.Del(outboundKeyFromMID(m.Details().MessageID))
|
|
case *packets.PublishPacket, *packets.PubrecPacket, *packets.PingrespPacket, *packets.ConnackPacket:
|
|
default:
|
|
ERROR.Println(STR, "Asked to persist an invalid messages type")
|
|
}
|
|
case 1:
|
|
switch m.(type) {
|
|
case *packets.PublishPacket, *packets.PubrelPacket:
|
|
// Received a publish. store it in ibound
|
|
// until puback sent
|
|
s.Put(inboundKeyFromMID(m.Details().MessageID), m)
|
|
default:
|
|
ERROR.Println(STR, "Asked to persist an invalid messages type")
|
|
}
|
|
case 2:
|
|
switch m.(type) {
|
|
case *packets.PublishPacket:
|
|
// Received a publish. store it in ibound
|
|
// until pubrel received
|
|
s.Put(inboundKeyFromMID(m.Details().MessageID), m)
|
|
default:
|
|
ERROR.Println(STR, "Asked to persist an invalid messages type")
|
|
}
|
|
}
|
|
}
|