store.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")
		}
	}
}