File tree Expand file tree Collapse file tree 3 files changed +9
-11
lines changed Expand file tree Collapse file tree 3 files changed +9
-11
lines changed Original file line number Diff line number Diff line change 1717 ElasticSearchConf struct {
1818 Hosts []string
1919 Index string
20- DocType string `json:",default=doc "`
20+ DocType string `json:",default=_doc "`
2121 TimeZone string `json:",optional"`
2222 MaxChunkBytes int `json:",default=15728640"` // default 15M
2323 Compress bool `json:",default=false"`
Original file line number Diff line number Diff line change @@ -2,6 +2,7 @@ package es
22
33import (
44 "context"
5+ "fmt"
56
67 "github.com/kevwan/go-stash/stash/config"
78 "github.com/olivere/elastic/v7"
@@ -10,7 +11,7 @@ import (
1011 "github.com/zeromicro/go-zero/core/logx"
1112)
1213
13- const es8Version = "8 .0.0"
14+ const es8Version = "v8 .0.0"
1415
1516type (
1617 Writer struct {
2223
2324 valueWithIndex struct {
2425 index string
25- val string
26+ val map [ string ] interface {}
2627 }
2728)
2829
@@ -45,13 +46,13 @@ func NewWriter(c config.ElasticSearchConf) (*Writer, error) {
4546 writer := Writer {
4647 docType : c .DocType ,
4748 client : client ,
48- esVersion : version ,
49+ esVersion : fmt . Sprintf ( "v%s" , version ) ,
4950 }
5051 writer .inserter = executors .NewChunkExecutor (writer .execute , executors .WithChunkBytes (c .MaxChunkBytes ))
5152 return & writer , nil
5253}
5354
54- func (w * Writer ) Write (index , val string ) error {
55+ func (w * Writer ) Write (index string , val map [ string ] interface {} ) error {
5556 return w .inserter .Add (valueWithIndex {
5657 index : index ,
5758 val : val ,
Original file line number Diff line number Diff line change 44 jsoniter "github.com/json-iterator/go"
55 "github.com/kevwan/go-stash/stash/es"
66 "github.com/kevwan/go-stash/stash/filter"
7+ "time"
78)
89
910type MessageHandler struct {
@@ -37,11 +38,7 @@ func (mh *MessageHandler) Consume(_, val string) error {
3738 return nil
3839 }
3940 }
41+ m ["timestamp" ] = time .Now ()
4042
41- bs , err := jsoniter .Marshal (m )
42- if err != nil {
43- return err
44- }
45-
46- return mh .writer .Write (index , string (bs ))
43+ return mh .writer .Write (index , m )
4744}
You can’t perform that action at this time.
0 commit comments