Documentation
¶
Overview ¶
Package redis provides a distributed stream.Service backed by Redis Streams.
The caller owns the *redis.Client (or redis.UniversalClient); this package never opens or closes the connection.
Maps each stream concept to native Redis Streams commands:
- Publish → XADD stream MAXLEN ~ N * payload ... h:... ...
- Subscribe → XGROUP CREATE on first call, then per-worker XREADGROUP loops with BLOCK for low-latency wake-up
- Reclaim → XAUTOCLAIM with min-idle = VisibilityTimeout
- Ack → XACK
- DLQ routing → XADD to <stream>.dead followed by XACK on the source
MaxLen trim is reasserted on every publish via MAXLEN ~. MaxAge trim runs as part of the publish path via XTRIM MINID.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrClosed = errors.New("stream/redis: service closed")
ErrClosed is returned by Publish or Subscribe after Close.
Functions ¶
This section is empty.
Types ¶
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service is the Redis Streams stream.Service implementation.
func New ¶
func New(client rds.UniversalClient, opts ...stream.ServiceOption) *Service
New constructs a stream/redis Service. The caller owns client.
func (*Service) Close ¶
Close drains all consumers. Idempotent. ctx bounds the overall close: cancellation aborts before the next consumer is drained and returns ctx.Err() — already-drained consumers stay drained.
func (*Service) Publish ¶
func (s *Service) Publish(ctx context.Context, name string, payload []byte, opts ...stream.PublishOption) error
Publish sends a message via XADD. The payload is stored under the "payload" field; headers are stored as additional fields with "h:" prefix. MaxLen is applied as XADD MAXLEN ~ (approximate trimming for speed). MaxAge is applied via XTRIM MINID after the publish.
Source Files