Documentation
¶
Overview ¶
Package stream reads the messages of a RabbitMQ stream without moving anything: over the stream protocol with an unnamed consumer (no offset stored, no single-active-consumer group joined), or over AMQP 0-9-1 with an x-stream-offset consumer. Neither path ever stores an offset.
Index ¶
Constants ¶
View Source
const MaxRead = 200
MaxRead is the hard ceiling of one read (read_max's ceiling).
Variables ¶
View Source
var ( ErrNeedsStreamProtocol = errors.New("stream: this needs the stream protocol: set stream_url on the profile") ErrNoTunnel = errors.New("stream: the stream protocol cannot go through an ssh tunnel; set amqp_url to read streams over AMQP") ErrNoSuchStream = errors.New("stream: no such stream") ErrAccess = errors.New("stream: access refused") ErrTimeout = errors.New("stream: timeout") ErrFailed = errors.New("stream: request failed") ErrMax = errors.New("stream: max must be 1 or more") )
Errors. Their text never holds a host, a user or a secret.
Functions ¶
This section is empty.
Types ¶
type From ¶
type From struct {
// Kind is first, last, offset or timestamp.
Kind string
Offset int64
Time time.Time
}
From is where a read starts.
type Reader ¶
type Reader interface {
// Read returns at most max messages from from. It ends at max, when no
// message came for a short while (the end of the stream), or at ctx's
// end, whichever comes first: an empty stream or an offset past its end
// returns no message, never blocks. Each message carries its offset in
// the header x-stream-offset.
Read(ctx context.Context, vhost, stream string, from From, max int) ([]msgmask.Message, error)
// Stats returns the first offset and the first offset of the last
// committed chunk; -1, -1 for an empty stream. Stream protocol only.
Stats(ctx context.Context, vhost, stream string) (first, last int64, err error)
// QueryOffset reads the offset stored by a named consumer. Stream
// protocol only.
QueryOffset(ctx context.Context, vhost, stream, consumerRef string) (int64, error)
}
Reader reads a stream.
func NewAMQP ¶
func NewAMQP(host string, port int, tlsCfg *tls.Config, user string, secret []byte, dial DialFunc) Reader
NewAMQP returns a reader over AMQP 0-9-1: a consumer with x-stream-offset, prefetch max, manual acks (on a stream an ack only returns credit) and a cancel after max. Stats and QueryOffset return ErrNeedsStreamProtocol.
Source Files
Click to show internal directories.
Click to hide internal directories.