84
logger log.Logger,
85
metricsHandler metrics.Handler,
87
>
88
>
p := &processorImpl{
89
>
status: common.DaemonStatusInitialized,
90
>
client: esClient,
91
>
logger: log.With(logger, tag.ComponentIndexerESProcessor),
92
>
metricsHandler: metricsHandler.WithTags(metrics.OperationTag(metrics.ElasticsearchBulkProcessor)),
93
>
indexerConcurrency: uint32(cfg.IndexerConcurrency()),
94
>
bulkProcessorParameters: &client.BulkProcessorParameters{
95
>
Name: visibilityProcessorName,
96
>
NumOfWorkers: cfg.ESProcessorNumOfWorkers(),
97
>
BulkActions: cfg.ESProcessorBulkActions(),
98
>
BulkSize: cfg.ESProcessorBulkSize(),
99
>
FlushInterval: cfg.ESProcessorFlushInterval(),
100
>
},
101
>
}
102
>
p.bulkProcessorParameters.AfterFunc = p.bulkAfterAction
103
>
p.bulkProcessorParameters.BeforeFunc = p.bulkBeforeAction
104
>
return p
105
>
}
106
107
func (p *processorImpl) Start() {