feat: pass config and es client to sink response handler context

Trendyol · May 14, 2024 · 80c6fea · 80c6fea
1 parent eb6147a
commit 80c6fea
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 6 deletions.
diff --git a/elasticsearch/bulk/bulk.go b/elasticsearch/bulk/bulk.go
@@ -29,6 +29,7 @@ type Bulk struct {
 	sinkResponseHandler    dcpElasticsearch.SinkResponseHandler
 	metric                 *Metric
 	collectionIndexMapping map[string]string
+	config                 *config.Config
 	batchKeys              map[string]int
 	dcpCheckpointCommit    func()
 	batchTicker            *time.Ticker
@@ -85,6 +86,7 @@ func NewBulk(
 		esClient:               esClient,
 		metric:                 &Metric{},
 		collectionIndexMapping: config.Elasticsearch.CollectionIndexMapping,
+		config:                 config,
 		typeName:               helper.Byte(config.Elasticsearch.TypeName),
 		readers:                readers,
 		concurrentRequest:      config.Elasticsearch.ConcurrentRequest,
@@ -374,12 +376,16 @@ func (b *Bulk) executeSinkResponseHandler(batchActions []*document.ESActionDocum
 		key := getActionKey(*action)
 		if _, ok := errorData[key]; ok {
 			b.sinkResponseHandler.OnError(&dcpElasticsearch.SinkResponseHandlerContext{
-				Action: action,
-				Err:    fmt.Errorf(errorData[key]),
+				Action:              action,
+				Err:                 fmt.Errorf(errorData[key]),
+				ElasticsearchClient: b.esClient,
+				Config:              b.config,
 			})
 		} else {
 			b.sinkResponseHandler.OnSuccess(&dcpElasticsearch.SinkResponseHandlerContext{
-				Action: action,
+				Action:              action,
+				ElasticsearchClient: b.esClient,
+				Config:              b.config,
 			})
 		}
 	}

diff --git a/elasticsearch/sink_response_handler.go b/elasticsearch/sink_response_handler.go
@@ -1,10 +1,16 @@
 package elasticsearch
 
-import "github.com/Trendyol/go-dcp-elasticsearch/elasticsearch/document"
+import (
+	"github.com/Trendyol/go-dcp-elasticsearch/config"
+	"github.com/Trendyol/go-dcp-elasticsearch/elasticsearch/document"
+	"github.com/elastic/go-elasticsearch/v7"
+)
 
 type SinkResponseHandlerContext struct {
-	Action *document.ESActionDocument
-	Err    error
+	Action              *document.ESActionDocument
+	Err                 error
+	Config              *config.Config
+	ElasticsearchClient *elasticsearch.Client
 }
 
 type SinkResponseHandler interface {