Documentation
¶
Index ¶
- type AddEvent
- type AddEventOptions
- type AddStream
- type AddStreamOptions
- type Reader
- func WithReaderBlockDuration(d time.Duration) Reader
- func WithReaderBufferSize(size int) Reader
- func WithReaderMaxPolled(n int64) Reader
- func WithReaderStartAfter(id string) Reader
- func WithReaderStartAt(startAt time.Time) Reader
- func WithReaderStartAtNewest() Reader
- func WithReaderStartAtOldest() Reader
- func WithReaderTopic(topic string) Reader
- func WithReaderTopicPattern(pattern string) Reader
- type ReaderOptions
- type Sink
- func WithSinkAckGracePeriod(d time.Duration) Sink
- func WithSinkBlockDuration(d time.Duration) Sink
- func WithSinkBufferSize(size int) Sink
- func WithSinkMaxPolled(n int64) Sink
- func WithSinkNoAck() Sink
- func WithSinkStartAfter(id string) Sink
- func WithSinkStartAt(startAt time.Time) Sink
- func WithSinkStartAtNewest() Sink
- func WithSinkStartAtOldest() Sink
- func WithSinkTopic(topic string) Sink
- func WithSinkTopicPattern(pattern string) Sink
- type SinkOptions
- type Stream
- type StreamOptions
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AddEvent ¶
type AddEvent func(*AddEventOptions)
AddEvent is an option for adding an event to a stream.
func WithOnlyIfStreamExists ¶
func WithOnlyIfStreamExists() AddEvent
WithOnlyIfStreamExists only adds the event if the stream exists.
type AddEventOptions ¶
func ParseAddEventOptions ¶
func ParseAddEventOptions(opts ...AddEvent) AddEventOptions
ParseAddEventOptions parses the given options and returns the corresponding AddEventOptions.
type AddStream ¶
type AddStream func(*AddStreamOptions)
AddStream is an option for adding a stream to a sink.
func WithAddStreamStartAfter ¶
WithAddStreamStartAfter sets the last read event ID, the sink will start reading from the next event. Only one of WithAddStreamStartAtNewest, WithAddStreamStartAtOldest, WithAddStreamStartAfter or WithAddStreamStartAt can be used.
func WithAddStreamStartAt ¶
WithAddStreamStartAt sets the start position for the added stream to the event added on or after startAt. Only one of WithAddStreamStartAtNewest, WithAddStreamStartAtOldest, WithAddStreamStartAfter or WithAddStreamStartAt can be used.
func WithAddStreamStartAtNewest ¶
func WithAddStreamStartAtNewest() AddStream
WithAddStreamStartAtNewest sets the sink start position for the added stream to the newest event. Only one of WithAddStreamStartAtNewest, WithAddStreamStartAtOldest, WithAddStreamStartAfter or WithAddStreamStartAt can be used.
func WithAddStreamStartAtOldest ¶
func WithAddStreamStartAtOldest() AddStream
WithAddStreamStartAtOldest sets the sink start position for the added stream to the oldest event. Only one of WithAddStreamStartAtNewest, WithAddStreamStartAtOldest, WithAddStreamStartAfter or WithAddStreamStartAt can be used.
type AddStreamOptions ¶
type AddStreamOptions struct {
// LastEventID is the ID after which delivery starts for this stream.
LastEventID string
// contains filtered or unexported fields
}
AddStreamOptions keeps its v1 field first; new fields are only ever appended so existing keyed construction remains source-compatible.
func ParseAddStreamOptions ¶
func ParseAddStreamOptions(opts ...AddStream) AddStreamOptions
ParseAddStreamOptions parses the options and returns the add stream options.
func (AddStreamOptions) HasConflictingStartOptions ¶ added in v1.10.0
func (o AddStreamOptions) HasConflictingStartOptions() bool
HasConflictingStartOptions reports whether more than one cursor-start option was supplied. AddStream rejects this instead of silently accepting the last option.
type Reader ¶
type Reader func(*ReaderOptions)
Reader is a sink creation option.
func WithReaderBlockDuration ¶
WithReaderBlockDuration sets the maximum amount of time the reader waits for MaxPolled events. The default block duration is 5 seconds. NewReader rejects durations below one millisecond because Redis block timing is millisecond precision and every read must have a finite shutdown bound.
func WithReaderBufferSize ¶
WithReaderBufferSize sets the reader channel buffer size. The default buffer size is 1000. If the buffer is full the reader blocks until the buffer has space available.
func WithReaderMaxPolled ¶
WithReaderMaxPolled sets the maximum number of events polled by the reader at once. The default maximum number of events is 1000.
func WithReaderStartAfter ¶
WithReaderStartAfter sets the last read event ID, the reader will start reading from the next event.
func WithReaderStartAt ¶
WithReaderStartAt sets the start position for the reader to the event added on or after startAt.
func WithReaderStartAtNewest ¶
func WithReaderStartAtNewest() Reader
WithReaderStartAtNewest sets the reader start position to the newest event, this is the default. Only one of WithReaderStartAtNewest, WithReaderStartAtOldest, WithReaderStartAfter or WithReaderStartAt can be used.
func WithReaderStartAtOldest ¶
func WithReaderStartAtOldest() Reader
WithReaderStartAtOldest sets the reader start position to the oldest event. Only one of WithReaderStartAtOldest, WithReaderStartAfter or WithReaderStartAt should be used.
func WithReaderTopic ¶
WithReaderTopic sets the reader topic.
func WithReaderTopicPattern ¶
WithReaderTopicPattern sets the reader topic pattern. NewReader returns an error when pattern is not a valid regular expression.
type ReaderOptions ¶
type ReaderOptions struct {
// BlockDuration is the XREAD block duration.
BlockDuration time.Duration
// MaxPolled is the maximum number of events read per XREAD call.
MaxPolled int64
// Topic delivers only events published with this exact topic.
Topic string
// TopicPattern delivers only events whose topic matches this regex.
TopicPattern string
// BufferSize is the capacity of each subscription channel.
BufferSize int
// LastEventID is the ID after which delivery starts.
LastEventID string
// contains filtered or unexported fields
}
ReaderOptions keeps its v1 fields first, in v1 order; new fields are only ever appended so existing keyed construction and field access remain source-compatible across feature releases.
func ParseReaderOptions ¶
func ParseReaderOptions(opts ...Reader) ReaderOptions
ParseReaderOptions parses the given options and returns the corresponding reader options.
func (ReaderOptions) HasConflictingStartOptions ¶ added in v1.10.0
func (o ReaderOptions) HasConflictingStartOptions() bool
HasConflictingStartOptions reports whether more than one cursor-start option was supplied. Constructors reject this instead of silently accepting the last option.
type Sink ¶
type Sink func(*SinkOptions)
Sink is a sink creation option.
func WithSinkAckGracePeriod ¶
WithSinkAckGracePeriod sets the grace period for acknowledging events. The default grace period is 20 seconds; NewSink rejects values below one millisecond. Note: all sinks with identical names must have the same ack grace period.
func WithSinkBlockDuration ¶
WithSinkBlockDuration sets the maximum amount of time the sink waits for MaxPolled events. The default block duration is 5 seconds. NewSink rejects durations below one millisecond because Redis block timing is millisecond precision and every read must have a finite shutdown bound.
func WithSinkBufferSize ¶
WithSinkBufferSize sets the sink channel buffer size. The default buffer size is 1000. If the buffer is full the sink blocks until the buffer has space available.
func WithSinkMaxPolled ¶
WithSinkMaxPolled sets the maximum number of events polled by the sink at once. The default maximum number of events is 1000.
func WithSinkNoAck ¶
func WithSinkNoAck() Sink
WithSinkNoAck atomically acknowledges each event before delivering it to subscribers, preserving at-most-once delivery without requiring Sink.Ack.
func WithSinkStartAfter ¶
WithSinkStartAfter sets the last read event ID, the sink will start reading from the next event. Only one of WithSinkStartAtNewest, WithSinkStartAtOldest, WithSinkStartAfter or WithSinkStartAt can be used.
func WithSinkStartAt ¶
WithSinkStartAt sets the start position for the sink, defaults to the last event. Only one of WithSinkStartAtNewest, WithSinkStartAtOldest, WithSinkStartAfter or WithSinkStartAt can be used.
func WithSinkStartAtNewest ¶
func WithSinkStartAtNewest() Sink
WithSinkStartAtNewest sets the sink start position to the newest event, this is the default. Only one of WithSinkStartAtNewest, WithSinkStartAtOldest, WithSinkStartAfter or WithSinkStartAt can be used.
func WithSinkStartAtOldest ¶
func WithSinkStartAtOldest() Sink
WithSinkStartAtOldest sets the sink start position to the oldest event. Only one of WithSinkStartAtNewest, WithSinkStartAtOldest, WithSinkStartAfter or WithSinkStartAt can be used.
func WithSinkTopicPattern ¶
WithSinkTopicPattern sets the sink topic pattern. NewSink returns an error when pattern is not a valid regular expression.
type SinkOptions ¶
type SinkOptions struct {
// BlockDuration is the XREADGROUP block duration.
BlockDuration time.Duration
// MaxPolled is the maximum number of events read per XREADGROUP call.
MaxPolled int64
// Topic delivers only events published with this exact topic.
Topic string
// TopicPattern delivers only events whose topic matches this regex.
TopicPattern string
// BufferSize is the capacity of each subscription channel.
BufferSize int
// LastEventID is the ID after which delivery starts.
LastEventID string
// NoAck atomically acknowledges each event before delivery.
NoAck bool
// AckGracePeriod bounds how long an unacknowledged event stays owned
// by one consumer before stale recovery may reclaim it.
AckGracePeriod time.Duration
// contains filtered or unexported fields
}
SinkOptions keeps its v1 fields first, in v1 order; new fields are only ever appended so existing keyed construction and field access remain source-compatible across feature releases.
func ParseSinkOptions ¶
func ParseSinkOptions(opts ...Sink) SinkOptions
ParseSinkOptions parses the options and returns the sink options.
func (SinkOptions) HasConflictingStartOptions ¶ added in v1.10.0
func (o SinkOptions) HasConflictingStartOptions() bool
HasConflictingStartOptions reports whether more than one cursor-start option was supplied. Constructors reject this instead of silently accepting the last option.
type Stream ¶
type Stream func(*StreamOptions)
Stream is a stream creation option.
func WithStreamDeadline ¶ added in v1.10.0
WithStreamDeadline makes the stream generation expire at deadline. The deadline is persisted when the handle first opens; handles for the same generation must not request a different deadline. TTL options cannot be combined with this option.
func WithStreamLogger ¶
WithStreamLogger sets the logger used by the stream.
func WithStreamMaxLen ¶
WithStreamMaxLen sets the positive maximum number of events stored by the stream. NewStream rejects zero and negative values.
func WithStreamSlidingTTL ¶ added in v1.5.0
WithStreamSlidingTTL sets a sliding TTL on the Redis key backing the stream. The TTL is refreshed on every published event; NewStream rejects values below Redis's one-millisecond precision.
func WithStreamTTL ¶ added in v1.5.0
WithStreamTTL sets an absolute TTL on the Redis key backing the stream. The TTL is set once and never extended; NewStream rejects values below Redis's one-millisecond precision.
func WithUnboundedStream ¶ added in v1.10.0
func WithUnboundedStream() Stream
WithUnboundedStream disables MAXLEN trimming. Callers must remove settled events explicitly. It cannot be combined with WithStreamMaxLen.
type StreamOptions ¶
type StreamOptions struct {
MaxLen int
Logger pulse.Logger
// TTL configures a retention window for the Redis key backing the stream.
// When zero, no TTL is applied.
TTL time.Duration
// TTLSliding controls whether the TTL is refreshed on every published event.
// When false, the TTL is applied once (absolute TTL) and never extended.
TTLSliding bool
// MaxLenSet distinguishes an omitted MaxLen from an explicit request.
MaxLenSet bool
// Unbounded disables MAXLEN trimming entirely.
Unbounded bool
// TTLSet distinguishes an omitted TTL from an explicitly invalid zero.
TTLSet bool
// Deadline is the immutable absolute expiry claimed by this stream
// generation when DeadlineSet is true.
Deadline time.Time
// DeadlineSet distinguishes an omitted deadline from the zero time.
DeadlineSet bool
}
StreamOptions keeps its v1 fields first, in v1 order; new fields are only ever appended so existing keyed construction and field access remain source-compatible across feature releases.
func ParseStreamOptions ¶
func ParseStreamOptions(opts ...Stream) StreamOptions
ParseStreamOptions parses the given options and returns the corresponding StreamOptions.
Source Files