indexing_merge

Documentation

Index

• func New(c *config.Config) (pipeline.Processor, error)
• type Config
• type IndexingMergeProcessor
  • func (processor *IndexingMergeProcessor) Name() string
  • func (processor *IndexingMergeProcessor) NewBulkWorker(ctx *pipeline.Context, count *int, bulkSizeInByte int, wg *sync.WaitGroup)
  • func (processor *IndexingMergeProcessor) 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"`
	IdleTimeoutInSeconds int `config:"idle_timeout_in_seconds"`
	BulkSizeInKB         int `config:"bulk_size_in_kb"`
	BulkSizeInMB         int `config:"bulk_size_in_mb"`

	IndexName string `config:"index_name"`
	TypeName  string `config:"type_name"`

	KeyField string `config:"key_field"` //the field name used as document's primary key aka `_id

	Elasticsearch string `config:"elasticsearch"`

	InputQueue string `config:"input_queue"`

	OutputQueue struct {
		Name   string                 `config:"name"`
		Labels map[string]interface{} `config:"label" json:"label,omitempty"`
	} `config:"output_queue"`

	FailureQueue string `config:"failure_queue"`
	InvalidQueue string `config:"invalid_queue"`
}

type IndexingMergeProcessor

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

func (*IndexingMergeProcessor) Name

func (processor *IndexingMergeProcessor) Name() string

处理纯 json 格式的消息索引

func (*IndexingMergeProcessor) NewBulkWorker

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

func (*IndexingMergeProcessor) Process

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

合并批量处理的操作，这里只用来合并请求和构造 bulk 请求。 TODO 重启子进程，当子进程挂了之后