broker

Dec 23, 2019

Changes in this version

type Broker

+ Messages *messages.Client

+ type MessagesStore interface

+ Put func(ctx context.Context, streamId string, shardKey string, payload []byte) error

Oct 12, 2019

Changes in this version

type Broker

+ Queues QueuesStore

+ type QueuesStore interface

+ Create func(ctx context.Context, id string) error

+ Delete func(ctx context.Context, id string) error

+ PutMessage func(ctx context.Context, id string, publish *packet.Publish) error

Sep 6, 2019

Changes in this version

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

Aug 29, 2019

Changes in this version

+ func EncodeSessionToken(signKey string, session *sessions.Session) (string, error)

type Broker

+ func (b *Broker) Health() string

+ func (b *Broker) SigningKey() string

type SessionStore

+ Create func(ctx context.Context, sess sessions.SessionCreateInput) error

+ RefreshKeepAlive func(ctx context.Context, id string, timestamp int64) error

+ type Token struct

+ PeerID string

+ SessionID string

+ SessionTenant string

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

Jul 12, 2019

Changes in this version

type Broker

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

Jul 11, 2019

Changes in this version

+ type Queue interface

+ Close func() error

+ Consume func(f func(*publishQueue.Message))

+ Enqueue func(p *publishQueue.Message)

Jul 8, 2019

Changes in this version

type Broker

+ func (b *Broker) JoinServiceLayer(layer cluster.ServiceLayer)

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

+ func (b *Broker) Shutdown()

Jul 7, 2019

Changes in this version

+ func Serve(port int, handler broker) net.Listener

type Broker

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

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

+ func (b *Broker) Publish(ctx context.Context, id string, p *packet.Publish) (puback *packet.PubAck, err error)

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

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

+ func (broker *Broker) Start(layer cluster.ServiceLayer)

+ type Listener interface

+ Close func(ctx context.Context) error

+ CloseSession func(ctx context.Context, id string) error

+ Publish func(ctx context.Context, id string, publish *packet.Publish) error

Jun 2, 2019

Changes in this version

type Broker

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

+ type RemoteRPCTransport struct

+ func (r *RemoteRPCTransport) Close() error

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

May 29, 2019

Changes in this version

+ type Job func() error

+ type JobWrap struct

+ type Pool struct

+ func NewPool(count int) *Pool

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

+ func (a *Pool) Cancel()

Apr 15, 2019

Changes in this version

type Config

+ RPCIdentity identity.Identity

Apr 15, 2019

Changes in this version

+ const EVENT_MESSAGE_PUBLISHED

+ const EVENT_STATE_UPDATED

type Broker

+ RPCCaller *rpc.Caller

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

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

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

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

type PeerStore

+ DumpPeers func() *peers.PeerList

type SessionStore

+ DumpSessions func() *sessions.SessionList

type SubscriptionStore

+ DumpSubscriptions func() *subscriptions.SubscriptionList

type TopicStore

+ DumpRetainedMessages func() *topics.RetainedMessageList

Feb 9, 2019

Changes in this version

type SessionStore

+ ByClientID func(id string) (sessions.SessionList, error)

Feb 8, 2019

Changes in this version

type Broker

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

Jan 13, 2019

Changes in this version

+ type STANMessage struct

+ Payload []byte

+ Tenant string

+ Timestamp time.Time

+ Topic []byte

Jan 12, 2019

Changes in this version

type Broker

+ STANOutput chan STANMessage

type Config

+ NATSURL string

Dec 31, 2018

Changes in this version

type Broker

+ WSTransport io.Closer

type Config

+ WSPort int

Dec 28, 2018

Changes in this version

type Broker

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

Dec 28, 2018

Changes in this version

type Broker

+ ID string

+ Peers PeerStore

+ type PeerStore interface

+ All func() (peers.PeerList, error)

+ ByID func(id string) (*peers.Peer, error)

+ ByMeshID func(id uint64) (*peers.Peer, error)

+ Delete func(id string) error

+ On func(event string, handler func(*peers.Peer)) func()

+ Upsert func(sess *peers.Peer) error

Nov 10, 2018

Changes in this version

type Broker

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

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

type Config

+ Session SessionConfig

+ type SessionConfig struct

+ MaxInflightSize int

Nov 8, 2018

Changes in this version

type SessionStore

+ On func(event string, handler func(*sessions.Session)) func()

type SubscriptionStore

+ On func(event string, handler func(*subscriptions.Subscription)) func()

type TopicStore

+ On func(event string, handler func(*topics.RetainedMessage)) func()

Nov 4, 2018

Changes in this version

+ type Broker struct

+ Listener io.Closer

+ Peer *peer.Peer

+ RPC io.Closer

+ Sessions SessionStore

+ Subscriptions SubscriptionStore

+ TCPTransport io.Closer

+ TLSTransport io.Closer

+ Topics TopicStore

+ WSSTransport io.Closer

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

+ func (b *Broker) Authenticate(transport listener.Transport, sessionID, username string, password string) (tenant string, id string, err 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) OnConnect(id, tenant string, ch *listener.Session, transport string)

+ func (b *Broker) OnPublish(id, tenant string, packet *packet.Publish) error

+ func (b *Broker) OnSessionClosed(id, tenant string)

+ func (b *Broker) OnSessionLost(id, tenant string)

+ func (b *Broker) OnSubscribe(id string, tenant string, packet *packet.Subscribe) error

+ func (b *Broker) OnUnsubscribe(id string, tenant string, packet *packet.Unsubscribe) error

+ func (b *Broker) Stop()

+ type Config struct

+ AuthHelper func(transport listener.Transport, sessionID, username string, password string) (tenant string, id string, err error)

+ GossipPort int

+ RPCPort int

+ TCPPort int

+ TLS *tls.Config

+ TLSPort int

+ WSSPort int

+ func DefaultConfig() Config

+ type MessagePublished struct

+ Dup bool

+ Payload []byte

+ Qos []int32

+ Recipient []string

+ Retained bool

+ Topic []byte

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

+ func (*MessagePublished) ProtoMessage()

+ 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 (m *MessagePublished) Reset()

+ func (m *MessagePublished) String() string

+ type SessionStore interface

+ All func() (sessions.SessionList, error)

+ ByID func(id string) (*sessions.Session, error)

+ ByPeer func(peer uint64) (sessions.SessionList, error)

+ Delete func(id string) error

+ Exists func(id string) bool

+ Upsert func(sess *sessions.Session) error

+ type SubscriptionStore interface

+ All func() (subscriptions.SubscriptionList, error)

+ ByID func(id string) (*subscriptions.Subscription, error)

+ ByPeer func(peer uint64) (subscriptions.SubscriptionList, error)

+ BySession func(id string) (subscriptions.SubscriptionList, error)

+ ByTopic func(tenant string, pattern []byte) (*subscriptions.SubscriptionList, error)

+ Create func(subscription *subscriptions.Subscription) error

+ Delete func(id string) error

+ Sessions func() ([]string, error)

+ type TopicStore interface

+ All func() (topics.RetainedMessageList, error)

+ ByTopicPattern func(tenant string, pattern []byte) (topics.RetainedMessageList, error)

+ Create func(message *topics.RetainedMessage) error