Repository: lf-edge/ekuiper
Pull request #2929 (2024-06-14 10:20:03)

fix(lookup): update schema
Tests will be covered by fvt


Patch:
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index 42987322..3453d63e 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -20,8 +20,10 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
 	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
+	"github.com/lf-edge/ekuiper/v2/internal/io/neuron"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/nng"
 )
 
 func init() {
@@ -30,7 +32,7 @@ func init() {
 	// modules.RegisterSource("httppush", func() api.Source { return &http.PushSource{} })
 	modules.RegisterSource("file", file.GetSource)
 	modules.RegisterSource("memory", func() api.Source { return memory.GetSource() })
-	// modules.RegisterSource("neuron", func() api.Source { return neuron.GetSource() })
+	modules.RegisterSource("neuron", neuron.GetSource)
 	// modules.RegisterSource("websocket", func() api.Source { return &websocket.WebsocketSource{} })
 	// modules.RegisterSource("simulator", func() api.Source { return &simulator.Source{} })
 
@@ -40,7 +42,7 @@ func init() {
 	// modules.RegisterSink("rest", func() api.Sink { return &http.RestSink{} })
 	modules.RegisterSink("nop", func() api.Sink { return &sink.NopSink{} })
 	modules.RegisterSink("memory", func() api.Sink { return memory.GetSink() })
-	// modules.RegisterSink("neuron", func() api.Sink { return neuron.GetSink() })
+	modules.RegisterSink("neuron", neuron.GetSink)
 	modules.RegisterSink("file", file.GetSink)
 	// modules.RegisterSink("websocket", func() api.Sink { return &websocket.WebSocketSink{} })
 
@@ -48,6 +50,7 @@ func init() {
 	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
 
 	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
+	modules.RegisterConnection("nng", nng.CreateConnection)
 }
 
 type Manager struct{}
diff --git a/internal/io/memory/pubsub/manager.go b/internal/io/memory/pubsub/manager.go
index ea004fc6..75b417db 100644
--- a/internal/io/memory/pubsub/manager.go
+++ b/internal/io/memory/pubsub/manager.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 EMQ Technologies Co., Ltd.
+// Copyright 2021-2024 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.
@@ -117,6 +117,10 @@ func ProduceList(ctx api.StreamContext, topic string, list []MemTuple) {
 	doProduce(ctx, topic, list)
 }
 
+func ProduceError(ctx api.StreamContext, topic string, err error) {
+	doProduce(ctx, topic, err)
+}
+
 func doProduce(ctx api.StreamContext, topic string, data any) {
 	c, exists := pubTopics[topic]
 	if !exists {
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index 460ee433..420fa0f6 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -105,8 +105,9 @@ func (conn *Connection) Subscribe(topic string, info *SubscriptionInfo) error {
 	return handleToken(token)
 }
 
-func (conn *Connection) Close(ctx api.StreamContext) {
+func (conn *Connection) Close(ctx api.StreamContext) error {
 	conn.Client.Disconnect(1)
+	return nil
 }
 
 func (conn *Connection) Ping(ctx api.StreamContext) error {
diff --git a/internal/io/neuron/connector.go b/internal/io/neuron/connector.go
new file mode 100644
index 00000000..4aed4d51
--- /dev/null
+++ b/internal/io/neuron/connector.go
@@ -0,0 +1,49 @@
+// Copyright 2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"time"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/nng"
+)
+
+func ping(ctx api.StreamContext, props map[string]any) error {
+	props["protocol"] = PROTOCOL
+	cli, err := nng.CreateConnection(ctx, props)
+	if err != nil {
+		return err
+	}
+	defer cli.Close(ctx)
+	time.Sleep(1000 * time.Millisecond)
+	return cli.Ping(ctx)
+}
+
+func connect(ctx api.StreamContext, url string, props map[string]any) (modules.Connection, error) {
+	ctx.GetLogger().Infof("Connecting to neuron")
+	connId := PROTOCOL + url
+	return connection.FetchConnection(ctx, connId, "nng", props)
+}
+
+func close(ctx api.StreamContext, conn modules.Connection, url string, props map[string]any) {
+	if conn != nil {
+		connId := PROTOCOL + url
+		_ = connection.DetachConnection(ctx, connId, props)
+		conn.DetachSub(ctx, props)
+	}
+}
diff --git a/internal/io/neuron/sink.go b/internal/io/neuron/sink.go
new file mode 100644
index 00000000..15ff075e
--- /dev/null
+++ b/internal/io/neuron/sink.go
@@ -0,0 +1,212 @@
+// Copyright 2022-2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/conf"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/nng"
+)
+
+type c struct {
+	NodeName  string   `json:"nodeName"`
+	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"`
+}
+
+type sink struct {
+	c     *c
+	cc    *nng.SockConf
+	cli   *nng.Sock
+	props map[string]any
+}
+
+type neuronTemplate struct {
+	GroupName string      `json:"group_name"`
+	NodeName  string      `json:"node_name"`
+	TagName   string      `json:"tag_name"`
+	Value     interface{} `json:"value"`
+}
+
+func (s *sink) Provision(_ api.StreamContext, props map[string]any) error {
+	props["protocol"] = PROTOCOL
+	sc, err := nng.ValidateConf(props)
+	if err != nil {
+		return err
+	}
+	s.cc = sc
+	cc := &c{
+		Raw: false,
+	}
+	err = cast.MapToStruct(props, cc)
+	if err != nil {
+		return err
+	}
+	if !cc.Raw {
+		if cc.NodeName == "" {
+			return fmt.Errorf("node name is required if raw is not set")
+		}
+		if cc.GroupName == "" {
+			return fmt.Errorf("group name is required if raw is not set")
+		}
+	}
+	s.c = cc
+	s.props = props
+	return nil
+}
+
+func (s *sink) Ping(ctx api.StreamContext, props map[string]interface{}) error {
+	props["protocol"] = PROTOCOL
+	return ping(ctx, props)
+}
+
+func (s *sink) Connect(ctx api.StreamContext) error {
+	cli, err := connect(ctx, s.cc.Url, s.props)
+	if err != nil {
+		return err
+	}
+	s.cli = cli.(*nng.Sock)
+	return nil
+}
+
+func (s *sink) Collect(ctx api.StreamContext, data api.MessageTuple) error {
+	ctx.GetLogger().Debugf("receive %+v", data)
+	if s.c.Raw {
+		m := data.ToMap()
+		r, err := json.Marshal(m)
+		if err != nil {
+			return err
+		}
+		return s.cli.Send(ctx, r)
+	} else {
+		return s.SendMapToNeuron(ctx, data)
+	}
+}
+
+// CollectList sends all data at best effort
+// It never return error, so it is not supported for cache and retry
+func (s *sink) CollectList(ctx api.StreamContext, data api.MessageTupleList) error {
+	data.RangeOfTuples(func(index int, tuple api.MessageTuple) bool {
+		err := s.Collect(ctx, tuple)
+		if err != nil {
+			ctx.GetLogger().Errorf("send data %v error %v", data, err)
+		}
+		return true
+	})
+	return nil
+}
+
+func (s *sink) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Debugf("closing neuron sink")
+	close(ctx, s.cli, s.cc.Url, s.props)
+	s.cli = nil
+	return nil
+}
+
+func (s *sink) SendMapToNeuron(ctx api.StreamContext, tuple api.MessageTuple) error {
+	n := s.c.NodeName
+	g := s.c.GroupName
+	if dp, ok := tuple.(api.HasDynamicProps); ok {
+		temp, transformed := dp.DynamicProps(n)
+		if transformed {
+			n = temp
+		}
+	}
+	if dp, ok := tuple.(api.HasDynamicProps); ok {
+		temp, transformed := dp.DynamicProps(g)
+		if transformed {
+			g = temp
+		}
+	}
+	t := &neuronTemplate{
+		NodeName:  n,
+		GroupName: g,
+	}
+	var (
+		ok  bool
+		err error
+	)
+	el := tuple.ToMap()
+	if len(s.c.Tags) == 0 {
+		if conf.IsTesting {
+			var keys []string
+			for k := range el {
+				keys = append(keys, k)
+			}
+			sort.Strings(keys)
+			for _, k := range keys {
+				t.TagName = k
+				t.Value = el[k]
+				err := doPublish(ctx, s.cli, t)
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			for k, v := range el {
+				t.TagName = k
+				t.Value = v
+				err := doPublish(ctx, s.cli, t)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	} else {
+		// Send as many tags as possible in order and drop the tag if it is invalid
+		for _, tag := range s.c.Tags {
+			t.TagName, err = ctx.ParseTemplate(tag, el)
+			if err != nil {
+				ctx.GetLogger().Errorf("Error parsing tag %s: %v", tag, err)
+				continue
+			}
+			t.Value, ok = el[t.TagName]
+			if !ok {
+				ctx.GetLogger().Errorf("Error get the value of tag %s: %v", t.TagName, err)
+				continue
+			}
+			err := doPublish(ctx, s.cli, t)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func doPublish(ctx api.StreamContext, cli *nng.Sock, t *neuronTemplate) error {
+	r, err := json.Marshal(t)
+	if err != nil {
+		return fmt.Errorf("Error marshall the tag payload %v: %v", t, err)
+	}
+	err = cli.Send(ctx, r)
+	if err != nil {
+		return errorx.NewIOErr(fmt.Sprintf(`Error publish the tag payload %s: %v`, t.TagName, err))
+	}
+	ctx.GetLogger().Debugf("Send %s", r)
+	return nil
+}
+
+func GetSink() api.Sink {
+	return &sink{}
+}
diff --git a/internal/io/neuron/source.go b/internal/io/neuron/source.go
new file mode 100644
index 00000000..569588e7
--- /dev/null
+++ b/internal/io/neuron/source.go
@@ -0,0 +1,110 @@
+// Copyright 2022-2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"go.nanomsg.org/mangos/v3"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/infra"
+	"github.com/lf-edge/ekuiper/v2/pkg/nng"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+const (
+	DefaultNeuronUrl = "ipc:///tmp/neuron-ekuiper.ipc"
+	PROTOCOL         = "pair"
+)
+
+type source struct {
+	c     *nng.SockConf
+	cli   *nng.Sock
+	props map[string]any
+}
+
+func (s *source) Provision(_ api.StreamContext, props map[string]any) error {
+	props["protocol"] = PROTOCOL
+	sc, err := nng.ValidateConf(props)
+	if err != nil {
+		return err
+	}
+	s.c = sc
+	s.props = props
+	return nil
+}
+
+func (s *source) Ping(ctx api.StreamContext, props map[string]any) error {
+	props["protocol"] = PROTOCOL
+	return ping(ctx, props)
+}
+
+func (s *source) SubId(props map[string]any) string {
+	var url string
+	u, ok := props["url"]
+	if ok {
+		url = u.(string)
+	}
+	return "nng:" + PROTOCOL + url
+}
+
+func (s *source) Connect(ctx api.StreamContext) error {
+	cli, err := connect(ctx, s.c.Url, s.props)
+	if err != nil {
+		return err
+	}
+	s.cli = cli.(*nng.Sock)
+	return nil
+}
+
+func (s *source) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestErr api.ErrorIngest) error {
+	ctx.GetLogger().Infof("neuron source receiving loop started")
+	go func() {
+		err := infra.SafeRun(func() error {
+			for {
+				// no receiving deadline, will wait until the socket closed
+				if msg, err := s.cli.Recv(); err == nil {
+					ctx.GetLogger().Debugf("nng received message %s", string(msg))
+					ingest(ctx, msg, nil, timex.GetNow())
+				} else if err == mangos.ErrClosed {
+					ctx.GetLogger().Infof("neuron connection closed, retry after 1 second")
+					ingestErr(ctx, errors.New("neuron connection closed"))
+					time.Sleep(1 * time.Second)
+					continue
+				} else {
+					ingestErr(ctx, fmt.Errorf("neuron receiving error %v", err))
+				}
+			}
+		})
+		if err != nil {
+			ctx.GetLogger().Errorf("exit neuron source subscribe for %v", err)
+		}
+	}()
+	return nil
+}
+
+func (s *source) Close(ctx api.StreamContext) error {
+	ctx.GetLogger().Infof("closing neuron source")
+	close(ctx, s.cli, s.c.Url, s.props)
+	s.cli = nil
+	return nil
+}
+
+func GetSource() api.Source {
+	return &source{}
+}
diff --git a/internal/topo/node/lookup_node.go b/internal/topo/node/lookup_node.go
index 8db80c76..0e9efc30 100644
--- a/internal/topo/node/lookup_node.go
+++ b/internal/topo/node/lookup_node.go
@@ -91,21 +91,16 @@ func NewLookupNode(ctx api.StreamContext, name string, isBytesLookup bool, field
 		for _, field := range fields {
 			sch[field] = nil
 		}
-		var fsch map[string]*ast.JsonStreamField
 		if sc.PayloadField != "" {
-			fsch = map[string]*ast.JsonStreamField{
-				sc.PayloadField: nil,
-			}
-		} else {
-			fsch = sch
+			sch[sc.PayloadField] = nil
 		}
 
 		decoder, err := converter.GetOrCreateConverter(ctx, &ast.Options{
 			FORMAT:     srcOptions.FORMAT,
 			SCHEMAID:   srcOptions.SCHEMAID,
 			DELIMITER:  srcOptions.DELIMITER,
-			IsWildCard: false,
-			Schema:     fsch,
+			IsWildCard: fields == nil,
+			Schema:     sch,
 		})
 		if err != nil {
 			msg := fmt.Sprintf("cannot get converter from format %s, schemaId %s: %v", srcOptions.FORMAT, srcOptions.SCHEMAID, err)
@@ -118,7 +113,7 @@ func NewLookupNode(ctx api.StreamContext, name string, isBytesLookup bool, field
 				FORMAT:     sc.PayloadFormat,
 				SCHEMAID:   sc.PayloadSchemaId,
 				DELIMITER:  sc.PayloadDelimiter,
-				IsWildCard: false,
+				IsWildCard: fields == nil,
 				Schema:     sch,
 			})
 			if err != nil {
diff --git a/internal/topo/node/sink_node.go b/internal/topo/node/sink_node.go
index af71fa1a..c41373e7 100644
--- a/internal/topo/node/sink_node.go
+++ b/internal/topo/node/sink_node.go
@@ -197,10 +197,11 @@ func NewTupleSinkNode(ctx api.StreamContext, name string, sink api.TupleCollecto
 // return error that cannot be sent
 func tupleCollect(ctx api.StreamContext, sink api.Sink, data any) (err error) {
 	switch d := data.(type) {
-	case api.MessageTuple:
-		err = sink.(api.TupleCollector).Collect(ctx, d)
+	// Some tuple list type also implements tuple. So need to handle list firstly
 	case api.MessageTupleList:
 		err = sink.(api.TupleCollector).CollectList(ctx, d)
+	case api.MessageTuple:
+		err = sink.(api.TupleCollector).Collect(ctx, d)
 	case error:
 		err = sink.(api.TupleCollector).Collect(ctx, model.NewDefaultSourceTuple(xsql.Message{"error": d.Error()}, nil, timex.GetNow()))
 	default:
diff --git a/internal/topo/node/source_node.go b/internal/topo/node/source_node.go
index 79c705ac..c3f3f877 100644
--- a/internal/topo/node/source_node.go
+++ b/internal/topo/node/source_node.go
@@ -99,6 +99,8 @@ func (m *SourceNode) ingestAnyTuple(ctx api.StreamContext, data any, meta map[st
 	// Maps are expected from user extension
 	case map[string]any:
 		m.ingestMap(mess, meta, ts)
+	case xsql.Message:
+		m.ingestMap(mess, meta, ts)
 	case []map[string]any:
 		for _, mm := range mess {
 			m.ingestMap(mm, meta, ts)
diff --git a/internal/topo/planner/planner_source.go b/internal/topo/planner/planner_source.go
index 11bde073..65d83d08 100644
--- a/internal/topo/planner/planner_source.go
+++ b/internal/topo/planner/planner_source.go
@@ -82,10 +82,17 @@ func splitSource(ctx api.StreamContext, t *DataSourcePlan, ss api.Source, option
 		err         error
 		srcConnNode node.DataSourceNode
 	)
-	if sp.SelId == "" {
+	// Some connection only allow one subscription. The source should implement UniqueSub to provide a subId to avoid multiple connection.
+	us, hasSubId := ss.(model.UniqueSub)
+	if sp.SelId == "" && !hasSubId {
 		srcConnNode, err = node.NewSourceNode(ctx, string(t.name), ss, props, options)
 	} else { // connection selector is set as a one node sub_topo
-		selName := fmt.Sprintf("%s/%s", sp.SelId, t.streamStmt.Options.DATASOURCE)
+		var selName string
+		if sp.SelId != "" {
+			selName = fmt.Sprintf("%s/%s", sp.SelId, t.streamStmt.Options.DATASOURCE)
+		} else {
+			selName = us.SubId(props)
+		}
 		srcSubtopo, existed := topo.GetSubTopo(selName)
 		if !existed {
 			var scn node.DataSourceNode
diff --git a/pkg/connection/connection_mock.go b/pkg/connection/connection_mock.go
index 7490fff7..dca29e49 100644
--- a/pkg/connection/connection_mock.go
+++ b/pkg/connection/connection_mock.go
@@ -38,8 +38,8 @@ func (m *mockConnection) Ping(ctx api.StreamContext) error {
 	return nil
 }
 
-func (m *mockConnection) Close(ctx api.StreamContext) {
-	return
+func (m *mockConnection) Close(ctx api.StreamContext) error {
+	return nil
 }
 
 func (m *mockConnection) Attach(ctx api.StreamContext) {
diff --git a/pkg/model/io.go b/pkg/model/io.go
index fc23b255..54407ce1 100644
--- a/pkg/model/io.go
+++ b/pkg/model/io.go
@@ -41,3 +41,7 @@ type NodeInfo struct {
 	HasCompress     bool
 	HasInterval     bool
 }
+
+type UniqueSub interface {
+	SubId(props map[string]any) string
+}
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
index d2e1fb13..9ff3361e 100644
--- a/pkg/modules/connection.go
+++ b/pkg/modules/connection.go
@@ -18,8 +18,8 @@ import "github.com/lf-edge/ekuiper/contract/v2/api"
 
 type Connection interface {
 	Ping(ctx api.StreamContext) error
-	Close(ctx api.StreamContext)
 	DetachSub(ctx api.StreamContext, props map[string]any)
+	api.Closable
 }
 
 type ConnectionProvider func(ctx api.StreamContext, props map[string]any) (Connection, error)
diff --git a/pkg/nng/sock.go b/pkg/nng/sock.go
new file mode 100644
index 00000000..1eae58f5
--- /dev/null
+++ b/pkg/nng/sock.go
@@ -0,0 +1,152 @@
+// Copyright 2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package nng
+
+import (
+	"fmt"
+	"net/url"
+	"sync/atomic"
+	"time"
+
+	"go.nanomsg.org/mangos/v3"
+	"go.nanomsg.org/mangos/v3/protocol/pair"
+	"go.nanomsg.org/mangos/v3/protocol/push"
+	"go.nanomsg.org/mangos/v3/protocol/req"
+	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+	_ "go.nanomsg.org/mangos/v3/transport/tcp"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+type SockConf struct {
+	Url      string `json:"url"`
+	Protocol string `json:"protocol"`
+}
+
+type Sock struct {
+	mangos.Socket
+	url       string
+	connected atomic.Bool
+}
+
+var nngTimeout = 5 * time.Second
+
+func CreateConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	c, err := ValidateConf(props)
+	if err != nil {
+		return nil, err
+	}
+	var sock mangos.Socket
+	switch c.Protocol {
+	case "pair":
+		sock, err = pair.NewSocket()
+	case "push":
+		sock, err = push.NewSocket()
+	case "req":
+		sock, err = req.NewSocket()
+	default:
+		return nil, fmt.Errorf("unsupported nng protocol %s", c.Protocol)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+	// options consider to export
+	_ = sock.SetOption(mangos.OptionSendDeadline, nngTimeout)
+	_ = sock.SetOption(mangos.OptionRecvDeadline, nngTimeout)
+	cli := &Sock{
+		url:       c.Url,
+		Socket:    sock,
+		connected: atomic.Bool{},
+	}
+	sock.SetPipeEventHook(func(ev mangos.PipeEvent, p mangos.Pipe) {
+		switch ev {
+		case mangos.PipeEventAttached:
+			cli.connected.Store(true)
+			ctx.GetLogger().Infof("nano connection attached")
+		case mangos.PipeEventAttaching:
+			ctx.GetLogger().Infof("nano connection is attaching")
+		case mangos.PipeEventDetached:
+			cli.connected.Store(false)
+			ctx.GetLogger().Warnf("nano connection detached")
+			// TODO how to let connection send error event?
+		}
+	})
+	// sock.SetOption(mangos.OptionWriteQLen, 100)
+	// sock.SetOption(mangos.OptionReadQLen, 100)
+	// sock.SetOption(mangos.OptionBestEffort, false)
+	if err = sock.DialOptions(c.Url, map[string]interface{}{
+		mangos.OptionDialAsynch:       true, // will not report error and keep connecting
+		mangos.OptionMaxReconnectTime: 5 * time.Second,
+		mangos.OptionReconnectTime:    100 * time.Millisecond,
+	}); err != nil {
+		return nil, fmt.Errorf("please make sure nng server side has started and configured, can't dial: %s", err.Error())
+	}
+	return cli, nil
+}
+
+func ValidateConf(props map[string]any) (*SockConf, error) {
+	c := &SockConf{
+		Protocol: "pair",
+	}
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return nil, err
+	}
+	if c.Url == "" {
+		return nil, fmt.Errorf("url is required")
+	} else {
+		// Parse the URL
+		parsedURL, err := url.Parse(c.Url)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing url %s: %s", c.Url, err)
+		}
+		if parsedURL.Scheme != "tcp" && parsedURL.Scheme != "ipc" {
+			return nil, fmt.Errorf("only tcp and ipc scheme are supported")
+		}
+	}
+	if c.Protocol != "pair" && c.Protocol != "push" && c.Protocol != "req" {
+		return nil, fmt.Errorf("unsupported protocol %s", c.Protocol)
+	}
+	return c, nil
+}
+
+func (s *Sock) Ping(_ api.StreamContext) error {
+	if !s.connected.Load() {
+		return fmt.Errorf("not connected")
+	}
+	return nil
+}
+
+func (s *Sock) Close(_ api.StreamContext) error {
+	return s.Socket.Close()
+}
+
+func (s *Sock) DetachSub(_ api.StreamContext, _ map[string]any) {
+	// do nothing
+}
+
+func (s *Sock) Send(ctx api.StreamContext, data []byte) error {
+	ctx.GetLogger().Debugf("ngg publish %x", data)
+	if s.Socket != nil && s.connected.Load() {
+		return s.Socket.Send(data)
+	}
+	return errorx.NewIOErr(`neuron connection is not established`)
+}
+
+var _ modules.Connection = &Sock{}


Test patch:
diff --git a/internal/io/neuron/neuron_test.go b/internal/io/neuron/neuron_test.go
new file mode 100644
index 00000000..1fb55c4e
--- /dev/null
+++ b/internal/io/neuron/neuron_test.go
@@ -0,0 +1,84 @@
+// Copyright 2022-2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"go.nanomsg.org/mangos/v3"
+	"go.nanomsg.org/mangos/v3/protocol/pair"
+	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+
+	"github.com/lf-edge/ekuiper/v2/internal/testx"
+	"github.com/lf-edge/ekuiper/v2/pkg/connection"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+	"github.com/lf-edge/ekuiper/v2/pkg/nng"
+)
+
+var data = [][]byte{
+	[]byte("{\"timestamp\": 1646125996000, \"node_name\": \"node1\", \"group_name\": \"group1\", \"values\": {\"tag_name1\": 11.22, \"tag_name2\": \"yellow\"}, \"errors\": {\"tag_name3\": 122}}"),
+	[]byte(`{"timestamp": 1646125996000, "node_name": "node1", "group_name": "group1", "values": {"tag_name1": 11.22, "tag_name2": "green","tag_name3":60}, "errors": {}}`),
+	[]byte(`{"timestamp": 1646125996000, "node_name": "node1", "group_name": "group1", "values": {"tag_name1": 15.4, "tag_name2": "green","tag_name3":70}, "errors": {}}`),
+}
+
+func init() {
+	connection.InitConnectionManager4Test()
+	testx.InitEnv("neuron_tests")
+	modules.RegisterConnection("nng", nng.CreateConnection)
+}
+
+// mockNeuron start the nng pair server
+func mockNeuron(send bool, recv bool, url string) (mangos.Socket, chan []byte) {
+	var (
+		sock mangos.Socket
+		err  error
+		ch   chan []byte
+	)
+	if sock, err = pair.NewSocket(); err != nil {
+		log.Fatalf("can't get new pair socket: %s", err)
+	}
+	if err = sock.Listen(url); err != nil {
+		log.Fatalf("can't listen on pair socket: %s", err.Error())
+	}
+	log.Printf("listen on pair socket")
+	if recv {
+		ch = make(chan []byte)
+		go func() {
+			for {
+				var msg []byte
+				var err error
+				if msg, err = sock.Recv(); err == nil {
+					fmt.Printf("Neuron RECEIVED: \"%s\"\n", string(msg))
+					ch <- msg
+					fmt.Println("Neuron Sent out")
+				}
+			}
+		}()
+	}
+	if send {
+		go func() {
+			for _, msg := range data {
+				time.Sleep(10 * time.Millisecond)
+				fmt.Printf("Neuron SENDING \"%s\"\n", msg)
+				if err := sock.Send(msg); err != nil {
+					fmt.Printf("failed sending: %s\n", err)
+				}
+			}
+		}()
+	}
+	return sock, ch
+}
diff --git a/internal/io/neuron/sink_test.go b/internal/io/neuron/sink_test.go
new file mode 100644
index 00000000..70713928
--- /dev/null
+++ b/internal/io/neuron/sink_test.go
@@ -0,0 +1,248 @@
+// Copyright 2022-2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+)
+
+func TestSink(t *testing.T) {
+	server, ch := mockNeuron(false, true, DefaultNeuronUrl)
+	defer server.Close()
+
+	s := GetSink().(api.TupleCollector)
+	ctx := mockContext.NewMockContext("t", "tt")
+	err := s.(*sink).Ping(ctx, map[string]any{"url": DefaultNeuronUrl})
+	assert.NoError(t, err)
+	data := []any{
+		&xsql.Tuple{
+			Message: map[string]any{
+				"temperature": 22,
+				"humidity":    50,
+				"status":      "green",
+			},
+			Props: map[string]string{
+				"test1": "dynamic",
+				"grp":   "dynamic",
+			},
+		},
+		&xsql.WindowTuples{
+			Content: []xsql.Row{
+				&xsql.Tuple{
+					Message: map[string]any{
+						"temperature": 25,
+						"humidity":    82,
+						"status":      "wet",
+					},
+				},
+				&xsql.Tuple{
+					Message: map[string]any{
+						"temperature": 33,
+						"humidity":    60,
+						"status":      "hot",
+					},
+				},
+			},
+		},
+	}
+	err = mock.RunTupleSinkCollect(s, data, map[string]any{
+		"url":       DefaultNeuronUrl,
+		"nodeName":  "test1",
+		"groupName": "grp",
+		"tags":      []string{"temperature", "status"},
+		"raw":       false,
+	})
+	assert.NoError(t, err)
+
+	exp := []string{
+		`{"group_name":"dynamic","node_name":"dynamic","tag_name":"temperature","value":22}`,
+		`{"group_name":"dynamic","node_name":"dynamic","tag_name":"status","value":"green"}`,
+		`{"group_name":"grp","node_name":"test1","tag_name":"temperature","value":25}`,
+		`{"group_name":"grp","node_name":"test1","tag_name":"status","value":"wet"}`,
+		`{"group_name":"grp","node_name":"test1","tag_name":"temperature","value":33}`,
+		`{"group_name":"grp","node_name":"test1","tag_name":"status","value":"hot"}`,
+	}
+	var actual []string
+	ticker := time.After(5 * time.Second)
+	for i := 0; i < len(exp); i++ {
+		select {
+		case <-ticker:
+			t.Errorf("timeout")
+			return
+		case d := <-ch:
+			actual = append(actual, string(d))
+		}
+	}
+
+	assert.Equal(t, exp, actual)
+}
+
+func TestSinkNoTags(t *testing.T) {
+	server, ch := mockNeuron(false, true, DefaultNeuronUrl)
+	defer server.Close()
+
+	s := GetSink().(api.TupleCollector)
+	ctx := mockContext.NewMockContext("t", "tt")
+	err := s.(*sink).Ping(ctx, map[string]any{"url": DefaultNeuronUrl})
+	assert.NoError(t, err)
+	data := []any{
+		&xsql.Tuple{
+			Message: map[string]any{
+				"temperature": 22,
+				"humidity":    50,
+				"status":      "green",
+			},
+			Props: map[string]string{
+				"test1": "dynamic",
+				"grp":   "dynamic",
+			},
+		},
+	}
+	err = mock.RunTupleSinkCollect(s, data, map[string]any{
+		"url":       DefaultNeuronUrl,
+		"nodeName":  "test1",
+		"groupName": "grp",
+		"raw":       false,
+	})
+	assert.NoError(t, err)
+
+	exp := []string{
+		`{"group_name":"dynamic","node_name":"dynamic","tag_name":"humidity","value":50}`,
+		`{"group_name":"dynamic","node_name":"dynamic","tag_name":"status","value":"green"}`,
+		`{"group_name":"dynamic","node_name":"dynamic","tag_name":"temperature","value":22}`,
+	}
+	var actual []string
+	ticker := time.After(5 * time.Second)
+	for i := 0; i < len(exp); i++ {
+		select {
+		case <-ticker:
+			t.Errorf("timeout")
+			return
+		case d := <-ch:
+			actual = append(actual, string(d))
+		}
+	}
+
+	assert.Equal(t, exp, actual)
+}
+
+func TestSinkRaw(t *testing.T) {
+	server, ch := mockNeuron(false, true, DefaultNeuronUrl)
+	defer server.Close()
+
+	s := GetSink().(api.TupleCollector)
+	data := []any{
+		&xsql.Tuple{
+			Message: map[string]any{
+				"temperature": 22,
+				"humidity":    50,
+				"status":      "green",
+			},
+		},
+		&xsql.Tuple{
+			Message: map[string]any{
+				"temperature": 25,
+				"humidity":    82,
+				"status":      "wet",
+			},
+		},
+		&xsql.Tuple{
+			Message: map[string]any{
+				"temperature": 33,
+				"humidity":    60,
+				"status":      "hot",
+			},
+		},
+	}
+	err := mock.RunTupleSinkCollect(s, data, map[string]any{
+		"url": DefaultNeuronUrl,
+		"raw": true,
+	})
+	assert.NoError(t, err)
+
+	exp := []string{
+		`{"humidity":50,"status":"green","temperature":22}`,
+		`{"humidity":82,"status":"wet","temperature":25}`,
+		`{"humidity":60,"status":"hot","temperature":33}`,
+	}
+	var actual []string
+	ticker := time.After(5 * time.Second)
+	for i := 0; i < len(exp); i++ {
+		select {
+		case <-ticker:
+			t.Errorf("timeout")
+			return
+		case d := <-ch:
+			actual = append(actual, string(d))
+		}
+	}
+
+	assert.Equal(t, exp, actual)
+	time.Sleep(100 * time.Millisecond)
+}
+
+func TestSinkProvision(t *testing.T) {
+	ctx := mockContext.NewMockContext("t", "tt")
+	s := &sink{}
+	err := s.Provision(ctx, map[string]any{
+		"url": "3434",
+	})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "only tcp and ipc scheme are supported")
+
+	err = s.Provision(ctx, map[string]any{
+		"url":  "tcp://127.0.0.1:8000",
+		"tags": "tag2",
+	})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "1 error(s) decoding:\n\n* 'tags': source data must be an array or slice, got string")
+
+	err = s.Provision(ctx, map[string]any{
+		"url": "tcp://127.0.0.1:8000",
+	})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "node name is required if raw is not set")
+
+	err = s.Provision(ctx, map[string]any{
+		"url":      "tcp://127.0.0.1:8000",
+		"nodeName": "test",
+	})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "group name is required if raw is not set")
+
+	err = s.Provision(ctx, map[string]any{
+		"url": "tcp://127.0.0.1:8000",
+		"raw": true,
+	})
+	assert.NoError(t, err)
+
+	err = s.Connect(ctx)
+	assert.NoError(t, err)
+
+	err = s.Connect(ctx)
+	assert.Error(t, err)
+	assert.EqualError(t, err, "connection pairtcp://127.0.0.1:8000 already been created")
+
+	err = s.Close(ctx)
+	assert.NoError(t, err)
+}
diff --git a/internal/io/neuron/source_test.go b/internal/io/neuron/source_test.go
new file mode 100644
index 00000000..fd8c7244
--- /dev/null
+++ b/internal/io/neuron/source_test.go
@@ -0,0 +1,81 @@
+// Copyright 2022-2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package neuron
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	_ "go.nanomsg.org/mangos/v3/transport/ipc"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/mock"
+	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/model"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+func TestRun(t *testing.T) {
+	exp := []api.MessageTuple{
+		model.NewDefaultRawTuple([]byte("{\"timestamp\": 1646125996000, \"node_name\": \"node1\", \"group_name\": \"group1\", \"values\": {\"tag_name1\": 11.22, \"tag_name2\": \"yellow\"}, \"errors\": {\"tag_name3\": 122}}"), nil, timex.GetNow()),
+		model.NewDefaultRawTuple([]byte(`{"timestamp": 1646125996000, "node_name": "node1", "group_name": "group1", "values": {"tag_name1": 11.22, "tag_name2": "green","tag_name3":60}, "errors": {}}`), nil, timex.GetNow()),
+		model.NewDefaultRawTuple([]byte(`{"timestamp": 1646125996000, "node_name": "node1", "group_name": "group1", "values": {"tag_name1": 15.4, "tag_name2": "green","tag_name3":70}, "errors": {}}`), nil, timex.GetNow()),
+	}
+	s := GetSource()
+	server, _ := mockNeuron(true, false, DefaultNeuronUrl)
+	defer server.Close()
+	mock.TestSourceConnector(t, s, map[string]any{
+		"datasource": "new",
+		"url":        DefaultNeuronUrl,
+	}, exp, func() {
+		// do nothing
+	})
+
+	ctx := mockContext.NewMockContext("t", "tt")
+	err := s.(*source).Ping(ctx, map[string]any{"url": DefaultNeuronUrl})
+	assert.NoError(t, err)
+}
+
+func TestProvision(t *testing.T) {
+	ctx := mockContext.NewMockContext("t", "tt")
+	s := GetSource()
+	err := s.Provision(ctx, map[string]any{
+		"url": "3434",
+	})
+	assert.Error(t, err)
+	assert.EqualError(t, err, "only tcp and ipc scheme are supported")
+
+	err = s.Provision(ctx, map[string]any{
+		"url": "tcp://127.0.0.1:8000",
+	})
+	assert.NoError(t, err)
+
+	su, ok := s.(model.UniqueSub)
+	assert.True(t, ok)
+	sid := su.SubId(map[string]any{
+		"url": "tcp://127.0.0.1:8000",
+	})
+	assert.Equal(t, "nng:pairtcp://127.0.0.1:8000", sid)
+
+	err = s.Connect(ctx)
+	assert.NoError(t, err)
+
+	err = s.Connect(ctx)
+	assert.Error(t, err)
+	assert.EqualError(t, err, "connection pairtcp://127.0.0.1:8000 already been created")
+
+	err = s.Close(ctx)
+	assert.NoError(t, err)
+}
diff --git a/internal/topo/planner/planner_source_test.go b/internal/topo/planner/planner_source_test.go
index 9b5c8bb3..8731dd62 100644
--- a/internal/topo/planner/planner_source_test.go
+++ b/internal/topo/planner/planner_source_test.go
@@ -56,6 +56,7 @@ func TestPlanTopo(t *testing.T) {
 		"filesrc1": `CREATE STREAM fs1 () WITH (FORMAT="json", TYPE="file",CONF_KEY="lines");`,
 		"filesrc2": `CREATE STREAM fs2 () WITH (FORMAT="delimited", TYPE="file",CONF_KEY="csv");`,
 		"filesrc3": `CREATE STREAM fs3 () WITH (FORMAT="json",TYPE="file",CONF_KEY="json");`,
+		"neuron1":  `CREATE STREAM neuron1 () WITH (FORMAT="json", TYPE="neuron",CONF_KEY="tcp");`,
 	}
 	for name, sql := range streamSqls {
 		s, err := json.Marshal(&xsql.StreamInfo{
@@ -124,6 +125,13 @@ func TestPlanTopo(t *testing.T) {
 			p: "file",
 			k: "json",
 		},
+		{
+			conf: map[string]any{
+				"url": "tcp://127.0.0.1:7777",
+			},
+			p: "neuron",
+			k: "tcp",
+		},
 	}
 	meta.InitYamlConfigManager()
 	dataDir, _ := conf.GetDataLoc()
@@ -352,6 +360,30 @@ func TestPlanTopo(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "testNngConnSplit",
+			sql:  `SELECT * FROM neuron1`,
+			topo: &def.PrintableTopo{
+				Sources: []string{"source_nng:pairtcp://127.0.0.1:7777"},
+				Edges: map[string][]any{
+					"source_nng:pairtcp://127.0.0.1:7777": {
+						"op_2_decoder",
+					},
+					"op_2_decoder": {
+						"op_3_project",
+					},
+					"op_3_project": {
+						"op_logToMemory_0_0_transform",
+					},
+					"op_logToMemory_0_0_transform": {
+						"op_logToMemory_0_1_encode",
+					},
+					"op_logToMemory_0_1_encode": {
+						"sink_logToMemory_0",
+					},
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/pkg/mock/test_sink.go b/pkg/mock/test_sink.go
index f4007fb1..cc69864f 100644
--- a/pkg/mock/test_sink.go
+++ b/pkg/mock/test_sink.go
@@ -59,10 +59,10 @@ func RunTupleSinkCollect(s api.TupleCollector, data []any, props map[string]any)
 	time.Sleep(time.Second)
 	for _, e := range data {
 		switch ee := e.(type) {
-		case api.MessageTuple:
-			err = s.Collect(ctx, ee)
 		case api.MessageTupleList:
 			err = s.CollectList(ctx, ee)
+		case api.MessageTuple:
+			err = s.Collect(ctx, ee)
 		default:
 			err = errors.New("unsupported data type")
 		}
diff --git a/pkg/mock/test_source.go b/pkg/mock/test_source.go
index f998a62c..fb366d86 100644
--- a/pkg/mock/test_source.go
+++ b/pkg/mock/test_source.go
@@ -16,6 +16,7 @@ package mock
 
 import (
 	"fmt"
+	"log"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -24,6 +25,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/internal/xsql"
 	"github.com/lf-edge/ekuiper/v2/pkg/cast"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
 	"github.com/lf-edge/ekuiper/v2/pkg/model"
@@ -97,6 +99,8 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 			}
 		case map[string]any:
 			result = append(result, model.NewDefaultSourceTuple(mt, meta, ts))
+		case xsql.Message:
+			result = append(result, model.NewDefaultSourceTuple(mt, meta, ts))
 		default:
 			panic("not supported yet")
 		}
@@ -109,7 +113,7 @@ func TestSourceConnector(t *testing.T, r api.Source, props map[string]any, expec
 		switch ss := r.(type) {
 		case api.BytesSource:
 			err = ss.Subscribe(ctx, ingestBytes, func(ctx api.StreamContext, err error) {
-				panic(err)
+				log.Println(err)
 			})
 		case api.TupleSource:
 			err = ss.Subscribe(ctx, ingestTuples, func(ctx api.StreamContext, err error) {
diff --git a/pkg/nng/sock_test.go b/pkg/nng/sock_test.go
new file mode 100644
index 00000000..4d4e132f
--- /dev/null
+++ b/pkg/nng/sock_test.go
@@ -0,0 +1,64 @@
+// Copyright 2024 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package nng
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name  string
+		props map[string]any
+		err   string
+	}{
+		{
+			name: "wrong format",
+			props: map[string]any{
+				"url": 345,
+			},
+			err: "1 error(s) decoding:\n\n* 'url' expected type 'string', got unconvertible type 'int', value: '345'",
+		},
+		{
+			name:  "missing url",
+			props: map[string]any{},
+			err:   "url is required",
+		},
+		{
+			name: "wrong url",
+			props: map[string]any{
+				"url": "file:////abc",
+			},
+			err: "only tcp and ipc scheme are supported",
+		},
+		{
+			name: "wrong protocol",
+			props: map[string]any{
+				"url":      "tcp://127.0.0.1:444",
+				"protocol": "pair1",
+			},
+			err: "unsupported protocol pair1",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, e := ValidateConf(tt.props)
+			assert.Error(t, e)
+			assert.EqualError(t, e, tt.err)
+		})
+	}
+}