persistence

README

Persistence

Persistence API for VolantMQ

Documentation

Index

• type Errors
  • func (e Errors) Error() string
• type Expiry
• type IFace
• type PacketLoader
• type Packets
• type PersistedPacket
• type PersistedPackets
• type Retained
• type SessionBase
• type SessionDelays
• type SessionLoader
• type SessionState
• type Sessions
• type State
• type Subscriptions
• type System
• type SystemState

Types

type Errors

type Errors int

Errors persistence errors

const (
	// ErrInvalidArgs invalid arguments provided
	ErrInvalidArgs Errors = iota
	// ErrUnknownProvider if provider is unknown
	ErrUnknownProvider
	// ErrInvalidConfig provided config is invalid
	ErrInvalidConfig
	// ErrAlreadyExists object already exists
	ErrAlreadyExists
	// ErrNotInitialized persistence provider not initialized yet
	ErrNotInitialized
	// ErrNotFound object not found
	ErrNotFound
	// ErrNotOpen storage is not open
	ErrNotOpen
	// ErrBrokenEntry persisted entry does not meet requirements
	ErrBrokenEntry
)

func (Errors) Error

func (e Errors) Error() string

Errors description during persistence

type Expiry

type Expiry interface {
	ExpiryStore(id []byte, delays *SessionDelays) error
	ExpiryDelete(id []byte) error
}

Expiry session expiration interface

type IFace

type IFace interface {
	Sessions() (Sessions, error)
	Retained() (Retained, error)
	System() (System, error)
	Shutdown() error
}

IFace interface implemented by different backends

type PacketLoader

type PacketLoader func(interface{}, *PersistedPacket) (bool, error)

PacketLoader application callback doing packed decode when return true in first return packet will be deleted if error presented load interrupted after current packet

type Packets

type Packets interface {
	PacketCountQoS0(id []byte) (uint64, error)
	PacketCountQoS12(id []byte) (uint64, error)
	PacketCountUnAck(id []byte) (uint64, error)
	PacketStoreQoS0(id []byte, packets *PersistedPacket) error
	PacketStoreQoS12(id []byte, packets *PersistedPacket) error
	PacketsForEachQoS0(id []byte, ctx interface{}, loader PacketLoader) error
	PacketsForEachQoS12(id []byte, ctx interface{}, loader PacketLoader) error
	PacketsForEachUnAck(id []byte, ctx interface{}, loader PacketLoader) error
	PacketsStore(id []byte, packets PersistedPackets) error
	PacketsDelete(id []byte) error
}

Packets interface for connection to handle packets

type PersistedPacket

type PersistedPacket struct {
	// ExpireAt shows either packet has expiration value
	ExpireAt string
	// Data is encoded byte stream as it goes over network
	Data []byte
}

PersistedPacket wraps packet to handle misc cases like expiration

type PersistedPackets

type PersistedPackets struct {
	QoS0  []*PersistedPacket
	QoS12 []*PersistedPacket
	UnAck []*PersistedPacket
}

PersistedPackets array of persisted packets

type Retained

type Retained interface {
	// Store persist retained message
	// it wipes previously set values
	Store([]*PersistedPacket) error
	// Load load retained messages
	Load() ([]*PersistedPacket, error)
	// Wipe retained storage
	Wipe() error
}

Retained provider for load/store retained messages

type SessionBase

type SessionBase struct {
	Timestamp string
	Version   byte
}

SessionBase ...

type SessionDelays

type SessionDelays struct {
	Since    string
	ExpireIn string
	Will     []byte
}

SessionDelays formerly known as expiry set timestamp to handle will delay and/or expiration

type SessionLoader

type SessionLoader interface {
	LoadSession(ctx interface{}, id []byte, state *SessionState) error
}

SessionLoader implemented by session manager to load persisted sessions when server starts

type SessionState

type SessionState struct {
	Subscriptions []byte
	Errors        []error
	Expire        *SessionDelays
	SessionBase
}

SessionState object

type Sessions

type Sessions interface {
	Packets
	Subscriptions
	State
	Expiry
	Create(id []byte, state *SessionBase) error
	Count() uint64
	LoadForEach(loader SessionLoader, ctx interface{}) error
	// Exists check session if presented in storage
	Exists(id []byte) bool
	Delete(id []byte) error
}

Sessions interface allows operating with sessions inside backend

type State

type State interface {
	StateStore(id []byte, state *SessionState) error
	StateDelete(id []byte) error
}

State session state interface

type Subscriptions

type Subscriptions interface {
	SubscriptionsStore([]byte, []byte) error
	SubscriptionsDelete([]byte) error
}

Subscriptions session subscriptions interface

type System

type System interface {
	GetInfo() (*SystemState, error)
}

System persistence state of the system configuration

type SystemState

type SystemState struct {
	Version   string
	CreatedAt string
}

SystemState system configuration