Documentation
¶
Index ¶
- Constants
- Variables
- type Listener
- type ListenerWithErr
- type MemoryPubsub
- func (*MemoryPubsub) Close() error
- func (m *MemoryPubsub) Publish(event string, message []byte) error
- func (m *MemoryPubsub) Subscribe(event string, listener Listener) (cancel func(), err error)
- func (m *MemoryPubsub) SubscribeWithErr(event string, listener ListenerWithErr) (cancel func(), err error)
- type PGPubsub
- func (p *PGPubsub) Close() error
- func (p *PGPubsub) Collect(metrics chan<- prometheus.Metric)
- func (p *PGPubsub) Describe(descs chan<- *prometheus.Desc)
- func (p *PGPubsub) Publish(event string, message []byte) error
- func (p *PGPubsub) Subscribe(event string, listener Listener) (cancel func(), err error)
- func (p *PGPubsub) SubscribeWithErr(event string, listener ListenerWithErr) (cancel func(), err error)
- type Pubsub
- type Watchdog
Constants ¶
const BufferSize = 2048
BufferSize is the maximum number of unhandled messages we will buffer for a subscriber before dropping messages.
const (
EventPubsubWatchdog = "pubsub_watchdog"
)
Variables ¶
var ErrDroppedMessages = xerrors.New("dropped messages")
ErrDroppedMessages is sent to ListenerWithErr if messages are dropped or might have been dropped.
Functions ¶
This section is empty.
Types ¶
type ListenerWithErr ¶
ListenerWithErr represents a pubsub handler that can also receive error indications
type MemoryPubsub ¶
type MemoryPubsub struct {
// contains filtered or unexported fields
}
MemoryPubsub is an in-memory Pubsub implementation. It's an exported type so that our test code can do type checks.
func (*MemoryPubsub) Close ¶
func (*MemoryPubsub) Close() error
func (*MemoryPubsub) Subscribe ¶
func (m *MemoryPubsub) Subscribe(event string, listener Listener) (cancel func(), err error)
func (*MemoryPubsub) SubscribeWithErr ¶
func (m *MemoryPubsub) SubscribeWithErr(event string, listener ListenerWithErr) (cancel func(), err error)
type PGPubsub ¶ added in v2.8.0
type PGPubsub struct {
// contains filtered or unexported fields
}
PGPubsub is a pubsub implementation using PostgreSQL.
func New ¶
func New(startCtx context.Context, logger slog.Logger, database *sql.DB, connectURL string) (*PGPubsub, error)
New creates a new Pubsub implementation using a PostgreSQL connection.
func (*PGPubsub) Collect ¶ added in v2.8.0
func (p *PGPubsub) Collect(metrics chan<- prometheus.Metric)
Collect implements, along with Describe, the prometheus.Collector interface for metrics
func (*PGPubsub) Describe ¶ added in v2.8.0
func (p *PGPubsub) Describe(descs chan<- *prometheus.Desc)
Describe implements, along with Collect, the prometheus.Collector interface for metrics.
func (*PGPubsub) Subscribe ¶ added in v2.8.0
Subscribe calls the listener when an event matching the name is received.
func (*PGPubsub) SubscribeWithErr ¶ added in v2.8.0
func (p *PGPubsub) SubscribeWithErr(event string, listener ListenerWithErr) (cancel func(), err error)
type Pubsub ¶
type Pubsub interface {
Subscribe(event string, listener Listener) (cancel func(), err error)
SubscribeWithErr(event string, listener ListenerWithErr) (cancel func(), err error)
Publish(event string, message []byte) error
Close() error
}
Pubsub is a generic interface for broadcasting and receiving messages. Implementors should assume high-availability with the backing implementation.
func NewInMemory ¶
func NewInMemory() Pubsub
type Watchdog ¶ added in v2.8.0
type Watchdog struct {
// contains filtered or unexported fields
}
func NewWatchdog ¶ added in v2.8.0
func NewWatchdogWithClock ¶ added in v2.8.0
func NewWatchdogWithClock(ctx context.Context, logger slog.Logger, ps Pubsub, c clock.Clock) *Watchdog
NewWatchdogWithClock returns a watchdog with the given clock. Product code should always call NewWatchDog.
func (*Watchdog) Timeout ¶ added in v2.8.0
func (w *Watchdog) Timeout() <-chan struct{}
Timeout returns a channel that is closed if the watchdog times out. Note that the Timeout() chan will NOT be closed if the Watchdog is Close'd or its context expires, so it is important to read from the Timeout() chan in a select e.g.
w := NewWatchDog(ctx, logger, ps) select { case <-ctx.Done(): case <-w.Timeout():
FreakOut() }
Source Files