consumer

Documentation

Index

• type Consumer
  • func CreateConsumer(ctx context.Context, opts *ConsumerOptions, ...) (Consumer, error)
• type ConsumerOptions

Types

type Consumer

type Consumer interface {
	// Error returns the error of scanner failed.
	// Will block until scanner is closed or Chan is dry out.
	Error() error

	// Done returns a channel which will be closed when scanner is finished or closed.
	Done() <-chan struct{}

	// Close the consumer, release the underlying resources.
	Close()
}

Consumer is the interface that wraps the basic consume method on grpc stream. Consumer is work on a single stream on grpc, so Consumer cannot recover from failure because of the stream is broken.

func CreateConsumer

func CreateConsumer(
	ctx context.Context,
	opts *ConsumerOptions,
	handlerClient streamingpb.StreamingNodeHandlerServiceClient,
) (Consumer, error)

CreateConsumer creates a new consumer client.

type ConsumerOptions

type ConsumerOptions struct {
	// The cosume target
	Assignment *types.PChannelInfoAssigned

	// VChannel is the vchannel of the consumer.
	VChannel string

	// DeliverPolicy is the deliver policy of the consumer.
	DeliverPolicy options.DeliverPolicy

	// DeliverFilters is the deliver filters of the consumer.
	DeliverFilters []options.DeliverFilter

	// Handler is the message handler used to handle message after recv from consumer.
	MessageHandler message.Handler
}

ConsumerOptions is the options for creating a consumer.