wasp

Documentation

Index

• Variables
• func AddFields(ctx context.Context, fields ...zap.Field) context.Context
• func L(ctx context.Context) *zap.Logger
• func ProcessPublish(ctx context.Context, id uint64, transport Membership, fsm FSM, state ReadState, ...) error
• func RunSession(ctx context.Context, peer uint64, fsm FSM, state ReadState, ...) error
• func StoreLogger(ctx context.Context, l *zap.Logger) context.Context
• func StorePublish(messageLog MessageLog, p []*packet.Publish) error
• type AuthenticationHandler
• type FSM
• type Membership
• type MessageLog
• type ReadState
• type State
  • func NewState() State
• type StateDump

Variables

var (
	Clock                    = time.Now
	ErrConnectNotDone        = errors.New("CONNECT not done")
	ErrSessionLost           = errors.New("Session lost")
	ErrSessionDisconnected   = errors.New("Session disconnected")
	ErrUnknownPacketReceived = errors.New("Received unknown packet type")
	ErrAuthenticationFailed  = errors.New("Authentication failed")
)

Functions

func AddFields

func AddFields(ctx context.Context, fields ...zap.Field) context.Context

func L

func L(ctx context.Context) *zap.Logger

func ProcessPublish

func ProcessPublish(ctx context.Context, id uint64, transport Membership, fsm FSM, state ReadState, local bool, p *packet.Publish) error

func RunSession

func RunSession(ctx context.Context, peer uint64, fsm FSM, state ReadState, c transport.TimeoutReadWriteCloser, ch chan *packet.Publish, authHandler AuthenticationHandler) error

func StoreLogger

func StoreLogger(ctx context.Context, l *zap.Logger) context.Context

func StorePublish

func StorePublish(messageLog MessageLog, p []*packet.Publish) error

Types

type AuthenticationHandler

type AuthenticationHandler func(ctx context.Context, mqtt auth.ApplicationContext) (mountpoint string, err error)

type FSM

type FSM interface {
	RetainedMessage(ctx context.Context, publish *packet.Publish) error
	DeleteRetainedMessage(ctx context.Context, topic []byte) error
	Subscribe(ctx context.Context, id string, pattern []byte, qos int32) error
	Unsubscribe(ctx context.Context, id string, pattern []byte) error
	DeleteSessionMetadata(ctx context.Context, id, mountpoint string) error
	CreateSessionMetadata(ctx context.Context, id, clientID string, lwt *packet.Publish, mountpoint string) error
}

type Membership

type Membership interface {
	Call(id uint64, f func(*grpc.ClientConn) error) error
}

type MessageLog

type MessageLog interface {
	io.Closer
	Append(b []*packet.Publish) error
}

type ReadState

type ReadState interface {
	Recipients(topic []byte) ([]uint64, []string, []int32, error)
	ListSubscriptions() ([][]byte, []uint64, []string, []int32, error)
	GetSession(id string) *sessions.Session
	SaveSession(id string, session *sessions.Session)
	CloseSession(id string)
	RetainedMessages(topic []byte) ([]*packet.Publish, error)
	ListSessionMetadatas() []*api.SessionMetadatas
	GetSessionMetadatas(id string) *api.SessionMetadatas
	GetSessionMetadatasByClientID(id string) *api.SessionMetadatas
}

type State

type State interface {
	ReadState
	Subscribe(peer uint64, id string, pattern []byte, qos int32) error
	Unsubscribe(id string, pattern []byte) error
	RemoveSubscriptionsForPeer(peer uint64)
	RemoveSubscriptionsForSession(id string)
	RetainMessage(msg *packet.Publish) error
	DeleteRetainedMessage(topic []byte) error
	Load([]byte) error
	MarshalBinary() ([]byte, error)
	CreateSessionMetadata(id string, peer uint64, clientID string, connectedAt int64, lwt *packet.Publish, mountpoint string) error
	DeleteSessionMetadata(id string, peer uint64) error
	DeleteSessionMetadatasByPeer(peer uint64)
}

func NewState

func NewState() State

type StateDump

type StateDump struct {
	Subscriptions    []byte
	Topics           []byte
	SessionMetadatas []byte
}