json_indexing

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"`
	IndexPrefix          string `config:"index_prefix"`
	IndexName            string `config:"index_name"`
	TypeName             string `config:"type_name"`
	Elasticsearch        string `config:"elasticsearch"`
	InputQueue           string `config:"input_queue"`
	FailureQueue         string `config:"failure_queue"`
	InvalidQueue         string `config:"invalid_queue"`
	CheckESAvailable     bool   `config:"check_available"`
}

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

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