broker

Documentation

Index

• Constants
• type Broker
  • func New(id identity.Identity, config Config) *Broker
  • func (b *Broker) Authenticate(transport transport.Metadata, sessionID []byte, username string, ...) (tenant string, id string, err error)
  • func (b *Broker) CloseSession(id string) error
  • func (b *Broker) DistributeMessage(message *rpc.MessagePublished) error
  • func (b *Broker) Join(hosts []string)
  • func (b *Broker) ListRetainedMessages() (topics.RetainedMessageSet, error)
  • func (b *Broker) ListSessions() (sessions.SessionSet, error)
  • func (b *Broker) ListSubscriptions() (subscriptions.SubscriptionSet, error)
  • func (b *Broker) NotifyJoin(n *memberlist.Node)
  • func (b *Broker) NotifyLeave(n *memberlist.Node)
  • func (b *Broker) NotifyUpdate(n *memberlist.Node)
  • func (b *Broker) OnConnect(transportSession *listener.Session) (int32, error)
  • func (b *Broker) OnPublish(sess sessions.Session, p *packet.Publish) error
  • func (b *Broker) OnSessionClosed(sess sessions.Session)
  • func (b *Broker) OnSessionLost(sess sessions.Session)
  • func (b *Broker) OnSubscribe(transportSession *listener.Session, sess sessions.Session, p *packet.Subscribe) error
  • func (b *Broker) OnUnsubscribe(sess sessions.Session, packet *packet.Unsubscribe) error
  • func (b *Broker) RemoteRPCProvider(addr, session string) sessions.Transport
  • func (b *Broker) RetainThenDispatchToLocalSessions(tenant string, topic []byte, payload []byte, qos int32)
  • func (b *Broker) Stop()
• type Config
  • func DefaultConfig() Config
• type Job
• type JobWrap
• type PeerStore
• type Pool
  • func NewPool(count int) *Pool
  • func (a *Pool) Call(job Job) error
  • func (a *Pool) Cancel()
• type RemoteRPCTransport
  • func (r *RemoteRPCTransport) Close() error
  • func (r *RemoteRPCTransport) Publish(publish *packet.Publish) error
• type STANMessage
• type SessionConfig
• type SessionStore
• type SubscriptionStore
• type TopicStore

Constants

const (
	EVENT_MESSAGE_PUBLISHED = "message_published"
	EVENT_STATE_UPDATED     = "state_updated"
)

Types

type Broker

type Broker struct {
	ID string

	Subscriptions SubscriptionStore
	Sessions      SessionStore
	Topics        TopicStore
	Peers         PeerStore
	STANOutput    chan STANMessage
	Listener      io.Closer
	TCPTransport  io.Closer
	TLSTransport  io.Closer
	WSSTransport  io.Closer
	WSTransport   io.Closer
	RPC           net.Listener
	RPCCaller     *rpc.Caller
	// contains filtered or unexported fields
}

func New

func New(id identity.Identity, config Config) *Broker

func (*Broker) Authenticate

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

func (*Broker) CloseSession

func (b *Broker) CloseSession(id string) error

func (*Broker) DistributeMessage

func (b *Broker) DistributeMessage(message *rpc.MessagePublished) error

func (*Broker) Join

func (b *Broker) Join(hosts []string)

func (*Broker) ListRetainedMessages

func (b *Broker) ListRetainedMessages() (topics.RetainedMessageSet, error)

func (*Broker) ListSessions

func (b *Broker) ListSessions() (sessions.SessionSet, error)

func (*Broker) ListSubscriptions

func (b *Broker) ListSubscriptions() (subscriptions.SubscriptionSet, error)

func (*Broker) NotifyJoin

func (b *Broker) NotifyJoin(n *memberlist.Node)

NotifyJoin is called if a peer joins the cluster.

func (*Broker) NotifyLeave

func (b *Broker) NotifyLeave(n *memberlist.Node)

NotifyLeave is called if a peer leaves the cluster.

func (*Broker) NotifyUpdate

func (b *Broker) NotifyUpdate(n *memberlist.Node)

