state

README

paho.golang/paho/session/state

This is an implementation of the paho.golang/paho/session.SessionManager interface and is the implementation used by default (should be suitable for mose use cases but can be replaced if it does not meet your needs).

Documentation

Index

• Variables
• type State
  • func New(client storer, server storer) *State
  • func NewInMemory() *State
  • func (s *State) Ack(pb *packets.Publish) error
  • func (s *State) AddToSession(ctx context.Context, packet session.Packet, resp chan<- packets.ControlPacket) error
  • func (s *State) AllocateClientPacketIDForTest(packetID uint16, forPacketType byte, resp chan<- packets.ControlPacket)
  • func (s *State) Close() error
  • func (s *State) ConAckReceived(conn io.Writer, cp *packets.Connect, ca *packets.Connack) error
  • func (s *State) ConnectionLost(dp *packets.Disconnect) error
  • func (s *State) PacketReceived(recv *packets.ControlPacket, pubChan chan<- *packets.Publish) error
  • func (s *State) SetDebugLogger(l paholog.Logger)
  • func (s *State) SetErrorLogger(l paholog.Logger)

Variables

var ErrUnexpectedRelease = errors.New("release called when quota at initial value")

ErrUnexpectedRelease is for logging only (to help identify if there are issues with state management)

Types

type State

type State struct {
	// contains filtered or unexported fields
}

State manages the session state. The client will send messages that may impact the state via us, and we will maintain the session state

func New

func New(client storer, server storer) *State

New creates a new state which will persist information using the passed in storer's.

func NewInMemory

func NewInMemory() *State

NewInMemory returns a default State that stores all information in memory

func (*State) Ack

func (s *State) Ack(pb *packets.Publish) error

Ack is called when the client message handlers have completed (or, if manual acknowledgements are enabled, when `client.ACK()` has been called - this may happen some time after the message was received and it is conceivable that the connection may have been dropped and reestablished in the interim). See issue 160 re issues when the State is called after the connection is dropped. We assume that the user will ensure that all ACK's are completed before the State is applied to a new connection (not doing this may have unpredictable results).

func (*State) AddToSession

func (s *State) AddToSession(ctx context.Context, packet session.Packet, resp chan<- packets.ControlPacket) error

AddToSession adds a packet to the session state (including allocation of a Message Identifier). If this function returns a nil then:

• A slot has been allocated if the packet is a PUBLISH (function will block if RECEIVE MAXIMUM messages are inflight)
• A message Identifier has been added to the passed in packet
• Publish messages will have been written to the store (and will be automatically transmitted if a new connection is established before the message is fully acknowledged - subject to state rules in the MQTTv5 spec)
• Something will be sent to `resp` when either the message is fully acknowledged or the packet is removed from the session (in which case nil will be sent).

If the function returns an error, then any actions taken will be rewound prior to return.

func (*State) AllocateClientPacketIDForTest

func (s *State) AllocateClientPacketIDForTest(packetID uint16, forPacketType byte, resp chan<- packets.ControlPacket)

AllocateClientPacketIDForTest is intended for use in tests only. It allocates a packet ID in the client session state This feels like a hack but makes it easier to test packet identifier exhaustion

func (*State) Close

func (s *State) Close() error

Close closes the session state

func (*State) ConAckReceived

func (s *State) ConAckReceived(conn io.Writer, cp *packets.Connect, ca *packets.Connack) error

ConAckReceived will be called when the client receives a CONACK that indicates the connection has been successfully established. This indicates that a new connection is live and the passed in connection should be used going forward. It is also the trigger to resend any queued messages. Note that this function should not be called concurrently with others (we should not begin sending/receiving packets until after the CONACK has been processed). TODO: Add errors() function so we can notify the client of errors whilst transmitting the session stuff?

func (*State) ConnectionLost

func (s *State) ConnectionLost(dp *packets.Disconnect) error

ConnectionLost will be called when the connection is lost; either because we received a DISCONNECT packet or due to a network error (`nil` will be passed in)

func (*State) PacketReceived

func (s *State) PacketReceived(recv *packets.ControlPacket, pubChan chan<- *packets.Publish) error

PacketReceived should be called whenever one of the following is received: `PUBLISH` (QOS1+ only), `PUBACK`, `PUBREC`, `PUBREL`, `PUBCOMP`, `SUBACK`, `UNSUBACK` It will handle sending any neccessary response or passing the message to the client. pubChan will be sent a `PUBLISH` if applicable (and a receiver must be active whilst this function runs)

func (*State) SetDebugLogger

func (s *State) SetDebugLogger(l paholog.Logger)

SetDebugLogger takes an instance of the paho Logger interface and sets it to be used by the debug log endpoint

func (*State) SetErrorLogger

func (s *State) SetErrorLogger(l paholog.Logger)

SetErrorLogger takes an instance of the paho Logger interface and sets it to be used by the error log endpoint