session

Documentation

Overview

Package session implements MQTT sessions.

Index

• Variables
• type Session
  • func New(ctx context.Context, conn net.Conn, deliver func(*packet.PublishPacket)) Session
• type Stats
• type Store
  • func SimpleStore() Store

Variables

var PublishBufferSize = 64

PublishBufferSize sets the size of publish channel buffers

Types

type Session

type Session interface {
	Context() context.Context

	AuthInfo() auth.Info

	PublishChan() <-chan *packet.PublishPacket

	Stats() Stats

	// Read and handle the connect packet
	ReadConnect() error

	// Read and handle the next control packet, optionally returning a response
	ReadPacket() (packet.ControlPacket, error)

	// Handle a Disconnect packet
	// unsets the will
	HandleDisconnect()

	// Send an outgoing Publish message if the session is subscribed to the topic
	// if subscription with QoS 0: sends the message
	// if subscription with QoS 1: sends the message, stores the message until Puback
	// if subscription with QoS 2: sends the message, stores the message until Pubrec
	// if authentication is enabled, the server checks if the client is allowed to receive on the topic
	Publish(pkt *packet.PublishPacket)

	// Handle an incoming Publish packet
	// if QoS 0: delivers the packet and returns nil
	// if QoS 1: delivers the packet and returns a *PubackPacket
	// if QoS 2: delivers the packet, stores Pubrec until Pubrel, returns *PubrecPacket
	// if authentication is enabled, the server checks if the client is allowed to publish on the topic
	HandlePublish(pkt *packet.PublishPacket) (packet.ControlPacket, error)

	// Handle a Puback packet
	// clears pkt that was waiting for Puback
	HandlePuback(pkt *packet.PubackPacket) error

	// Handle a Pubrec packet
	// clears pkt that was waiting for Pubrec, stores Pubrel until Pubcomp, returns *PubrelPacket
	HandlePubrec(pkt *packet.PubrecPacket) (*packet.PubrelPacket, error)

	// Handle a Pubrel packet
	// clears Pubrec that was waiting for Pubrel, returns *PubcompPacket
	HandlePubrel(pkt *packet.PubrelPacket) (*packet.PubcompPacket, error)

	// Handle a Pubcomp packet
	// clears Pubrel that was waiting for Pubcomp
	HandlePubcomp(pkt *packet.PubcompPacket) error

	// Pending messages that should be retransmitted on a reconnect
	Pending() []packet.ControlPacket

	// Handle a Subscribe packet
	// adds subscriptions, returns *SubackPacket
	// if authentication is enabled, the server checks if the client is allowed to subscribe to the topic
	HandleSubscribe(pkt *packet.SubscribePacket) (*packet.SubackPacket, error)

	// Handle an Unsubscribe packet
	// removes subscriptions, returns *UnsubackPacket
	HandleUnsubscribe(pkt *packet.UnsubscribePacket) (*packet.UnsubackPacket, error)

	// Subscriptions of the session
	Subscriptions() map[string]byte

	// Close the session
	// closes the connection
	// delivers the will (if set) and then unsets it
	// clears the session state
	Close()
}

Session interface

func New

func New(ctx context.Context, conn net.Conn, deliver func(*packet.PublishPacket)) Session

type Stats

type Stats struct {
	Published uint64
	Delivered uint64
}

Stats for the session

type Store

type Store interface {
	All() []Session
	Store(Session)
	Delete(Session)
	Publish(pkt *packet.PublishPacket)
}

Store interface

func SimpleStore

func SimpleStore() Store

SimpleStore returns a simple Store implementation and starts a goroutine that keeps the store clean