fix(context): context transform api change

Do not save the data value each time or it would not be released

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

docs/en_US/extension/native/sink.md

@@ -28,7 +28,7 @@ The main task for a Sink is to implement _collect_ method. The function will be
 
 Most of the time, the map content will be the selective fields. But if `sendError` property is enabled and there are errors happen in the rule, the map content will be like `{"error":"error message here"}`.
 
-The developer can fetch the transformed result from the context method `ctx.TransformOutput()`. The return values are the transformed value of `[]byte` type. Currently, it will be transformed to the json byte array be default or formatted with the set [`dataTemlate` property](../../rules/overview.md#data-template). If the value is transformed by dataTemplate, the second return value will be true. 
+The developer can fetch the transformed result from the context method `ctx.TransformOutput(data)`. The return values are the transformed value of `[]byte` type. Currently, it will be transformed to the json byte array be default or formatted with the set [`dataTemlate` property](../../rules/overview.md#data-template). If the value is transformed by dataTemplate, the second return value will be true. 
 
 The developer can return any errors. However, to leverage the retry feature of eKuiper, the developer must return an error whose message starts with "io error".
 

docs/zh_CN/extension/native/sink.md

@@ -29,9 +29,9 @@ Sink （目标）的主要任务是实现 _collect_ 方法。 当 eKuiper 将任
 
 大多数时候，收到到的 map 的内容为规则选择的列的值。但是，如果 `sendError` 属性设置为 true 且规则有错误，则错误信息会放到 map 里，形似 `{"error":"error message here"}` 。
 
-开发者可通过 context 方法`ctx.TransformOutput()` 获取转换后的字节数组。默认情况下，该方法将返回 json 编码的字节数组，若 [`dataTemlate` 属性](../../rules/overview.md#数据模板) 有设置，则返回格式化后的字符串数组，且第二个返回值设为 true，表示结果已经过变换。
+开发者可通过 context 方法`ctx.TransformOutput(data)` 获取转换后的字节数组。默认情况下，该方法将返回 json 编码的字节数组，若 [`dataTemlate` 属性](../../rules/overview.md#数据模板) 有设置，则返回格式化后的字符串数组，且第二个返回值设为 true，表示结果已经过变换。
 
-需要注意的是，当 [`dataTemlate` 属性](../../rules/overview.md#数据模板) 设置时，开发者可通过 context 方法`ctx.TransformOutput()` 获取转换后的数据。若数据模板未设置，则该方法返回空值。
+需要注意的是，当 [`dataTemlate` 属性](../../rules/overview.md#数据模板) 设置时，开发者可通过 context 方法`ctx.TransformOutput(data)` 获取转换后的数据。若数据模板未设置，则该方法返回空值。
 
 该方法可以返回任何错误类型。但是，如果想要让自动重试机制生效，返回的错误消息必须以 "io error" 开头。大多数情况下，也只有 io 问题才有重试的需要。
 

extensions/sinks/file/file.go

@@ -107,7 +107,7 @@ func (m *fileSink) save(logger api.Logger) {
 
 func (m *fileSink) Collect(ctx api.StreamContext, item interface{}) error {
 	logger := ctx.GetLogger()
-	if v, _, err := ctx.TransformOutput(); err == nil {
+	if v, _, err := ctx.TransformOutput(item); err == nil {
 		logger.Debugf("file sink receive %s", item)
 		m.mux.Lock()
 		m.results = append(m.results, v)

extensions/sinks/redis/redis.go

@@ -126,7 +126,7 @@ func (r *RedisSink) Open(ctx api.StreamContext) (err error) {
 
 func (r *RedisSink) Collect(ctx api.StreamContext, data interface{}) error {
 	logger := ctx.GetLogger()
-	v, _, err := ctx.TransformOutput()
+	v, _, err := ctx.TransformOutput(data)
 	if err != nil {
 		logger.Error(err)
 		return err

extensions/sinks/zmq/zmq.go

@@ -67,7 +67,7 @@ func (m *zmqSink) Open(ctx api.StreamContext) (err error) {
 
 func (m *zmqSink) Collect(ctx api.StreamContext, item interface{}) (err error) {
 	logger := ctx.GetLogger()
-	if v, _, err := ctx.TransformOutput(); err == nil {
+	if v, _, err := ctx.TransformOutput(item); err == nil {
 		logger.Debugf("zmq sink receive %s", item)
 		if m.topic == "" {
 			_, err = m.publisher.Send(string(v), 0)

internal/plugin/portable/runtime/sink.go

@@ -82,7 +82,7 @@ func (ps *PortableSink) Open(ctx api.StreamContext) error {
 
 func (ps *PortableSink) Collect(ctx api.StreamContext, item interface{}) error {
 	ctx.GetLogger().Debugf("Receive %+v", item)
-	if val, _, err := ctx.TransformOutput(); err == nil {
+	if val, _, err := ctx.TransformOutput(item); err == nil {
 		ctx.GetLogger().Debugf("Send %s", val)
 		e := ps.dataCh.Send(val)
 		if e != nil {

internal/topo/context/default_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/lf-edge/ekuiper/internal/conf"
 	"github.com/lf-edge/ekuiper/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/internal/topo/state"
+	"github.com/lf-edge/ekuiper/internal/topo/transform"
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"log"
 	"os"
@@ -200,39 +201,30 @@ func TestDynamicProp(t *testing.T) {
 }
 
 func TestTransition(t *testing.T) {
-	mockFunc := func(d interface{}) ([]byte, bool, error) {
+	var mockFunc transform.TransFunc = func(d interface{}) ([]byte, bool, error) {
 		return []byte(fmt.Sprintf("%v", d)), true, nil
 	}
 	var tests = []struct {
-		trans *TransConfig
-		r     []byte
+		data interface{}
+		r    []byte
 	}{
 		{
-			trans: &TransConfig{
-				Data:  "hello",
-				TFunc: mockFunc,
-			},
-			r: []byte(`hello`),
+			data: "hello",
+			r:    []byte(`hello`),
 		}, {
-			trans: &TransConfig{
-				Data:  "world",
-				TFunc: mockFunc,
-			},
-			r: []byte(`world`),
+			data: "world",
+			r:    []byte(`world`),
 		}, {
-			trans: &TransConfig{
-				Data:  map[string]interface{}{"a": "hello"},
-				TFunc: mockFunc,
-			},
-			r: []byte(`map[a:hello]`),
+			data: map[string]interface{}{"a": "hello"},
+			r:    []byte(`map[a:hello]`),
 		},
 	}
 
 	fmt.Printf("The test bucket size is %d.\n\n", len(tests))
 	ctx := Background().WithMeta("testTransRule", "op1", &state.MemoryStore{}).(*DefaultContext)
+	nc := WithValue(ctx, TransKey, mockFunc)
 	for i, tt := range tests {
-		nc := WithValue(ctx, TransKey, tt.trans)
-		r, _, _ := nc.TransformOutput()
+		r, _, _ := nc.TransformOutput(tt.data)
 		if !reflect.DeepEqual(tt.r, r) {
 			t.Errorf("%d\n\nstmt mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, string(tt.r), string(r))
 		}

internal/topo/context/transform.go

@@ -21,16 +21,12 @@ import (
 
 const TransKey = "$$trans"
 
-type TransConfig struct {
-	Data  interface{}
-	TFunc transform.TransFunc
-}
-
 // TransformOutput Lazy transform output to bytes
-func (c *DefaultContext) TransformOutput() ([]byte, bool, error) {
-	cc, ok := c.Value(TransKey).(*TransConfig)
+func (c *DefaultContext) TransformOutput(data interface{}) ([]byte, bool, error) {
+	v := c.Value(TransKey)
+	f, ok := v.(transform.TransFunc)
 	if ok {
-		return cc.TFunc(cc.Data)
+		return f(data)
 	}
 	return nil, false, fmt.Errorf("no transform configured")
 }

internal/topo/node/sink_node.go

@@ -146,6 +146,7 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 			result <- fmt.Errorf(msg)
 			return
 		}
+		ctx = context.WithValue(ctx.(*context.DefaultContext), context.TransKey, tf)
 
 		m.reset()
 		logger.Infof("open sink node %d instances", m.concurrency)
@@ -194,9 +195,9 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 							}
 							stats.SetBufferLength(int64(len(m.input)))
 							if sconf.RunAsync {
-								go doCollect(ctx, sink, data, stats, sconf, tf, nil)
+								go doCollect(ctx, sink, data, stats, sconf, nil)
 							} else {
-								doCollect(ctx, sink, data, stats, sconf, tf, nil)
+								doCollect(ctx, sink, data, stats, sconf, nil)
 							}
 						case <-ctx.Done():
 							logger.Infof("sink node %s instance %d done", m.name, instance)
@@ -226,9 +227,9 @@ func (m *SinkNode) Open(ctx api.StreamContext, result chan<- error) {
 							}
 							stats.SetBufferLength(int64(len(m.input)))
 							if sconf.RunAsync {
-								go doCollect(ctx, sink, data, stats, sconf, tf, cache.Complete)
+								go doCollect(ctx, sink, data, stats, sconf, cache.Complete)
 							} else {
-								doCollect(ctx, sink, data, stats, sconf, tf, cache.Complete)
+								doCollect(ctx, sink, data, stats, sconf, cache.Complete)
 							}
 						case <-ctx.Done():
 							logger.Infof("sink node %s instance %d done", m.name, instance)
@@ -251,7 +252,7 @@ func (m *SinkNode) reset() {
 	m.statManagers = nil
 }
 
-func doCollect(ctx api.StreamContext, sink api.Sink, item interface{}, stats StatManager, sconf *SinkConf, tp transform.TransFunc, signalCh chan<- int) {
+func doCollect(ctx api.StreamContext, sink api.Sink, item interface{}, stats StatManager, sconf *SinkConf, signalCh chan<- int) {
 	stats.IncTotalRecordsIn()
 	stats.ProcessTimeStart()
 	defer stats.ProcessTimeEnd()
@@ -273,20 +274,16 @@ func doCollect(ctx api.StreamContext, sink api.Sink, item interface{}, stats Sta
 		return
 	}
 	if !sconf.SendSingle {
-		doCollectData(ctx, sink, outs, stats, sconf, tp, signalCh)
+		doCollectData(ctx, sink, outs, stats, sconf, signalCh)
 	} else {
 		for _, d := range outs {
-			doCollectData(ctx, sink, d, stats, sconf, tp, signalCh)
+			doCollectData(ctx, sink, d, stats, sconf, signalCh)
 		}
 	}
 }
 
 // doCollectData outData must be map or []map
-func doCollectData(ctx api.StreamContext, sink api.Sink, outData interface{}, stats StatManager, sconf *SinkConf, tf transform.TransFunc, signalCh chan<- int) {
-	vCtx := context.WithValue(ctx.(*context.DefaultContext), context.TransKey, &context.TransConfig{
-		Data:  outData,
-		TFunc: tf,
-	})
+func doCollectData(ctx api.StreamContext, sink api.Sink, outData interface{}, stats StatManager, sconf *SinkConf, signalCh chan<- int) {
 	retries := sconf.RetryCount
 	for {
 		select {
@@ -294,7 +291,7 @@ func doCollectData(ctx api.StreamContext, sink api.Sink, outData interface{}, st
 			ctx.GetLogger().Infof("sink node %s instance %d stops data resending", ctx.GetOpId(), ctx.GetInstanceId())
 			return
 		default:
-			if err := sink.Collect(vCtx, outData); err != nil {
+			if err := sink.Collect(ctx, outData); err != nil {
 				stats.IncTotalExceptions()
 				ctx.GetLogger().Warnf("sink node %s instance %d publish %s error: %v", ctx.GetOpId(), ctx.GetInstanceId(), outData, err)
 				if sconf.RetryInterval > 0 && retries > 0 && strings.HasPrefix(err.Error(), errorx.IOErr) {

internal/topo/sink/mqtt_sink.go

@@ -244,7 +244,7 @@ func (ms *MQTTSink) Open(ctx api.StreamContext) error {
 
 func (ms *MQTTSink) Collect(ctx api.StreamContext, item interface{}) error {
 	logger := ctx.GetLogger()
-	jsonBytes, _, err := ctx.TransformOutput()
+	jsonBytes, _, err := ctx.TransformOutput(item)
 	if err != nil {
 		return err
 	}

internal/topo/sink/rest_sink.go

@@ -182,7 +182,7 @@ func (me MultiErrors) Error() string {
 func (ms *RestSink) Collect(ctx api.StreamContext, item interface{}) error {
 	logger := ctx.GetLogger()
 	logger.Debugf("rest sink receive %s", item)
-	output, transed, err := ctx.TransformOutput()
+	output, transed, err := ctx.TransformOutput(item)
 	if err != nil {
 		logger.Warnf("rest sink decode data error: %v", err)
 		return nil

internal/topo/sink/rest_sink_test.go

@@ -197,19 +197,12 @@ func TestRestSink_Apply(t *testing.T) {
 		tt.config["url"] = ts.URL
 		s.Configure(tt.config)
 		s.Open(ctx)
+		vCtx := context.WithValue(ctx, context.TransKey, tf)
 		if ss.(bool) {
 			for _, d := range tt.data {
-				vCtx := context.WithValue(ctx, context.TransKey, &context.TransConfig{
-					Data:  d,
-					TFunc: tf,
-				})
 				s.Collect(vCtx, d)
 			}
 		} else {
-			vCtx := context.WithValue(ctx, context.TransKey, &context.TransConfig{
-				Data:  tt.data,
-				TFunc: tf,
-			})
 			s.Collect(vCtx, tt.data)
 		}
 
@@ -359,13 +352,10 @@ func TestRestSinkTemplate_Apply(t *testing.T) {
 		tt.config["url"] = ts.URL
 		s.Configure(tt.config)
 		s.Open(ctx)
+		vCtx := context.WithValue(ctx, context.TransKey, transform.TransFunc(func(d interface{}) ([]byte, bool, error) {
+			return d.([]byte), true, nil
+		}))
 		for _, d := range tt.data {
-			vCtx := context.WithValue(ctx, context.TransKey, &context.TransConfig{
-				Data: d,
-				TFunc: func(_ interface{}) ([]byte, bool, error) {
-					return d, true, nil
-				},
-			})
 			s.Collect(vCtx, d)
 		}
 		s.Close(ctx)

internal/topo/topotest/mocknode/mock_sink.go

@@ -36,7 +36,7 @@ func (m *MockSink) Open(ctx api.StreamContext) error {
 
 func (m *MockSink) Collect(ctx api.StreamContext, item interface{}) error {
 	logger := ctx.GetLogger()
-	if v, _, err := ctx.TransformOutput(); err == nil {
+	if v, _, err := ctx.TransformOutput(item); err == nil {
 		logger.Debugf("mock sink receive %s", item)
 		m.results = append(m.results, v)
 	} else {

pkg/api/stream.go

@@ -155,7 +155,7 @@ type StreamContext interface {
 	// Properties processing, prop is a json path
 	ParseDynamicProp(prop string, data interface{}) (interface{}, error)
 	// Transform output according to the properties like syntax
-	TransformOutput() ([]byte, bool, error)
+	TransformOutput(data interface{}) ([]byte, bool, error)
 }
 
 type Operator interface {