README
¶
Batched PubSub Reader
This package implements batched reading of messages from a pubsub.Subscription.
It provides a BatchedSubscription with a ReceiveBatch method to read messages in batches
based on a given batch capacity or batching interval.
Usage
capacity := 100
interval := time.Second
sub := batsub.NewBatchedSubscription(subscription, capacity, interval)
err := sub.ReceiveBatch(ctx, func(ctx context.Context, messages []*pubsub.Message){
// handle batch of messages using a batch-processing library
errors := mylib.BatchProcessMessages(messages)
for i, err := errors {
if err != nil {
// TODO: handle error
continue
}
messages[i].Ack()
}
})
Also see the PubSub to BigQuery example.
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BatchedSubscription ¶
type BatchedSubscription struct {
Receiver
// contains filtered or unexported fields
}
BatchedSubscription implements automatic batching based on a defined batch Capacity and a FlushInterval.
func NewBatchedSubscription ¶
func NewBatchedSubscription(receiver Receiver, capacity int, flushInterval time.Duration) *BatchedSubscription
NewBatchedSubscription returns an initalized BatNewBatchedSubscription.
func (*BatchedSubscription) ReceiveBatch ¶
func (sub *BatchedSubscription) ReceiveBatch(ctx context.Context, f BatchFunc) error
ReceiveBatch calls f with the outstanding batched messages from the subscription.
Basic Example:
err := sub.ReceiveBatch(ctx, func(ctx context.Context, messages []*pubsub.Message){
for i, m := range messages {
// TODO: handle message
m.Ack()
}
})
Batch Processing Example:
err := sub.ReceiveBatch(ctx, func(ctx context.Context, messages []*pubsub.Message){
// handle batch of messages using a batch-processing library
errors := mylib.BatchProcessMessages(messages)
for i, err := errors {
if err != nil {
// TODO: handle error
continue
}
messages[i].Ack()
}
})
Source Files
Directories
Click to show internal directories.
Click to hide internal directories.