broker

Aug 1, 2018

Changes in this version

type Backend

+ Log func(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, ...)

type MemoryBackend

+ Logger func(LogEvent, *Client, packet.Generic, *packet.Message, error)

+ func (m *MemoryBackend) Log(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, ...)

Jul 16, 2018

Changes in this version

type Backend

+ Restore func(client *Client) error

type Client

+ ParallelSubscribes int

type MemoryBackend

+ func (m *MemoryBackend) Restore(client *Client) error

Jul 14, 2018

Changes in this version

type LogEvent

+ const MessageDequeued

Jul 10, 2018

Changes in this version

+ var ErrClientClosed = errors.New("client has been closed")

+ var ErrClientDisconnected = errors.New("client has disconnected")

+ var ErrClosing = errors.New("closing")

+ var ErrKillTimeout = errors.New("kill timeout")

+ var ErrQueueFull = errors.New("queue full")

+ type Ack func(message *packet.Message)

type Backend

+ Dequeue func(client *Client) (*packet.Message, Ack, error)

+ Restored func(client *Client) error

type Client

+ PacketPrefetch int

+ ParallelDequeues int

+ ParallelPublishes int

+ func NewClient(backend Backend, logger Logger, conn transport.Conn) *Client

+ func (c *Client) Closed() <-chan struct{}

+ func (c *Client) Closing() <-chan struct{}

+ func (c *Client) Conn() transport.Conn

+ func (c *Client) ID() string

type Engine

+ DefaultReadBuffer int

+ DefaultWriteBuffer int

+ OnError func(error)

type LogEvent

+ const ClientDisconnected

+ const MessageAcknowledged

type MemoryBackend

+ KillTimeout time.Duration

+ SessionQueueSize int

+ func (m *MemoryBackend) Close(timeout time.Duration) bool

+ func (m *MemoryBackend) Dequeue(client *Client) (*packet.Message, Ack, error)

+ func (m *MemoryBackend) Restored(client *Client) error

Apr 29, 2018

Changes in this version

+ var ErrNotAuthorized = errors.New("client is not authorized")

Apr 17, 2018

Changes in this version

+ var ErrMissingSession = errors.New("no session returned from Backend")

Dec 28, 2017

Changes in this version

+ type MessageQueue struct

+ func NewMessageQueue(size int) *MessageQueue

+ func (q *MessageQueue) Len() int

+ func (q *MessageQueue) Pop() *packet.Message

+ func (q *MessageQueue) Push(msg *packet.Message)

+ func (q *MessageQueue) Range(fn func(*packet.Message) bool)

+ func (q *MessageQueue) Reset()

Nov 7, 2017

Changes in this version

type Client

+ Ref interface{}

Oct 30, 2017

Changes in this version

type MemoryBackend

+ Credentials map[string]string

type Session

+ DequeueOffline func() *packet.Message

+ NextID func() packet.ID

+ QueueOffline func(*packet.Message)

Sep 14, 2017

Changes in this version

+ var ErrExpectedConnect = errors.New("expected a ConnectPacket as the first packet")

+ func Run(t *testing.T, engine *Engine, protocol string) (string, chan struct{}, chan struct{})

+ type Backend interface

+ Authenticate func(client *Client, user, password string) (bool, error)

+ ClearRetained func(client *Client, topic string) error

+ Publish func(client *Client, msg *packet.Message) error

+ QueueOffline func(client *Client) error

+ QueueRetained func(client *Client, topic string) error

+ Setup func(client *Client, id string) (Session, bool, error)

+ StoreRetained func(client *Client, msg *packet.Message) error

+ Subscribe func(client *Client, topic string) error

+ Terminate func(client *Client) error

+ Unsubscribe func(client *Client, topic string) error

+ type Client struct

+ Context *Context

+ func (c *Client) CleanSession() bool

+ func (c *Client) ClientID() string

+ func (c *Client) Close(clean bool)

