msgbus

Apr 23, 2023

Changes in this version

type Message

+ func (m Message) String() string

Apr 4, 2022

Changes in this version

+ func GenerateULID() (string, error)

+ func MustGenerateULID() string

Apr 4, 2022

Changes in this version

+ const DefaultBind

+ const DefaultLogPath

+ const DefaultMetrics

+ const DefaultNoSync

+ var ErrBufferFull = errors.New("error: subscriber buffer full")

+ func SafeParseInt64(s string, d int64) int64

type Message

+ func LoadMessage(data []byte) (m Message, err error)

+ func (m Message) Bytes() ([]byte, error)

+ type Option func(opts *Options) error

+ func WithBufferLength(bufferLength int) Option

+ func WithLogPath(logPath string) Option

+ func WithMaxPayloadSize(maxPayloadSize int) Option

+ func WithMaxQueueSize(maxQueueSize int) Option

+ func WithMetrics(metrics bool) Option

+ func WithNoSync(noSync bool) Option

type Options

+ LogPath string

+ Metrics bool

+ NoSync bool

+ func NewDefaultOptions() *Options

Apr 2, 2022

Changes in this version

+ func SafeParseInt(s string, d int) int

type Queue

+ func (q *Queue) ForEach(f func(elem interface{}) error) error

+ type SubscribeOption func(*SubscriberOptions)

+ func WithIndex(index int) SubscribeOption

+ type SubscriberConfig struct

+ BufferLength int

+ type SubscriberOptions struct

+ Index int

+ type Subscribers struct

+ func NewSubscribers(config *SubscriberConfig) *Subscribers

+ func (subs *Subscribers) AddSubscriber(id string) chan Message

+ func (subs *Subscribers) GetSubscriber(id string) (chan Message, bool)

+ func (subs *Subscribers) HasSubscriber(id string) bool

+ func (subs *Subscribers) Len() int

+ func (subs *Subscribers) NotifyAll(message Message) int

+ func (subs *Subscribers) RemoveSubscriber(id string)

Mar 20, 2022

Changes in this version

+ var Commit = "HEAD"

Mar 16, 2022

Changes in this version

+ const DefaultBufferLength

+ const DefaultMaxPayloadSize

+ const DefaultMaxQueueSize

+ type ListenerOptions struct

+ BufferLength int

type MessageBus

+ func New(options *Options) *MessageBus

type Metrics

+ func (m *Metrics) CounterVec(subsystem, name string) *prometheus.CounterVec

+ func (m *Metrics) NewCounterVec(subsystem, name, help string, labels []string) *prometheus.CounterVec

type Options

+ BufferLength int

+ MaxPayloadSize int

+ MaxQueueSize int

type Queue

+ func NewQueue(maxlen ...int) *Queue

+ func (q *Queue) Empty() bool

+ func (q *Queue) Full() bool

+ func (q *Queue) MaxLen() int

+ func (q *Queue) Size() int

type Topic

+ func (t *Topic) String() string

May 8, 2018

Changes in this version

+ const DefaultTTL

+ var Build = "dev"

+ var DefObjectives = map[float64]float64

+ var GitCommit = "HEAD"

+ var Version = "0.1.0"

+ func FullVersion() string

+ type Client struct

+ func NewClient(conn *websocket.Conn, topic *Topic, bus *MessageBus) *Client

+ func (c *Client) Start()

+ type HandlerFunc func(msg *Message) error

+ type Listeners struct

+ func NewListeners() *Listeners

+ func (ls *Listeners) Add(id string) chan Message

+ func (ls *Listeners) Exists(id string) bool

+ func (ls *Listeners) Get(id string) (chan Message, bool)

+ func (ls *Listeners) Length() int

+ func (ls *Listeners) NotifyAll(message Message) int

+ func (ls *Listeners) Remove(id string)

+ type Message struct

+ Created time.Time

+ ID uint64

+ Payload []byte

+ Topic *Topic

+ type MessageBus struct

+ func NewMessageBus(options *Options) *MessageBus

+ func (mb *MessageBus) Get(t *Topic) (Message, bool)

+ func (mb *MessageBus) Len() int

+ func (mb *MessageBus) Metrics() *Metrics

+ func (mb *MessageBus) NewMessage(topic *Topic, payload []byte) Message

+ func (mb *MessageBus) NewTopic(topic string) *Topic

+ func (mb *MessageBus) NotifyAll(message Message)

+ func (mb *MessageBus) Put(message Message)

+ func (mb *MessageBus) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ func (mb *MessageBus) Subscribe(id, topic string) chan Message

+ func (mb *MessageBus) Unsubscribe(id, topic string)

+ type Metrics struct

+ func NewMetrics(namespace string) *Metrics

+ func (m *Metrics) Counter(subsystem, name string) prometheus.Counter

+ func (m *Metrics) Gauge(subsystem, name string) prometheus.Gauge

+ func (m *Metrics) GaugeVec(subsystem, name string) *prometheus.GaugeVec

+ func (m *Metrics) Handler() http.Handler

+ func (m *Metrics) NewCounter(subsystem, name, help string) prometheus.Counter

+ func (m *Metrics) NewCounterFunc(subsystem, name, help string, f func() float64) prometheus.CounterFunc

+ func (m *Metrics) NewGauge(subsystem, name, help string) prometheus.Gauge

+ func (m *Metrics) NewGaugeFunc(subsystem, name, help string, f func() float64) prometheus.GaugeFunc

+ func (m *Metrics) NewGaugeVec(subsystem, name, help string, labels []string) *prometheus.GaugeVec

+ func (m *Metrics) NewSummary(subsystem, name, help string) prometheus.Summary

+ func (m *Metrics) NewSummaryVec(subsystem, name, help string, labels []string) *prometheus.SummaryVec

+ func (m *Metrics) Run(addr string)

+ func (m *Metrics) Summary(subsystem, name string) prometheus.Summary

+ func (m *Metrics) SummaryVec(subsystem, name string) *prometheus.SummaryVec

+ type Options struct

+ DefaultTTL time.Duration

+ WithMetrics bool

+ type Queue struct

+ func (q *Queue) Len() int

+ func (q *Queue) Peek() interface{}

+ func (q *Queue) Pop() interface{}

+ func (q *Queue) Push(item interface{})

+ type QueueNode struct

+ type Topic struct

+ Created time.Time

+ Name string

+ Sequence uint64

+ TTL time.Duration