Repository: lf-edge/ekuiper
Pull request #1643 (2023-02-23 03:03:20)

feat: support multiple neuron connection


Patch:
diff --git a/etc/sinks/neuron.json b/etc/sinks/neuron.json
index e15cf8d0..dc5448c5 100644
--- a/etc/sinks/neuron.json
+++ b/etc/sinks/neuron.json
@@ -17,6 +17,21 @@
     }
   },
   "properties": [
+    {
+      "name": "url",
+      "default": "tcp://127.0.0.1:7081",
+      "optional": false,
+      "control": "text",
+      "type": "string",
+      "hint": {
+        "en_US": "The nng url to connect to neuron.",
+        "zh_CN": "连接 Neuron 的 nng url"
+      },
+      "label": {
+        "en_US": "URL",
+        "zh_CN": "路径"
+      }
+    },
     {
       "name": "nodeName",
       "optional": true,
diff --git a/etc/sources/neuron.json b/etc/sources/neuron.json
index 721fde59..a98ddb22 100644
--- a/etc/sources/neuron.json
+++ b/etc/sources/neuron.json
@@ -17,7 +17,25 @@
     }
   },
   "libs": [],
-  "properties": {},
+  "properties": {
+    "default": [
+      {
+        "name": "url",
+        "default": "tcp://127.0.0.1:7081",
+        "optional": false,
+        "control": "text",
+        "type": "string",
+        "hint": {
+          "en_US": "The nng url to connect to neuron.",
+          "zh_CN": "连接 Neuron 的 nng url"
+        },
+        "label": {
+          "en_US": "URL",
+          "zh_CN": "路径"
+        }
+      }
+    ]
+  },
   "outputs": [
     {
       "label": {
diff --git a/etc/sources/neuron.yaml b/etc/sources/neuron.yaml
index e69de29b..d545533e 100644
--- a/etc/sources/neuron.yaml
+++ b/etc/sources/neuron.yaml
@@ -0,0 +1,5 @@
+default:
+  # The nng connection url to connect to the neuron
+  url: tcp://127.0.0.1:7081
+ipc:
+  url: ipc:///tmp/neuron-ekuiper.ipc
\ No newline at end of file
diff --git a/internal/topo/neuron/connection.go b/internal/topo/neuron/connection.go
index 10b33051..60b6422c 100644
--- a/internal/topo/neuron/connection.go
+++ b/internal/topo/neuron/connection.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2023 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.
@@ -26,101 +26,115 @@ import (
 	"go.nanomsg.org/mangos/v3"
 	"go.nanomsg.org/mangos/v3/protocol/pair"
 	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+	_ "go.nanomsg.org/mangos/v3/transport/tcp"
 	"sync"
 	"sync/atomic"
 	"time"
 )
 
 const (
-	NeuronTopic = "$$neuron"
-	NeuronUrl   = "ipc:///tmp/neuron-ekuiper.ipc"
+	TopicPrefix      = "$$neuron_"
+	DefaultNeuronUrl = "ipc:///tmp/neuron-ekuiper.ipc"
 )
 
+type conninfo struct {
+	count  int
+	sock   mangos.Socket
+	opened int32
+}
+
 var (
-	m               sync.RWMutex
-	connectionCount int
-	sock            mangos.Socket
-	opened          int32
-	sendTimeout     = 100
+	m             sync.RWMutex
+	connectionReg = make(map[string]*conninfo)
+	sendTimeout   = 100
 )
 
 // createOrGetNeuronConnection creates a new neuron connection or returns an existing one
 // This is the entry function for creating a neuron connection singleton
 // The context is from a rule, but the singleton will server for multiple rules
-func createOrGetConnection(sc api.StreamContext, url string) error {
+func createOrGetConnection(sc api.StreamContext, url string) (*conninfo, error) {
 	m.Lock()
 	defer m.Unlock()
-	sc.GetLogger().Infof("createOrGetConnection count: %d", connectionCount)
-	if connectionCount == 0 {
-		sc.GetLogger().Infof("Creating neuron connection")
-		contextLogger := conf.Log.WithField("neuron_connection", 0)
+	sc.GetLogger().Infof("createOrGetConnection for %s", url)
+	info, ok := connectionReg[url]
+	if !ok || info.count <= 0 {
+		sc.GetLogger().Infof("Creating neuron connection for %s", url)
+		contextLogger := conf.Log.WithField("neuron_connection_url", url)
 		ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
-		ruleId := "$$neuron_connection"
-		opId := "$$neuron_connection"
+		ruleId := "$$neuron_connection_" + url
+		opId := "$$neuron_connection_" + url
 		store, err := state.CreateStore(ruleId, 0)
 		if err != nil {
 			ctx.GetLogger().Errorf("neuron connection create store error %v", err)
-			return err
+			return nil, err
 		}
 		sctx := ctx.WithMeta(ruleId, opId, store)
-		err = connect(sctx, url)
+		info = &conninfo{count: 0}
+		connectionReg[url] = info
+		err = connect(sctx, url, info)
 		if err != nil {
-			return err
+			return nil, err
 		}
-		sc.GetLogger().Infof("Neuron connected")
-		pubsub.CreatePub(NeuronTopic)
-		go run(sctx)
+		sc.GetLogger().Infof("Neuron %s connected", url)
+		pubsub.CreatePub(TopicPrefix + url)
+		go run(sctx, info, url)
 	}
-	connectionCount++
-	return nil
+	info.count++
+	return info, nil
 }
 
 func closeConnection(ctx api.StreamContext, url string) error {
 	m.Lock()
 	defer m.Unlock()
-	ctx.GetLogger().Infof("closeConnection count: %d", connectionCount)
-	pubsub.RemovePub(NeuronTopic)
-	if connectionCount == 1 {
-		err := disconnect(url)
-		if err != nil {
-			return err
+	ctx.GetLogger().Infof("closeConnection %s", url)
+	info, ok := connectionReg[url]
+	if !ok {
+		return fmt.Errorf("no connection for %s", url)
+	}
+	pubsub.RemovePub(TopicPrefix + url)
+	if info.count == 1 {
+		if info.sock != nil {
+			err := info.sock.Close()
+			if err != nil {
+				return err
+			}
 		}
 	}
-	connectionCount--
+	info.count--
 	return nil
 }
 
 // nng connections
 
 // connect to nng
-func connect(ctx api.StreamContext, url string) error {
+func connect(ctx api.StreamContext, url string, info *conninfo) error {
 	var err error
-	sock, err = pair.NewSocket()
+	info.sock, err = pair.NewSocket()
 	if err != nil {
 		return err
 	}
 	// options consider to export
-	err = sock.SetOption(mangos.OptionSendDeadline, time.Duration(sendTimeout)*time.Millisecond)
+	err = info.sock.SetOption(mangos.OptionSendDeadline, time.Duration(sendTimeout)*time.Millisecond)
 	if err != nil {
 		return err
 	}
-	sock.SetPipeEventHook(func(ev mangos.PipeEvent, p mangos.Pipe) {
+	info.sock.SetPipeEventHook(func(ev mangos.PipeEvent, p mangos.Pipe) {
 		switch ev {
 		case mangos.PipeEventAttached:
-			atomic.StoreInt32(&opened, 1)
+			atomic.StoreInt32(&info.opened, 1)
 			conf.Log.Infof("neuron connection attached")
 		case mangos.PipeEventAttaching:
 			conf.Log.Infof("neuron connection is attaching")
 		case mangos.PipeEventDetached:
-			atomic.StoreInt32(&opened, 0)
+			atomic.StoreInt32(&info.opened, 0)
 			conf.Log.Warnf("neuron connection detached")
-			pubsub.ProduceError(ctx, NeuronTopic, fmt.Errorf("neuron connection detached"))
+			pubsub.ProduceError(ctx, TopicPrefix+url, fmt.Errorf("neuron connection detached"))
 		}
 	})
 	//sock.SetOption(mangos.OptionWriteQLen, 100)
 	//sock.SetOption(mangos.OptionReadQLen, 100)
 	//sock.SetOption(mangos.OptionBestEffort, false)
-	if err = sock.DialOptions(url, map[string]interface{}{
+	if err = info.sock.DialOptions(url, map[string]interface{}{
 		mangos.OptionDialAsynch:       true, // will not report error and keep connecting
 		mangos.OptionMaxReconnectTime: 5 * time.Second,
 		mangos.OptionReconnectTime:    100 * time.Millisecond,
@@ -133,11 +147,11 @@ func connect(ctx api.StreamContext, url string) error {
 
 // run the loop to receive message from the nng connection singleton
 // exit when connection is closed
-func run(ctx api.StreamContext) {
+func run(ctx api.StreamContext, info *conninfo, url string) {
 	ctx.GetLogger().Infof("neuron source receiving loop started")
 	for {
 		// no receiving deadline, will wait until the socket closed
-		if msg, err := sock.Recv(); err == nil {
+		if msg, err := info.sock.Recv(); err == nil {
 			ctx.GetLogger().Debugf("neuron received message %s", string(msg))
 			result := make(map[string]interface{})
 			err := json.Unmarshal(msg, &result)
@@ -145,7 +159,7 @@ func run(ctx api.StreamContext) {
 				ctx.GetLogger().Errorf("neuron decode message error %v", err)
 				continue
 			}
-			pubsub.Produce(ctx, NeuronTopic, result)
+			pubsub.Produce(ctx, TopicPrefix+url, result)
 		} else if err == mangos.ErrClosed {
 			ctx.GetLogger().Infof("neuron connection closed, exit receiving loop")
 			return
@@ -155,20 +169,10 @@ func run(ctx api.StreamContext) {
 	}
 }
 
-func publish(ctx api.StreamContext, data []byte) error {
+func publish(ctx api.StreamContext, data []byte, info *conninfo) error {
 	ctx.GetLogger().Debugf("publish to neuron: %s", string(data))
-	if sock != nil && atomic.LoadInt32(&opened) == 1 {
-		return sock.Send(data)
+	if info.sock != nil && atomic.LoadInt32(&info.opened) == 1 {
+		return info.sock.Send(data)
 	}
 	return fmt.Errorf("%s: neuron connection is not established", errorx.IOErr)
 }
-
-func disconnect(_ string) error {
-	if sock != nil {
-		err := sock.Close()
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
diff --git a/internal/topo/neuron/sink.go b/internal/topo/neuron/sink.go
index bc34d070..4e18e3e6 100644
--- a/internal/topo/neuron/sink.go
+++ b/internal/topo/neuron/sink.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -25,9 +25,8 @@ import (
 )
 
 type sink struct {
-	url       string
-	c         *c
-	connected bool
+	c   *c
+	cli *conninfo
 }
 
 type c struct {
@@ -35,7 +34,8 @@ type c struct {
 	GroupName string   `json:"groupName"`
 	Tags      []string `json:"tags"`
 	// If sent with the raw converted string or let us range over the result map
-	Raw bool `json:"raw"`
+	Raw bool   `json:"raw"`
+	Url string `json:"url"`
 }
 
 type neuronTemplate struct {
@@ -46,9 +46,9 @@ type neuronTemplate struct {
 }
 
 func (s *sink) Configure(props map[string]interface{}) error {
-	s.url = NeuronUrl
 	cc := &c{
 		Raw: false,
+		Url: DefaultNeuronUrl,
 	}
 	err := cast.MapToStruct(props, cc)
 	if err != nil {
@@ -68,11 +68,11 @@ func (s *sink) Configure(props map[string]interface{}) error {
 
 func (s *sink) Open(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("Opening neuron sink")
-	err := createOrGetConnection(ctx, s.url)
+	cli, err := createOrGetConnection(ctx, s.c.Url)
 	if err != nil {
 		return err
 	}
-	s.connected = true
+	s.cli = cli
 	return nil
 }
 
@@ -83,7 +83,7 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 		if err != nil {
 			return err
 		}
-		return publish(ctx, r)
+		return publish(ctx, r, s.cli)
 	} else {
 		switch d := data.(type) {
 		case []map[string]interface{}:
@@ -104,8 +104,8 @@ func (s *sink) Collect(ctx api.StreamContext, data interface{}) error {
 
 func (s *sink) Close(ctx api.StreamContext) error {
 	ctx.GetLogger().Debugf("closing neuron sink")
-	if s.connected {
-		return closeConnection(ctx, s.url)
+	if s.cli != nil {
+		return closeConnection(ctx, s.c.Url)
 	}
 	return nil
 }
@@ -136,7 +136,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 			for _, k := range keys {
 				t.TagName = k
 				t.Value = el[k]
-				err := doPublish(ctx, t)
+				err := doPublish(ctx, t, s.cli)
 				if err != nil {
 					return err
 				}
@@ -145,7 +145,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 			for k, v := range el {
 				t.TagName = k
 				t.Value = v
-				err := doPublish(ctx, t)
+				err := doPublish(ctx, t, s.cli)
 				if err != nil {
 					return err
 				}
@@ -164,7 +164,7 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 				ctx.GetLogger().Errorf("Error get the value of tag %s: %v", t.TagName, err)
 				continue
 			}
-			err := doPublish(ctx, t)
+			err := doPublish(ctx, t, s.cli)
 			if err != nil {
 				return err
 			}
@@ -173,12 +173,12 @@ func (s *sink) SendMapToNeuron(ctx api.StreamContext, el map[string]interface{})
 	return nil
 }
 
-func doPublish(ctx api.StreamContext, t *neuronTemplate) error {
+func doPublish(ctx api.StreamContext, t *neuronTemplate, cli *conninfo) error {
 	r, err := json.Marshal(t)
 	if err != nil {
 		return fmt.Errorf("Error marshall the tag payload %v: %v", t, err)
 	}
-	err = publish(ctx, r)
+	err = publish(ctx, r, cli)
 	if err != nil {
 		return fmt.Errorf("%s: Error publish the tag payload %s: %v", errorx.IOErr, t.TagName, err)
 	}
diff --git a/internal/topo/neuron/source.go b/internal/topo/neuron/source.go
index bd8e2064..80dfd67a 100644
--- a/internal/topo/neuron/source.go
+++ b/internal/topo/neuron/source.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -22,31 +22,37 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/infra"
 )
 
+type sc struct {
+	Url          string `json:"url,omitempty"`
+	BufferLength int    `json:"bufferLength,omitempty"`
+}
+
 type source struct {
-	url          string
-	bufferLength int
+	c *sc
 }
 
 func (s *source) Configure(_ string, props map[string]interface{}) error {
-	s.url = NeuronUrl
-	s.bufferLength = 1024
-	if c, ok := props["bufferLength"]; ok {
-		if bl, err := cast.ToInt(c, cast.STRICT); err != nil || bl > 0 {
-			s.bufferLength = bl
-		}
+	cc := &sc{
+		BufferLength: 1024,
+		Url:          DefaultNeuronUrl,
+	}
+	err := cast.MapToStruct(props, cc)
+	if err != nil {
+		return err
 	}
+	s.c = cc
 	return nil
 }
 
 func (s *source) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
-	err := createOrGetConnection(ctx, s.url)
+	_, err := createOrGetConnection(ctx, s.c.Url)
 	if err != nil {
 		infra.DrainError(ctx, err, errCh)
 		return
 	}
-	defer closeConnection(ctx, s.url)
-	ch := pubsub.CreateSub(NeuronTopic, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.bufferLength)
-	defer pubsub.CloseSourceConsumerChannel(NeuronTopic, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
+	defer closeConnection(ctx, s.c.Url)
+	ch := pubsub.CreateSub(TopicPrefix+s.c.Url, nil, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()), s.c.BufferLength)
+	defer pubsub.CloseSourceConsumerChannel(TopicPrefix+s.c.Url, fmt.Sprintf("%s_%s_%d", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId()))
 	for {
 		select {
 		case v, opened := <-ch:
diff --git a/internal/topo/node/node.go b/internal/topo/node/node.go
index c0f3bb90..a7258056 100644
--- a/internal/topo/node/node.go
+++ b/internal/topo/node/node.go
@@ -201,19 +201,37 @@ func SourceOpen(sourceType string, config map[string]interface{}) error {
 	if v, ok := config["DATASOURCE"]; ok {
 		dataSource = v.(string)
 	}
-
 	ns, err := io.Source(sourceType)
 	if err != nil {
 		return err
 	}
-	err = ns.Configure(dataSource, config)
-	if err != nil {
-		return err
-	}
+	if ns == nil {
+		lns, err := io.LookupSource(sourceType)
+		if err != nil {
+			return err
+		}
+		if lns == nil {
+			// should not happen
+			return fmt.Errorf("source %s not found", sourceType)
+		}
+		err = lns.Configure(dataSource, config)
+		if err != nil {
+			return err
+		}
 
-	contextLogger := conf.Log.WithField("rule", "TestSourceOpen"+"_"+sourceType)
-	ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
-	defer ns.Close(ctx)
+		contextLogger := conf.Log.WithField("rule", "TestSourceOpen"+"_"+sourceType)
+		ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+		lns.Close(ctx)
+	} else {
+		err = ns.Configure(dataSource, config)
+		if err != nil {
+			return err
+		}
+
+		contextLogger := conf.Log.WithField("rule", "TestSourceOpen"+"_"+sourceType)
+		ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
+		ns.Close(ctx)
+	}
 
 	return nil
 
diff --git a/internal/topo/redis/lookup.go b/internal/topo/redis/lookup.go
index 54381b85..66f6a5e6 100644
--- a/internal/topo/redis/lookup.go
+++ b/internal/topo/redis/lookup.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -41,13 +41,17 @@ type lookupSource struct {
 }
 
 func (s *lookupSource) Configure(datasource string, props map[string]interface{}) error {
-	db, err := cast.ToInt(datasource, cast.CONVERT_ALL)
-	if err != nil {
-		return fmt.Errorf("invalid datasource, it must be an integer but got %s", datasource)
+	if datasource != "/$$TEST_CONNECTION$$" {
+		db, err := cast.ToInt(datasource, cast.CONVERT_ALL)
+		if err != nil {
+			return fmt.Errorf("invalid datasource, it must be an integer but got %s", datasource)
+		}
+		s.db = db
+	} else {
+		s.db = 0
 	}
-	s.db = db
 	cfg := &conf{}
-	err = cast.MapToStruct(props, cfg)
+	err := cast.MapToStruct(props, cfg)
 	if err != nil {
 		return err
 	}
@@ -58,17 +62,18 @@ func (s *lookupSource) Configure(datasource string, props map[string]interface{}
 		return errors.New("redis dataType must be string or list")
 	}
 	s.c = cfg
-	return nil
-}
-
-func (s *lookupSource) Open(ctx api.StreamContext) error {
-	ctx.GetLogger().Infof("Opening redis lookup source with conf %v", s.c)
 	s.cli = redis.NewClient(&redis.Options{
 		Addr:     s.c.Addr,
 		Username: s.c.Username,
 		Password: s.c.Password,
 		DB:       s.db,
 	})
+	_, err = s.cli.Ping().Result()
+	return err
+}
+
+func (s *lookupSource) Open(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("Opening redis lookup source with conf %v", s.c)
 	return nil
 }
 
diff --git a/sdk/go/example/mirror/go.mod b/sdk/go/example/mirror/go.mod
index 7d4de1e1..006a1c68 100644
--- a/sdk/go/example/mirror/go.mod
+++ b/sdk/go/example/mirror/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
 	go.nanomsg.org/mangos/v3 v3.2.1 // indirect
-	golang.org/x/sys v0.0.0-20210510120138-977fb7262007 // indirect
+	golang.org/x/sys v0.1.0 // indirect
 )
 
 replace github.com/lf-edge/ekuiper/sdk/go => ../../
diff --git a/sdk/go/example/mirror/go.sum b/sdk/go/example/mirror/go.sum
index 02ba623e..910f24db 100644
--- a/sdk/go/example/mirror/go.sum
+++ b/sdk/go/example/mirror/go.sum
@@ -18,5 +18,5 @@ go.nanomsg.org/mangos/v3 v3.2.1 h1:/7pG6tUJO5ZGznG+waoMy6WrurArODDRJu18848oQnw=
 go.nanomsg.org/mangos/v3 v3.2.1/go.mod h1:RxVwsn46YtfJ74mF8MeVo+MFjg545KCI50NuZrFXmzc=
 golang.org/x/sys v0.0.0-20181128092732-4ed8d59d0b35/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20210510120138-977fb7262007 h1:gG67DSER+11cZvqIMb8S8bt0vZtiN6xWYARwirrOSfE=
-golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0 h1:kunALQeHf1/185U1i0GOB/fy1IPRDDpuoOOqRReG57U=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=


Test patch:
diff --git a/internal/server/rest_test.go b/internal/server/rest_test.go
index 8e602006..a26e62fe 100644
--- a/internal/server/rest_test.go
+++ b/internal/server/rest_test.go
@@ -85,7 +85,7 @@ func Test_sourcesManageHandler(t *testing.T) {
 	}
 
 	//create table
-	buf := bytes.NewBuffer([]byte(` {"sql":"CREATE TABLE alertTable() WITH (DATASOURCE=\"0\", TYPE=\"redis\", KIND=\"lookup\")"}`))
+	buf := bytes.NewBuffer([]byte(` {"sql":"CREATE TABLE alertTable() WITH (DATASOURCE=\"0\", TYPE=\"memory\", KEY=\"id\", KIND=\"lookup\")"}`))
 	req, _ = http.NewRequest(http.MethodPost, "http://localhost:8080/streams?kind=lookup", buf)
 	w = httptest.NewRecorder()
 
@@ -132,7 +132,7 @@ func Test_sourcesManageHandler(t *testing.T) {
 	w = httptest.NewRecorder()
 	r.ServeHTTP(w, req)
 
-	expect = []byte(`{"Name":"alertTable","Options":{"datasource":"0","type":"redis","kind":"lookup"},"Statement":null,"StreamFields":null,"StreamType":1}`)
+	expect = []byte(`{"Name":"alertTable","Options":{"datasource":"0","type":"memory", "key":"id","kind":"lookup"},"Statement":null,"StreamFields":null,"StreamType":1}`)
 	exp = map[string]interface{}{}
 	_ = json.NewDecoder(bytes.NewBuffer(expect)).Decode(&exp)
 
diff --git a/internal/topo/neuron/neuron_test.go b/internal/topo/neuron/neuron_test.go
index 17483886..c8a4b7bc 100644
--- a/internal/topo/neuron/neuron_test.go
+++ b/internal/topo/neuron/neuron_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -81,9 +81,9 @@ func mockNeuron(send bool, recv bool) (mangos.Socket, chan []byte) {
 func TestMultiSourceSink(t *testing.T) {
 	// start and test 2 sources
 	exp := []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "yellow"}, "errors": map[string]interface{}{"tag_name3": 122.0}}, map[string]interface{}{"topic": "$$neuron"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "green", "tag_name3": 60.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 15.4, "tag_name2": "green", "tag_name3": 70.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "yellow"}, "errors": map[string]interface{}{"tag_name3": 122.0}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "green", "tag_name3": 60.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 15.4, "tag_name2": "green", "tag_name3": 70.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
 	}
 	s1 := GetSource()
 	err := s1.Configure("new", nil)
diff --git a/internal/topo/neuron/source_test.go b/internal/topo/neuron/source_test.go
index e9b42d1f..0eb8995e 100644
--- a/internal/topo/neuron/source_test.go
+++ b/internal/topo/neuron/source_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2023 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.
@@ -27,9 +27,9 @@ import (
 
 func TestRun(t *testing.T) {
 	exp := []api.SourceTuple{
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "yellow"}, "errors": map[string]interface{}{"tag_name3": 122.0}}, map[string]interface{}{"topic": "$$neuron"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "green", "tag_name3": 60.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron"}),
-		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 15.4, "tag_name2": "green", "tag_name3": 70.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "yellow"}, "errors": map[string]interface{}{"tag_name3": 122.0}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 11.22, "tag_name2": "green", "tag_name3": 60.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
+		api.NewDefaultSourceTuple(map[string]interface{}{"group_name": "group1", "timestamp": 1646125996000.0, "node_name": "node1", "values": map[string]interface{}{"tag_name1": 15.4, "tag_name2": "green", "tag_name3": 70.0}, "errors": map[string]interface{}{}}, map[string]interface{}{"topic": "$$neuron_ipc:///tmp/neuron-ekuiper.ipc"}),
 	}
 	s := GetSource()
 	err := s.Configure("new", nil)