broker

Documentation

Index

• Variables
• func EncodeSessionToken(signKey string, tenant string, id string) (string, error)
• type Broker
  • func New(id string, logger *zap.Logger, mesh cluster.DiscoveryLayer, config Config) *Broker
  • func (b *Broker) Authenticate(transport transport.Metadata, sessionID []byte, username string, ...) (tenant string, err error)
  • func (b *Broker) CloseSession(ctx context.Context, token string) error
  • func (b *Broker) Connect(ctx context.Context, metadata transport.Metadata, p *packet.Connect) (string, string, *packet.ConnAck, error)
  • func (b *Broker) Disconnect(ctx context.Context, token string, p *packet.Disconnect) error
  • func (b *Broker) Health() string
  • func (b *Broker) JoinServiceLayer(name string, logger *zap.Logger, config cluster.ServiceConfig, ...)
  • func (b *Broker) PingReq(ctx context.Context, id string, _ *packet.PingReq) (*packet.PingResp, error)
  • func (b *Broker) Publish(ctx context.Context, token string, p *packet.Publish) (*packet.PubAck, error)
  • func (b *Broker) Serve(port int) net.Listener
  • func (b *Broker) Shutdown()
  • func (b *Broker) SigningKey() string
  • func (b *Broker) Subscribe(ctx context.Context, token string, p *packet.Subscribe) (*packet.SubAck, error)
  • func (b *Broker) Unsubscribe(ctx context.Context, token string, p *packet.Unsubscribe) (*packet.UnsubAck, error)
• type Config
  • func DefaultConfig() Config
• type MessagesStore
• type QueuesStore
• type SessionConfig
• type SessionStore
• type Token
  • func DecodeSessionToken(signKey string, signedToken string) (Token, error)
• type TopicStore

Variables

var (
	ErrAuthenticationFailed = errors.New("authentication failed")
)

Functions

func EncodeSessionToken

func EncodeSessionToken(signKey string, tenant string, id string) (string, error)

Types

type Broker

type Broker struct {
	ID string

	Sessions SessionStore
	Queues   QueuesStore
	Messages *messages.Client
	// contains filtered or unexported fields
}

func New

func New(id string, logger *zap.Logger, mesh cluster.DiscoveryLayer, config Config) *Broker

func (*Broker) Authenticate

func (b *Broker) Authenticate(transport transport.Metadata, sessionID []byte, username string, password string) (tenant string, err error)

func (*Broker) CloseSession

func (b *Broker) CloseSession(ctx context.Context, token string) error

func (*Broker) Connect

func (b *Broker) Connect(ctx context.Context, metadata transport.Metadata, p *packet.Connect) (string, string, *packet.ConnAck, error)

func (*Broker) Disconnect

func (b *Broker) Disconnect(ctx context.Context, token string, p *packet.Disconnect) error

func (*Broker) Health

func (b *Broker) Health() string

func (*Broker) JoinServiceLayer

func (b *Broker) JoinServiceLayer(name string, logger *zap.Logger, config cluster.ServiceConfig, rpcConfig cluster.ServiceConfig, mesh cluster.DiscoveryLayer)

func (*Broker) PingReq

func (b *Broker) PingReq(ctx context.Context, id string, _ *packet.PingReq) (*packet.PingResp, error)

func (*Broker) Publish

func (b *Broker) Publish(ctx context.Context, token string, p *packet.Publish) (*packet.PubAck, error)

func (*Broker) Serve

func (b *Broker) Serve(port int) net.Listener

func (*Broker) Shutdown

func (b *Broker) Shutdown()

func (*Broker) SigningKey

func (b *Broker) SigningKey() string

func (*Broker) Subscribe

func (b *Broker) Subscribe(ctx context.Context, token string, p *packet.Subscribe) (*packet.SubAck, error)

func (*Broker) Unsubscribe

func (b *Broker) Unsubscribe(ctx context.Context, token string, p *packet.Unsubscribe) (*packet.UnsubAck, error)

type Config

type Config struct {
	AuthHelper func(transport transport.Metadata, sessionID []byte, username string, password string) (tenant string, err error)
	Session    SessionConfig
}

func DefaultConfig

func DefaultConfig() Config

type MessagesStore

type MessagesStore interface {
	Put(ctx context.Context, streamId string, shardKey string, payload []byte) error
}

type QueuesStore

type QueuesStore interface {
	Create(ctx context.Context, id string) error
	Delete(ctx context.Context, id string) error
}

type SessionConfig

type SessionConfig struct {
	MaxInflightSize int
}

type SessionStore

type SessionStore interface {
	RefreshKeepAlive(ctx context.Context, id string, timestamp int64) error
}

type Token

type Token struct {
	SessionID     string `json:"session_id"`
	SessionTenant string `json:"session_tenant"`
	jwt.StandardClaims
}

func DecodeSessionToken

func DecodeSessionToken(signKey string, signedToken string) (Token, error)

type TopicStore

type TopicStore interface {
	ByTopicPattern(ctx context.Context, tenant string, pattern []byte) ([]*topics.RetainedMessage, error)
}