queue_consumer

Documentation

Index

• func New(c *config.Config) (pipeline.Processor, error)
• type Config
• type DiskQueueConsumer
  • func (processor *DiskQueueConsumer) Name() string
  • func (processor *DiskQueueConsumer) NewBulkWorker(ctx *pipeline.Context, count *int, wg *sync.WaitGroup)
  • func (processor *DiskQueueConsumer) Process(ctx *pipeline.Context) error

Functions

func New

func New(c *config.Config) (pipeline.Processor, error)

Types

type Config

type Config struct {
	NumOfWorkers        int `config:"worker_size"`
	IdleTimeoutInSecond int `config:"idle_timeout_in_seconds"`

	FetchMinBytes    int `config:"fetch_min_bytes"`
	FetchMaxBytes    int `config:"fetch_max_bytes"`
	FetchMaxMessages int `config:"fetch_max_messages"`
	FetchMaxWaitMs   int `config:"fetch_max_wait_ms"`

	SaveSuccessDocsToQueue bool   `config:"save_partial_success_requests"`
	PartialSuccessQueue    string `config:"partial_success_queue"`

	InputQueue   string `config:"input_queue"`
	FailureQueue string `config:"failure_queue"`
	InvalidQueue string `config:"invalid_queue"`

	Elasticsearch string   `config:"elasticsearch"`
	WaitingAfter  []string `config:"waiting_after"`
	Compress      bool     `config:"compress"`

	SafetyParse   bool `config:"safety_parse"`
	DocBufferSize int  `config:"doc_buffer_size"`
}

type DiskQueueConsumer

type DiskQueueConsumer struct {
	// contains filtered or unexported fields
}

func (*DiskQueueConsumer) Name

func (processor *DiskQueueConsumer) Name() string

func (*DiskQueueConsumer) NewBulkWorker

func (processor *DiskQueueConsumer) NewBulkWorker(ctx *pipeline.Context, count *int, wg *sync.WaitGroup)

func (*DiskQueueConsumer) Process

func (processor *DiskQueueConsumer) Process(ctx *pipeline.Context) error