broker

Documentation

Overview

Package broker is a generated protocol buffer package.

It is generated from these files:

events.proto

It has these top-level messages:

MessagePublished

Index

• type Broker
  • func New(id identity.Identity, config Config) *Broker
  • func (b *Broker) Authenticate(transport listener.Transport, sessionID []byte, username string, ...) (tenant string, id string, err error)
  • func (b *Broker) CloseSession(id string) error
  • func (b *Broker) Join(hosts []string)
  • func (b *Broker) ListRetainedMessages() (topics.RetainedMessageList, error)
  • func (b *Broker) ListSessions() (sessions.SessionList, error)
  • func (b *Broker) ListSubscriptions() (subscriptions.SubscriptionList, error)
  • func (b *Broker) OnBrokerStopped(f func()) func()
  • func (b *Broker) OnConnect(transportSession *listener.Session) (int32, error)
  • func (b *Broker) OnMessagePublished(recipient string, f func(p *packet.Publish)) func()
  • func (b *Broker) OnPublish(sess sessions.Session, packet *packet.Publish) error
  • func (b *Broker) OnSessionClosed(sess sessions.Session)
  • func (b *Broker) OnSessionLost(sess sessions.Session)
  • func (b *Broker) OnSubscribe(sess sessions.Session, packet *packet.Subscribe) error
  • func (b *Broker) OnUnsubscribe(sess sessions.Session, packet *packet.Unsubscribe) error
  • func (b *Broker) RetainThenDispatchToLocalSessions(tenant string, topic []byte, payload []byte, qos int32)
  • func (b *Broker) Stop()
• type Config
  • func DefaultConfig() Config
• type MessagePublished
  • func (*MessagePublished) Descriptor() ([]byte, []int)
  • func (m *MessagePublished) GetDup() bool
  • func (m *MessagePublished) GetPayload() []byte
  • func (m *MessagePublished) GetQos() []int32
  • func (m *MessagePublished) GetRecipient() []string
  • func (m *MessagePublished) GetRetained() bool
  • func (m *MessagePublished) GetTopic() []byte
  • func (*MessagePublished) ProtoMessage()
  • func (m *MessagePublished) Reset()
  • func (m *MessagePublished) String() string
• type PeerStore
• type STANMessage
• type SessionConfig
• type SessionStore
• type SubscriptionStore
• type TopicStore

Types

type Broker

type Broker struct {
	ID string

	Peer          *peer.Peer
	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          io.Closer
	// contains filtered or unexported fields
}

func New

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

func (*Broker) Authenticate

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

func (*Broker) CloseSession

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

func (*Broker) Join

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

func (*Broker) ListRetainedMessages

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

func (*Broker) ListSessions

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

func (*Broker) ListSubscriptions

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

func (*Broker) OnBrokerStopped

func (b *Broker) OnBrokerStopped(f func()) func()

func (*Broker) OnConnect

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

func (*Broker) OnMessagePublished

func (b *Broker) OnMessagePublished(recipient string, f func(p *packet.Publish)) func()

func (*Broker) OnPublish

func (b *Broker) OnPublish(sess sessions.Session, packet *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(sess sessions.Session, packet *packet.Subscribe) error

func (*Broker) OnUnsubscribe

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

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
	GossipPort int
	NATSURL    string
	AuthHelper func(transport listener.Transport, sessionID []byte, username string, password string) (tenant string, err error)
	Session    SessionConfig
}

func DefaultConfig

func DefaultConfig() Config

type MessagePublished

type MessagePublished struct {
	Recipient []string `protobuf:"bytes,1,rep,name=Recipient" json:"Recipient,omitempty"`
	Topic     []byte   `protobuf:"bytes,2,opt,name=Topic,proto3" json:"Topic,omitempty"`
	Qos       []int32  `protobuf:"varint,3,rep,packed,name=Qos" json:"Qos,omitempty"`
	Payload   []byte   `protobuf:"bytes,4,opt,name=Payload,proto3" json:"Payload,omitempty"`
	Retained  bool     `protobuf:"varint,5,opt,name=Retained" json:"Retained,omitempty"`
	Dup       bool     `protobuf:"varint,6,opt,name=Dup" json:"Dup,omitempty"`
}

func (*MessagePublished) Descriptor

func (*MessagePublished) Descriptor() ([]byte, []int)

func (*MessagePublished) GetDup

func (m *MessagePublished) GetDup() bool

func (*MessagePublished) GetPayload

func (m *MessagePublished) GetPayload() []byte

func (*MessagePublished) GetQos

func (m *MessagePublished) GetQos() []int32

func (*MessagePublished) GetRecipient

func (m *MessagePublished) GetRecipient() []string

func (*MessagePublished) GetRetained

func (m *MessagePublished) GetRetained() bool

func (*MessagePublished) GetTopic

func (m *MessagePublished) GetTopic() []byte

func (*MessagePublished) ProtoMessage

func (*MessagePublished) ProtoMessage()

func (*MessagePublished) Reset

func (m *MessagePublished) Reset()

func (*MessagePublished) String

func (m *MessagePublished) String() string

type PeerStore

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

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)
	ByPeer(peer uint64) (sessions.SessionList, error)
	ByClientID(id string) (sessions.SessionList, error)
	All() (sessions.SessionList, error)
	Exists(id string) bool
	Upsert(sess sessions.Session) 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.SubscriptionList, error)
	ByID(id string) (*subscriptions.Subscription, error)
	All() (subscriptions.SubscriptionList, error)
	ByPeer(peer uint64) (subscriptions.SubscriptionList, error)
	BySession(id string) (subscriptions.SubscriptionList, error)
	Sessions() ([]string, error)
	Create(subscription *subscriptions.Subscription) 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.RetainedMessageList, error)
	All() (topics.RetainedMessageList, error)
	On(event string, handler func(*topics.RetainedMessage)) func()
}