Versions in this module Expand all Collapse all v0 v0.1.0 Jun 14, 2022 Changes in this version + var ErrStreamNotFound error = errors.New("stream not found") + func PrintSampleConfig() + type Config struct + DSN string + GCIntervalInSec int + GCKeepItems int + MaxBatchSize int + PollIntervalInMs int + func DefaultConfig() *Config + func LoadConfig(path string) (*Config, error) + func MustLoadConfig(path string) *Config + func (c *Config) String() string + type Hub struct + func NewHub(c *Config) (*Hub, error) + func (m *Hub) DB() *sql.DB + func (m *Hub) ForceGC(streamName string) error + func (m *Hub) GetStreamNames() ([]string, error) + func (m *Hub) MessagesSinceOffset(streamName string, offset Offset) ([]Message, error) + func (m *Hub) MinMaxID(streamName string) (int64, int64, error) + func (m *Hub) PollStat(streamName string) map[string]interface{} + func (m *Hub) Publish(streamName string, msg *Message) error + func (m *Hub) Subscribe(streamName string, subscriberID string) (<-chan Message, error) + func (m *Hub) Unsubscribe(streamName string, subscriberID string) + type Message struct + Data string + ID int64 + Ts int64 + func (m Message) String() string + type Offset int64 + var LatestId Offset = -1 + func (o Offset) String() string + type PollWorker struct + func (pw *PollWorker) Stat() map[string]interface{} + func (pw *PollWorker) Stop() + type Stat struct + func GetStat() *Stat + func (s *Stat) Inc(key string, val int) + func (s *Stat) JSON() (string, error) + func (s *Stat) Set(key string, val interface{}) + type Store interface + CreateStream func(streamName string) error + DB func() *sql.DB + FetchMessages func(streamName string, offset Offset, limit int) ([]Message, Offset, error) + GetStreamNames func() ([]string, error) + Init func() error + MinMaxID func(streamName string) (int64, int64, error) + PutMessages func(streamName string, messages []*Message) error + func OpenStore(dsn string) (Store, error) + type Stream struct + func NewStream(cfg *Config, s Store, name string) (*Stream, error) + func (s *Stream) MinMaxID() (int64, int64, error) + func (s *Stream) Name() string + func (s *Stream) Open() error + func (s *Stream) Publish(m *Message) + type TiDBStore struct + func NewTiDBStore(dsn string) *TiDBStore + func (s *TiDBStore) CreateStream(streamName string) error + func (s *TiDBStore) DB() *sql.DB + func (s *TiDBStore) FetchMessages(streamName string, idOffset Offset, limit int) ([]Message, Offset, error) + func (s *TiDBStore) GetStreamNames() ([]string, error) + func (s *TiDBStore) Init() error + func (s *TiDBStore) MinMaxID(streamName string) (int64, int64, error) + func (s *TiDBStore) PutMessages(streamName string, messages []*Message) error
Go to main page