+ func (c *Client) Publish(msg *packet.Message) bool

+ func (c *Client) RemoteAddr() net.Addr

+ func (c *Client) Session() Session

+ type Context struct

+ func NewContext() *Context

+ func (c *Context) Get(key string) interface{}

+ func (c *Context) Set(key string, value interface{})

+ type Engine struct

+ Backend Backend

+ ConnectTimeout time.Duration

+ DefaultReadLimit int64

+ Logger Logger

+ func NewEngine() *Engine

+ func NewEngineWithBackend(backend Backend) *Engine

+ func (e *Engine) Accept(server transport.Server)

+ func (e *Engine) Clients() []*Client

+ func (e *Engine) Close()

+ func (e *Engine) Handle(conn transport.Conn) bool

+ func (e *Engine) Wait(timeout time.Duration) bool

+ type LogEvent int

+ const BackendError

+ const ClientError

+ const LostConnection

+ const MessageForwarded

+ const MessagePublished

+ const NewConnection

+ const PacketReceived

+ const PacketSent

+ const SessionError

+ const TransportError

+ type Logger func(LogEvent, *Client, packet.Packet, *packet.Message, error)

+ type MemoryBackend struct

+ Logins map[string]string

+ func NewMemoryBackend() *MemoryBackend

+ func (m *MemoryBackend) Authenticate(client *Client, user, password string) (bool, error)

+ func (m *MemoryBackend) ClearRetained(client *Client, topic string) error

+ func (m *MemoryBackend) Publish(client *Client, msg *packet.Message) error

+ func (m *MemoryBackend) QueueOffline(client *Client) error

+ func (m *MemoryBackend) QueueRetained(client *Client, topic string) error

+ func (m *MemoryBackend) Setup(client *Client, id string) (Session, bool, error)

+ func (m *MemoryBackend) StoreRetained(client *Client, msg *packet.Message) error

+ func (m *MemoryBackend) Subscribe(client *Client, topic string) error

+ func (m *MemoryBackend) Terminate(client *Client) error

+ func (m *MemoryBackend) Unsubscribe(client *Client, topic string) error

+ type MemorySession struct

+ func NewMemorySession() *MemorySession

+ func (s *MemorySession) AllPackets(direction string) ([]packet.Packet, error)

+ func (s *MemorySession) AllSubscriptions() ([]*packet.Subscription, error)

+ func (s *MemorySession) ClearWill() error

+ func (s *MemorySession) DeletePacket(direction string, id uint16) error

+ func (s *MemorySession) DeleteSubscription(topic string) error

+ func (s *MemorySession) LookupPacket(direction string, id uint16) (packet.Packet, error)

+ func (s *MemorySession) LookupSubscription(topic string) (*packet.Subscription, error)

+ func (s *MemorySession) LookupWill() (*packet.Message, error)

+ func (s *MemorySession) PacketID() uint16

+ func (s *MemorySession) Reset() error

+ func (s *MemorySession) SavePacket(direction string, pkt packet.Packet) error

+ func (s *MemorySession) SaveSubscription(sub *packet.Subscription) error

+ func (s *MemorySession) SaveWill(newWill *packet.Message) error

+ type Session interface

+ AllPackets func(direction string) ([]packet.Packet, error)

+ AllSubscriptions func() ([]*packet.Subscription, error)

+ ClearWill func() error

+ DeletePacket func(direction string, id uint16) error

+ DeleteSubscription func(topic string) error

+ LookupPacket func(direction string, id uint16) (packet.Packet, error)

+ LookupSubscription func(topic string) (*packet.Subscription, error)

+ LookupWill func() (*packet.Message, error)

+ PacketID func() uint16

+ Reset func() error

+ SavePacket func(direction string, pkt packet.Packet) error

+ SaveSubscription func(sub *packet.Subscription) error

+ SaveWill func(msg *packet.Message) error