Repository: lf-edge/ekuiper
Pull request #1227 (2022-04-12 10:10:07)

fix(memory): memory channels should have buffer to avoid message drops


Patch:
diff --git a/internal/topo/memory/manager.go b/internal/topo/memory/manager.go
index 869162b5..c9d3a70d 100644
--- a/internal/topo/memory/manager.go
+++ b/internal/topo/memory/manager.go
@@ -67,10 +67,10 @@ func createPub(topic string) {
 	}
 }
 
-func createSub(wildcard string, regex *regexp.Regexp, sourceId string) chan api.SourceTuple {
+func createSub(wildcard string, regex *regexp.Regexp, sourceId string, bufferLength int) chan api.SourceTuple {
 	mu.Lock()
 	defer mu.Unlock()
-	ch := make(chan api.SourceTuple)
+	ch := make(chan api.SourceTuple, bufferLength)
 	if regex != nil {
 		subExps[sourceId] = &subChan{
 			regex: regex,
diff --git a/internal/topo/memory/source.go b/internal/topo/memory/source.go
index df081169..a4ea11a1 100644
--- a/internal/topo/memory/source.go
+++ b/internal/topo/memory/source.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -17,22 +17,22 @@ package memory
 import (
 	"fmt"
 	"github.com/lf-edge/ekuiper/pkg/api"
+	"github.com/lf-edge/ekuiper/pkg/cast"
 	"regexp"
 	"strings"
 )
 
 type source struct {
-	topic      string
-	topicRegex *regexp.Regexp
-	input      <-chan api.SourceTuple
+	topic        string
+	topicRegex   *regexp.Regexp
+	bufferLength int
 }
 
 func (s *source) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, _ chan<- error) {
-	ch := createSub(s.topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
-	s.input = ch
+	ch := createSub(s.topic, s.topicRegex, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
 	for {
 		select {
-		case v, opened := <-s.input:
+		case v, opened := <-ch:
 			if !opened {
 				return
 			}
@@ -43,8 +43,14 @@ func (s *source) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, _
 	}
 }
 
-func (s *source) Configure(datasource string, _ map[string]interface{}) error {
+func (s *source) Configure(datasource string, props map[string]interface{}) error {
 	s.topic = datasource
+	s.bufferLength = 1024
+	if c, ok := props["bufferLength"]; ok {
+		if bl, err := cast.ToInt(c, cast.STRICT); err != nil || bl > 0 {
+			s.bufferLength = bl
+		}
+	}
 	if strings.ContainsAny(datasource, "+#") {
 		r, err := getRegexp(datasource)
 		if err != nil {


Test patch:
diff --git a/internal/topo/memory/manager_test.go b/internal/topo/memory/manager_test.go
index 97f7daf6..7ce8d1a6 100644
--- a/internal/topo/memory/manager_test.go
+++ b/internal/topo/memory/manager_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -112,7 +112,7 @@ func TestCreateAndClose(t *testing.T) {
 				return
 			}
 		}
-		c := createSub(sourceTopics[i], r, fmt.Sprintf("%d", i))
+		c := createSub(sourceTopics[i], r, fmt.Sprintf("%d", i), 100)
 		chans = append(chans, c)
 	}