feat(io): restore edgex source/sink

etc/sources/edgex.yaml

@@ -56,5 +56,4 @@ share_conf: #Conf_key
   server: 127.0.0.1
   port: 1883
   topic: events
-  type: mqtt
-  connectionSelector: edgex.redisMsgBus
+  connectionSelector: redisMsgBus

go.mod

@@ -14,15 +14,16 @@ require (
 	github.com/eclipse/paho.mqtt.golang v1.4.3
 	github.com/edgexfoundry/go-mod-core-contracts/v3 v3.1.0
 	github.com/edgexfoundry/go-mod-messaging/v3 v3.1.0
+	github.com/fxamacker/cbor/v2 v2.6.0
 	github.com/gdexlab/go-render v1.0.1
 	github.com/go-sql-driver/mysql v1.8.1
 	github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
 	github.com/golang-jwt/jwt/v5 v5.2.1
 	github.com/golang/protobuf v1.5.4
-	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.1
+	github.com/gorilla/websocket v1.5.1
 	github.com/jhump/protoreflect v1.15.6
 	github.com/jinzhu/now v1.1.5
 	github.com/keepeye/logrus-filename v0.0.0-20190711075016-ce01a4391dd1
@@ -80,7 +81,6 @@ require (
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
-	github.com/fxamacker/cbor/v2 v2.6.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/go-kit/kit v0.10.0 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
@@ -89,7 +89,7 @@ require (
 	github.com/go-playground/validator/v10 v10.19.0 // indirect
 	github.com/go-redis/redis/v7 v7.4.1 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/gorilla/websocket v1.5.1 // indirect
+	github.com/google/pprof v0.0.0-20240528025155-186aa0362fba // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
@@ -127,6 +127,7 @@ require (
 	github.com/tklauser/go-sysconf v0.3.13 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/u2takey/go-utils v0.3.1 // indirect
+	github.com/x448/float16 v0.8.4 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/otel v1.7.0 // indirect

internal/binder/io/ext_edgex.go

@@ -0,0 +1,29 @@
+// 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.
+
+//go:build edgex || full
+
+package io
+
+import (
+	"github.com/lf-edge/ekuiper/v2/internal/io/edgex"
+	edgexCon "github.com/lf-edge/ekuiper/v2/internal/io/edgex/client"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
+)
+
+func init() {
+	modules.RegisterConnection("edgex", edgexCon.GetConnection)
+	modules.RegisterSource("edgex", edgex.GetSource)
+	modules.RegisterSink("edgex", edgex.GetSink)
+}

internal/io/edgex/client/edgex.go

@@ -0,0 +1,189 @@
+// 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 client
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/edgexfoundry/go-mod-messaging/v3/messaging"
+	"github.com/edgexfoundry/go-mod-messaging/v3/pkg/types"
+
+	"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/modules"
+)
+
+type Client struct {
+	mbconf types.MessageBusConfig
+	client messaging.MessageClient
+}
+
+func (es *Client) Ping(_ api.StreamContext) error {
+	if es.client != nil {
+		return nil
+	}
+	return fmt.Errorf("client is nil")
+}
+
+func (es *Client) DetachSub(ctx api.StreamContext, props map[string]any) {
+	topic, ok := props["topic"]
+	ctx.GetLogger().Infof("detach edgex sub %v", topic)
+	if ok {
+		err := es.client.Unsubscribe(topic.(string))
+		if err != nil {
+			ctx.GetLogger().Error(err)
+		}
+	}
+}
+
+func (es *Client) Close(ctx api.StreamContext) error {
+	if es.client != nil {
+		return es.client.Disconnect()
+	}
+	return nil
+}
+
+var _ modules.Connection = &Client{}
+
+func GetConnection(ctx api.StreamContext, props map[string]any) (modules.Connection, error) {
+	ctx.GetLogger().Infof("connect to edgex")
+	c := &Client{}
+	err := c.CfgValidate(props)
+	if err != nil {
+		return nil, err
+	}
+	client, err := messaging.NewMessageClient(c.mbconf)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := client.Connect(); err != nil {
+		conf.Log.Errorf("The connection to edgex messagebus failed.")
+		return nil, fmt.Errorf("Failed to connect to edgex message bus: " + err.Error())
+	}
+	conf.Log.Infof("The connection to edgex messagebus is established successfully.")
+
+	c.client = client
+	return c, nil
+}
+
+type EdgexConf struct {
+	Protocol string            `json:"protocol"`
+	Server   string            `json:"server"`
+	Host     string            `json:"host"`
+	Port     int               `json:"port"`
+	Type     string            `json:"type"`
+	Optional map[string]string `json:"optional"`
+}
+
+// Modify the copied conf to print no password.
+func printConf(mbconf types.MessageBusConfig) {
+	printableOptional := make(map[string]string)
+	for k, v := range mbconf.Optional {
+		if strings.EqualFold(k, "password") {
+			printableOptional[k] = "*"
+		} else {
+			printableOptional[k] = v
+		}
+	}
+	mbconf.Optional = printableOptional
+	conf.Log.Infof("Use configuration for edgex messagebus %v", mbconf)
+}
+
+func (es *Client) CfgValidate(props map[string]interface{}) error {
+	edgeAddr := "localhost"
+	c := &EdgexConf{
+		Protocol: "redis",
+		Port:     6379,
+		Type:     messaging.Redis,
+		Optional: nil,
+	}
+
+	if o, ok := props["optional"]; ok {
+		switch ot := o.(type) {
+		case map[string]string:
+			c.Optional = ot
+		case map[string]interface{}:
+			c.Optional = make(map[string]string)
+			for k, v := range ot {
+				c.Optional[k] = fmt.Sprintf("%v", v)
+			}
+		default:
+			return fmt.Errorf("invalid optional config %v, must be a map", o)
+		}
+		delete(props, "optional")
+	}
+
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return fmt.Errorf("map config map to struct fail with error: %v", err)
+	}
+
+	if c.Host != "" {
+		edgeAddr = c.Host
+	} else if c.Server != "" {
+		edgeAddr = c.Server
+	}
+
+	if c.Type != messaging.MQTT && c.Type != messaging.Redis &&
+		c.Type != messaging.NatsCore && c.Type != messaging.NatsJetStream {
+		return fmt.Errorf("specified wrong type value %s", c.Type)
+	}
+	if c.Port < 0 {
+		return fmt.Errorf("specified wrong port value, expect positive integer but got %d", c.Port)
+	}
+
+	mbconf := types.MessageBusConfig{
+		Broker: types.HostInfo{
+			Host:     edgeAddr,
+			Port:     c.Port,
+			Protocol: c.Protocol,
+		},
+		Type: c.Type,
+	}
+	mbconf.Optional = c.Optional
+	es.mbconf = mbconf
+
+	printConf(mbconf)
+
+	return nil
+}
+
+func (es *Client) Publish(env types.MessageEnvelope, topic string) error {
+	if err := es.client.Publish(env, topic); err != nil {
+		conf.Log.Errorf("Publish to topic %s has error : %s.", topic, err.Error())
+		return fmt.Errorf("Failed to publish to edgex message bus: " + err.Error())
+	}
+	return nil
+}
+
+func (es *Client) Subscribe(msg chan types.MessageEnvelope, topic string, err chan error) error {
+	topics := []types.TopicChannel{{Topic: topic, Messages: msg}}
+	if err := es.client.Subscribe(topics, err); err != nil {
+		conf.Log.Errorf("Failed to subscribe to edgex messagebus with topic %s has error : %s.", topic, err.Error())
+		return err
+	}
+	return nil
+}
+
+func (es *Client) Disconnect() error {
+	conf.Log.Infof("Closing the connection to edgex messagebus.")
+	if e := es.client.Disconnect(); e != nil {
+		return e
+	}
+	return nil
+}