fix bug

@@ -3,7 +3,6 @@ package es
 import (
 import (
 	"context"
 	"context"
 
 
-	jsoniter "github.com/json-iterator/go"
 	"github.com/olivere/elastic"
 	"github.com/olivere/elastic"
 	"github.com/tal-tech/go-stash/stash/config"
 	"github.com/tal-tech/go-stash/stash/config"
 	"github.com/tal-tech/go-zero/core/executors"
 	"github.com/tal-tech/go-zero/core/executors"
@@ -14,7 +13,6 @@ type (
 	Writer struct {
 	Writer struct {
 		docType  string
 		docType  string
 		client   *elastic.Client
 		client   *elastic.Client
-		indexer  *Index
 		inserter *executors.ChunkExecutor
 		inserter *executors.ChunkExecutor
 	}
 	}
 
 
@@ -24,7 +22,7 @@ type (
 	}
 	}
 )
 )
 
 
-func NewWriter(c config.ElasticSearchConf, indexer *Index) (*Writer, error) {
+func NewWriter(c config.ElasticSearchConf) (*Writer, error) {
 	client, err := elastic.NewClient(
 	client, err := elastic.NewClient(
 		elastic.SetSniff(false),
 		elastic.SetSniff(false),
 		elastic.SetURL(c.Hosts...),
 		elastic.SetURL(c.Hosts...),
@@ -37,20 +35,12 @@ func NewWriter(c config.ElasticSearchConf, indexer *Index) (*Writer, error) {
 	writer := Writer{
 	writer := Writer{
 		docType: c.DocType,
 		docType: c.DocType,
 		client:  client,
 		client:  client,
-		indexer: indexer,
 	}
 	}
 	writer.inserter = executors.NewChunkExecutor(writer.execute, executors.WithChunkBytes(c.MaxChunkBytes))
 	writer.inserter = executors.NewChunkExecutor(writer.execute, executors.WithChunkBytes(c.MaxChunkBytes))
 	return &writer, nil
 	return &writer, nil
 }
 }
 
 
-func (w *Writer) Write(m map[string]interface{}) error {
-	bs, err := jsoniter.Marshal(m)
-	if err != nil {
-		return err
-	}
-
-	index := w.indexer.GetIndex(m)
-	val := string(bs)
+func (w *Writer) Write(index, val string) error {
 	return w.inserter.Add(valueWithIndex{
 	return w.inserter.Add(valueWithIndex{
 		index: index,
 		index: index,
 		val:   val,
 		val:   val,

@@ -8,12 +8,14 @@ import (
 
 
 type MessageHandler struct {
 type MessageHandler struct {
 	writer  *es.Writer
 	writer  *es.Writer
+	indexer *es.Index
 	filters []filter.FilterFunc
 	filters []filter.FilterFunc
 }
 }
 
 
-func NewHandler(writer *es.Writer) *MessageHandler {
+func NewHandler(writer *es.Writer, indexer *es.Index) *MessageHandler {
 	return &MessageHandler{
 	return &MessageHandler{
 		writer:  writer,
 		writer:  writer,
+		indexer: indexer,
 	}
 	}
 }
 }
 
 
@@ -29,11 +31,17 @@ func (mh *MessageHandler) Consume(_, val string) error {
 		return err
 		return err
 	}
 	}
 
 
+	index := mh.indexer.GetIndex(m)
 	for _, proc := range mh.filters {
 	for _, proc := range mh.filters {
 		if m = proc(m); m == nil {
 		if m = proc(m); m == nil {
 			return nil
 			return nil
 		}
 		}
 	}
 	}
 
 
-	return mh.writer.Write(m)
+	bs, err := jsoniter.Marshal(m)
+	if err != nil {
+		return err
+	}
+
+	return mh.writer.Write(index, string(bs))
 }
 }

@@ -35,6 +35,10 @@ func main() {
 		)
 		)
 		logx.Must(err)
 		logx.Must(err)
 
 
+		filters := filter.CreateFilters(processor)
+		writer, err := es.NewWriter(processor.Output.ElasticSearch)
+		logx.Must(err)
+
 		var loc *time.Location
 		var loc *time.Location
 		if len(processor.Output.ElasticSearch.TimeZone) > 0 {
 		if len(processor.Output.ElasticSearch.TimeZone) > 0 {
 			loc, err = time.LoadLocation(processor.Output.ElasticSearch.TimeZone)
 			loc, err = time.LoadLocation(processor.Output.ElasticSearch.TimeZone)
@@ -43,11 +47,7 @@ func main() {
 			loc = time.Local
 			loc = time.Local
 		}
 		}
 		indexer := es.NewIndex(client, processor.Output.ElasticSearch.Index, loc)
 		indexer := es.NewIndex(client, processor.Output.ElasticSearch.Index, loc)
-		filters := filter.CreateFilters(processor)
-		writer, err := es.NewWriter(processor.Output.ElasticSearch, indexer)
-		logx.Must(err)
-
-		handle := handler.NewHandler(writer)
+		handle := handler.NewHandler(writer, indexer)
 		handle.AddFilters(filters...)
 		handle.AddFilters(filters...)
 		handle.AddFilters(filter.AddUriFieldFilter("url", "uri"))
 		handle.AddFilters(filter.AddUriFieldFilter("url", "uri"))
 		group.Add(kq.MustNewQueue(processor.Input.Kafka, handle))
 		group.Add(kq.MustNewQueue(processor.Input.Kafka, handle))