messaging

Documentation

Index

• type Channel
• type HandlerFunc
• type PubSub
  • func NewPubSub(pool *jack.Pool, logger *ll.Logger, hooks *pipeline.Hooks) *PubSub
  • func (ps *PubSub) Count(bucket, channel string) int
  • func (ps *PubSub) IDs(bucket, channel string) []int64
  • func (ps *PubSub) Publish(bucket, channel string, payload []byte) (int, error)
  • func (ps *PubSub) Subscribe(bucket, channel string, fn func([]byte)) (int64, func())
  • func (ps *PubSub) SubscribeSimple(bucket, channel string, fn func([]byte)) func()
  • func (ps *PubSub) UnsubscribeAll(bucket, channel string)
• type ReqResp
  • func NewReqResp(timeout time.Duration) *ReqResp
  • func (r *ReqResp) Call(name string, payload []byte) ([]byte, error)
  • func (r *ReqResp) CallDirect(name string, payload []byte) ([]byte, error)
  • func (r *ReqResp) Handlers() []string
  • func (r *ReqResp) Has(name string) bool
  • func (r *ReqResp) Register(name string, fn HandlerFunc)
  • func (r *ReqResp) Unregister(name string)

Types

type Channel

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

Channel holds subscribers for a specific bucket+channel

type HandlerFunc

type HandlerFunc func(req []byte) ([]byte, error)

HandlerFunc is the signature every REQ/RESP handler must satisfy.

type PubSub

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

PubSub is a high-performance publish-subscribe system

func NewPubSub

func NewPubSub(pool *jack.Pool, logger *ll.Logger, hooks *pipeline.Hooks) *PubSub

NewPubSub creates a new PubSub instance

func (*PubSub) Count

func (ps *PubSub) Count(bucket, channel string) int

Count returns the number of subscribers for a channel

func (*PubSub) IDs

func (ps *PubSub) IDs(bucket, channel string) []int64

IDs returns all subscriber IDs for a channel

func (*PubSub) Publish

func (ps *PubSub) Publish(bucket, channel string, payload []byte) (int, error)

Publish sends a payload to all subscribers of a channel

func (*PubSub) Subscribe

func (ps *PubSub) Subscribe(bucket, channel string, fn func([]byte)) (int64, func())

Subscribe adds a subscriber and returns its ID and an unsubscribe function

func (*PubSub) SubscribeSimple

func (ps *PubSub) SubscribeSimple(bucket, channel string, fn func([]byte)) func()

SubscribeSimple adds a subscriber and returns only the unsubscribe function

func (*PubSub) UnsubscribeAll

func (ps *PubSub) UnsubscribeAll(bucket, channel string)

UnsubscribeAll removes all subscribers from a channel

type ReqResp

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

ReqResp manages named request/response handlers using a sharded map.

func NewReqResp

func NewReqResp(timeout time.Duration) *ReqResp

NewReqResp creates a new ReqResp instance.

func (*ReqResp) Call

func (r *ReqResp) Call(name string, payload []byte) ([]byte, error)

Call invokes the named handler with timeout protection. Optimized to use sync.Once for the goroutine to reduce allocations.

func (*ReqResp) CallDirect

func (r *ReqResp) CallDirect(name string, payload []byte) ([]byte, error)

CallDirect invokes the handler in the current goroutine.

func (*ReqResp) Handlers

func (r *ReqResp) Handlers() []string

Handlers returns a snapshot of all registered handler names.

func (*ReqResp) Has

func (r *ReqResp) Has(name string) bool

Has returns true if a handler exists.

func (*ReqResp) Register

func (r *ReqResp) Register(name string, fn HandlerFunc)

Register binds name → handler.

func (*ReqResp) Unregister

func (r *ReqResp) Unregister(name string)

Unregister removes the named handler.