NotifyUpdate is called if a cluster peer gets updated.

func (*Broker) OnConnect

func (b *Broker) OnConnect(transportSession *listener.Session) (int32, error)

func (*Broker) OnPublish

func (b *Broker) OnPublish(sess sessions.Session, p *packet.Publish) error

func (*Broker) OnSessionClosed

func (b *Broker) OnSessionClosed(sess sessions.Session)

func (*Broker) OnSessionLost

func (b *Broker) OnSessionLost(sess sessions.Session)

func (*Broker) OnSubscribe

func (b *Broker) OnSubscribe(transportSession *listener.Session, sess sessions.Session, p *packet.Subscribe) error

func (*Broker) OnUnsubscribe

func (b *Broker) OnUnsubscribe(sess sessions.Session, packet *packet.Unsubscribe) error

func (*Broker) RemoteRPCProvider

func (b *Broker) RemoteRPCProvider(addr, session string) sessions.Transport

func (*Broker) RetainThenDispatchToLocalSessions

func (b *Broker) RetainThenDispatchToLocalSessions(tenant string, topic []byte, payload []byte, qos int32)

func (*Broker) Stop

func (b *Broker) Stop()

type Config

type Config struct {
	TCPPort     int
	TLS         *tls.Config
	TLSPort     int
	WSSPort     int
	WSPort      int
	RPCPort     int
	RPCIdentity identity.Identity
	GossipPort  int
	NATSURL     string
	AuthHelper  func(transport transport.Metadata, sessionID []byte, username string, password string) (tenant string, err error)
	Session     SessionConfig
}

func DefaultConfig

func DefaultConfig() Config

type Job

type Job func() error

type JobWrap

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

type PeerStore

type PeerStore interface {
	ByID(id string) (peers.Peer, error)
	All() (peers.SubscriptionSet, error)
	Upsert(sess peers.Peer) error
	Delete(id string) error
	On(event string, handler func(peers.Peer)) func()
}

type Pool

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

func NewPool

func NewPool(count int) *Pool

func (*Pool) Call

func (a *Pool) Call(job Job) error

func (*Pool) Cancel

func (a *Pool) Cancel()

type RemoteRPCTransport

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

func (*RemoteRPCTransport) Close

func (r *RemoteRPCTransport) Close() error

func (*RemoteRPCTransport) Publish

func (r *RemoteRPCTransport) Publish(publish *packet.Publish) error

type STANMessage

type STANMessage struct {
	Timestamp time.Time `json:"timestamp"`
	Tenant    string    `json:"tenant"`
	Payload   []byte    `json:"payload"`
	Topic     []byte    `json:"topic"`
}

type SessionConfig

type SessionConfig struct {
	MaxInflightSize int
}

type SessionStore

type SessionStore interface {
	ByID(id string) (sessions.Session, error)
	ByClientID(id string) (sessions.SessionSet, error)
	ByPeer(peer string) (sessions.SessionSet, error)
	All() (sessions.SessionSet, error)
	Exists(id string) bool
	Upsert(sess sessions.Session, transport sessions.Transport) error
	Delete(id, reason string) error
	On(event string, handler func(sessions.Session)) func()
}

type SubscriptionStore

type SubscriptionStore interface {
	ByTopic(tenant string, pattern []byte) (subscriptions.SubscriptionSet, error)
	ByID(id string) (subscriptions.Subscription, error)
	All() (subscriptions.SubscriptionSet, error)
	ByPeer(peer string) (subscriptions.SubscriptionSet, error)
	BySession(id string) (subscriptions.SubscriptionSet, error)
	Sessions() ([]string, error)
	Create(message subscriptions.Subscription, sender func(packet.Publish) error) error
	Delete(id string) error
	On(event string, handler func(subscriptions.Subscription)) func()
}

type TopicStore

type TopicStore interface {
	Create(message topics.RetainedMessage) error
	ByTopicPattern(tenant string, pattern []byte) (topics.RetainedMessageSet, error)
	All() (topics.RetainedMessageSet, error)
	On(event string, handler func(topics.RetainedMessage)) func()
}