Repository: lf-edge/ekuiper
Pull request #900 (2021-07-20 03:39:22)

feat(edgex): support message type config


Patch:
diff --git a/deploy/docker/conf_util.go b/deploy/docker/conf_util.go
index 90e37b30..d72f2cf3 100644
--- a/deploy/docker/conf_util.go
+++ b/deploy/docker/conf_util.go
@@ -49,6 +49,7 @@ var file_keys_map = map[string]map[string]string{
 		"CERTPEMBLOCK":      "CertPEMBlock",
 		"KEYPEMBLOCK":       "KeyPEMBlock",
 		"SKIPCERTVERIFY":    "SkipCertVerify",
+		"MESSAGETYPE":       "messageType",
 	},
 	"mqtt_source": {
 		"SHAREDSUBSCRIPTION": "sharedSubscription",
diff --git a/docs/en_US/rules/sources/edgex.md b/docs/en_US/rules/sources/edgex.md
index b3c98e97..af12890b 100644
--- a/docs/en_US/rules/sources/edgex.md
+++ b/docs/en_US/rules/sources/edgex.md
@@ -64,6 +64,7 @@ default:
   server: localhost
   port: 5573
   topic: events
+  messageType: event
 #  optional:
 #    ClientId: client1
 #    Username: user1
@@ -88,19 +89,33 @@ The port of EdgeX message bus, default value is ``5573``.
 
 ### topic
 
-The topic name of EdgeX message bus,  default value is ``events``.
+The topic name of EdgeX message bus, default value is ``events``. Users can subscribe to the topics of message bus
+directly or subscribe to topics exported by EdgeX application service. Notice that, the message type of the two types of
+topics are different, remember to set the appropriate messageType property.
 
 ### type
 
-The EdgeX message bus type, currently two types of message buses are supported. If specified other values, then will use the default ``zero`` value.
+The EdgeX message bus type, currently three types of message buses are supported. If specified other values, then will
+use the default ``redis`` value.
 
-- ``zero``: Use ZeroMQ as EdgeX message bus. 
+- ``zero``: Use ZeroMQ as EdgeX message bus.
 - ``mqtt``: Use the MQTT broker as EdgeX message bus.
 - ``redis``: Use Redis as EdgeX message bus. When using EdgeX docker compose, the type will be set to this by default.
 
+### messageType
+
+The EdgeX message model type. If connected to the topic of EdgeX application service, the message model is an "event".
+Otherwise, if connected to the topic of EdgeX message bus directly to receive the message from device service or core
+data, the message is a "request". There are two available types of messageType property:
+
+- ``event``: The message will be decoded as a `dtos.Event` type. This is the default.
+- ``request``: The message will be decoded as a `requests.AddEventRequest` type.
+
 ### optional
 
-If MQTT message bus is used, some other optional configurations can be specified. Please notice that all of values in optional are **<u>string type</u>**, so values for these configurations should be string - such as ``KeepAlive: "5000"``. Below optional configurations are supported, please check MQTT specification for the detailed information.
+If MQTT message bus is used, some other optional configurations can be specified. Please notice that all of values in
+optional are **<u>string type</u>**, so values for these configurations should be string - such as ``KeepAlive: "5000"``
+. Below optional configurations are supported, please check MQTT specification for the detailed information.
 
 - ClientId
 
diff --git a/docs/zh_CN/rules/sources/edgex.md b/docs/zh_CN/rules/sources/edgex.md
index 155df2e7..7b06d6d1 100644
--- a/docs/zh_CN/rules/sources/edgex.md
+++ b/docs/zh_CN/rules/sources/edgex.md
@@ -66,6 +66,7 @@ default:
   server: localhost
   port: 5573
   topic: events
+  messageType: event
 #  optional:
 #    ClientId: client1
 #    Username: user1
@@ -88,19 +89,29 @@ EdgeX 消息总线的端口，缺省为 `5573`
 
 ## topic
 
-EdgeX 消息总线上监听的主题名称，缺省为 `events`
+EdgeX 消息总线上监听的主题名称，缺省为 `events`。用户可以直接连接到 EdgeX 消息总线上的主题也可以连接到 application service 暴露的主题。需要注意的是，两种主题的消息数据类型不同，需要设置正确的
+messageType 类型。
 
 ## type
 
-EdgeX 消息总线类型，目前支持两种消息总线。如果指定了错误的消息总线类型，那么会使用缺省 `zero` 类型。
+EdgeX 消息总线类型，目前支持三种消息总线。如果指定了错误的消息总线类型，那么会使用缺省 `zero` 类型。
 
-- `zero`：使用 ZeroMQ 类型的消息总线 
+- `zero`：使用 ZeroMQ 类型的消息总线
 - `mqtt`：使用 MQTT 服务器作为消息总线
 - `redis`: 使用 Redis 服务器作为消息总线。使用 EdgeX docker compose 启动时，type参数会默认设置为该类型。
 
+### messageType
+
+EdgeX 消息模型类型。如果连接到 EdgeX application service 的 topic, 则消息为 "event" 类型。否则，如果直接连接到消息总线的 topic，接收到 device service 或者 core
+data 发出的数据，则消息类型为 "request"。该参数支持两种类型：
+
+- ``event``: 消息将会解码为 `dtos.Event` 类型。该选项为默认值。
+- ``request``: 消息将会解码为 `requests.AddEventRequest` 类型。
+
 ## optional
 
-如果使用了 MQTT 消息总线，还可以指定别的一些可选配置项。请注意，所有在可选的配置项里指定的值都必须为**<u>字符类型</u>**，因此这里出现的所有的配置应该是字符类型的 - 例如 `KeepAlive: "5000"`。以下为支持的可选的配置列表，您可以参考 MQTT 协议规范来获取更详尽的信息。
+如果使用了 MQTT 消息总线，还可以指定别的一些可选配置项。请注意，所有在可选的配置项里指定的值都必须为**<u>字符类型</u>**，因此这里出现的所有的配置应该是字符类型的 - 例如 `KeepAlive: "5000"`
+。以下为支持的可选的配置列表，您可以参考 MQTT 协议规范来获取更详尽的信息。
 
 - ClientId
 
diff --git a/etc/sources/edgex.yaml b/etc/sources/edgex.yaml
index d9dde370..185d4314 100644
--- a/etc/sources/edgex.yaml
+++ b/etc/sources/edgex.yaml
@@ -4,6 +4,11 @@ default:
   server: localhost
   port: 5563
   topic: events
+  type: zero
+  # Could be 'event' or 'request'.
+  # If the message is from app service, the message type is an event;
+  # Otherwise, if it is from the message bus directly, it should be a request
+  messageType: event
 #  Below is optional configurations settings for mqtt
 #  type: mqtt
 #  optional:
@@ -26,6 +31,7 @@ application_conf: #Conf_key
   server: localhost
   port: 5571
   topic: application
+  messageType: request
 
 mqtt_conf: #Conf_key
   protocol: tcp
diff --git a/internal/topo/source/edgex_source.go b/internal/topo/source/edgex_source.go
index bf0c37cb..1500ee9f 100644
--- a/internal/topo/source/edgex_source.go
+++ b/internal/topo/source/edgex_source.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	v2 "github.com/edgexfoundry/go-mod-core-contracts/v2/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
 	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
 	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -32,65 +33,62 @@ import (
 )
 
 type EdgexSource struct {
-	client     messaging.MessageClient
-	subscribed bool
-	topic      string
+	client      messaging.MessageClient
+	subscribed  bool
+	topic       string
+	messageType messageType
 }
 
+type EdgexConf struct {
+	Format      string            `json:"format"`
+	Protocol    string            `json:"protocol"`
+	Server      string            `json:"server"`
+	Port        int               `json:"port"`
+	Topic       string            `json:"topic"`
+	Type        string            `json:"type"`
+	MessageType messageType       `json:"messageType"`
+	Optional    map[string]string `json:"optional"`
+}
+
+type messageType string
+
+const (
+	MessageTypeEvent   messageType = "event"
+	MessageTypeRequest messageType = "request"
+)
+
 func (es *EdgexSource) Configure(_ string, props map[string]interface{}) error {
-	if f, ok := props["format"]; ok {
-		if f != message.FormatJson {
-			return fmt.Errorf("edgex source only supports `json` format")
-		}
-	}
-	var protocol = "tcp"
-	if p, ok := props["protocol"]; ok {
-		protocol = p.(string)
+	c := &EdgexConf{
+		Format:      message.FormatJson,
+		Protocol:    "tcp",
+		Server:      "localhost",
+		Port:        5563,
+		Type:        messaging.Redis,
+		MessageType: MessageTypeEvent,
 	}
-	var server = "localhost"
-	if s, ok := props["server"]; ok {
-		server = s.(string)
+	err := cast.MapToStruct(props, c)
+	if err != nil {
+		return fmt.Errorf("read properties %v fail with error: %v", props, err)
 	}
-	var port = 5563
-	if p, ok := props["port"]; ok {
-		port = p.(int)
+	if c.Format != message.FormatJson {
+		return fmt.Errorf("edgex source only supports `json` format")
 	}
 
-	if tpc, ok := props["topic"]; ok {
-		es.topic = tpc.(string)
-	}
-
-	var mbusType = messaging.ZeroMQ
-	if t, ok := props["type"]; ok {
-		mbusType = t.(string)
-		if mbusType != messaging.ZeroMQ && mbusType != messaging.MQTT && mbusType != messaging.Redis {
-			return fmt.Errorf("Specified wrong message type value %s, will use zeromq messagebus.\n", mbusType)
-		}
+	if c.Type != messaging.ZeroMQ && c.Type != messaging.MQTT && c.Type != messaging.Redis {
+		return fmt.Errorf("Specified wrong message type value %s, will use zeromq messagebus.\n", c.Type)
 	}
 
-	mbconf := types.MessageBusConfig{SubscribeHost: types.HostInfo{Protocol: protocol, Host: server, Port: port}, Type: mbusType}
-
-	var optional = make(map[string]string)
-	if ops, ok := props["optional"]; ok {
-		if ops1, ok1 := ops.(map[string]interface{}); ok1 {
-			for k, v := range ops1 {
-				if cv, err := cast.ToString(v, cast.CONVERT_ALL); err == nil {
-					optional[k] = cv
-				} else {
-					conf.Log.Infof("Cannot convert configuration %s: %s to string type.\n", k, v)
-				}
-			}
-		}
-		mbconf.Optional = optional
-	}
+	mbconf := types.MessageBusConfig{SubscribeHost: types.HostInfo{Protocol: c.Protocol, Host: c.Server, Port: c.Port}, Type: c.Type}
+	mbconf.Optional = c.Optional
 	printConf(mbconf)
 	if client, err := messaging.NewMessageClient(mbconf); err != nil {
 		return err
 	} else {
 		es.client = client
+		es.messageType = c.MessageType
+		es.topic = c.Topic
 		return nil
 	}
-
 }
 
 // Modify the copied conf to print no password.
@@ -135,8 +133,14 @@ func (es *EdgexSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTup
 					return
 				}
 				if strings.ToLower(env.ContentType) == "application/json" {
-					e := &dtos.Event{}
-					if err := json.Unmarshal(env.Payload, e); err != nil {
+					var r interface{}
+					switch es.messageType {
+					case MessageTypeEvent:
+						r = &dtos.Event{}
+					case MessageTypeRequest:
+						r = &requests.AddEventRequest{}
+					}
+					if err := json.Unmarshal(env.Payload, r); err != nil {
 						l := len(env.Payload)
 						if l > 200 {
 							l = 200
@@ -145,6 +149,13 @@ func (es *EdgexSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTup
 					} else {
 						result := make(map[string]interface{})
 						meta := make(map[string]interface{})
+						var e *dtos.Event
+						switch t := r.(type) {
+						case *dtos.Event:
+							e = t
+						case *requests.AddEventRequest:
+							e = &t.Event
+						}
 
 						log.Debugf("receive message %s from device %s", env.Payload, e.DeviceName)
 						for _, r := range e.Readings {


Test patch:
diff --git a/deploy/docker/conf_util_test.go b/deploy/docker/conf_util_test.go
index d7f7e107..26af50e1 100644
--- a/deploy/docker/conf_util_test.go
+++ b/deploy/docker/conf_util_test.go
@@ -191,6 +191,7 @@ func TestProcessEnv(t *testing.T) {
 		{
 			vars: []string{
 				"EDGEX__DEFAULT__TYPE=zmq",
+				"EDGEX__DEFAULT__MESSAGETYPE=event",
 				"EDGEX__DEFAULT__OPTIONAL__CLIENTID=clientid_0000",
 				"EDGEX__DEFAULT__OPTIONAL__PASSWORD=should_not_print",
 				"EDGEX__APPLICATION_CONF__PROTOCOL=ssl",
@@ -198,8 +199,9 @@ func TestProcessEnv(t *testing.T) {
 			file: "edgex",
 			expt: map[string]interface{}{
 				"default": map[string]interface{}{
-					"protocol": "tcp",
-					"type":     "zmq",
+					"messageType": "event",
+					"protocol":    "tcp",
+					"type":        "zmq",
 					"optional": map[string]interface{}{
 						"ClientId": "clientid_0000",
 						"Password": "should_not_print",
@@ -209,7 +211,7 @@ func TestProcessEnv(t *testing.T) {
 					"protocol": "ssl",
 				},
 			},
-			out: "application_conf:\n    protocol: ssl\ndefault:\n    optional:\n        ClientId: clientid_0000\n        Password: '*'\n    protocol: tcp\n    type: zmq\n",
+			out: "application_conf:\n    protocol: ssl\ndefault:\n    messageType: event\n    optional:\n        ClientId: clientid_0000\n        Password: '*'\n    protocol: tcp\n    type: zmq\n",
 		},
 	}
 	files := make(map[string]map[string]interface{})
diff --git a/test/edgex/pub.go b/test/edgex/pub.go
index e1b0b163..3ea135ef 100644
--- a/test/edgex/pub.go
+++ b/test/edgex/pub.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/common"
 	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos"
+	"github.com/edgexfoundry/go-mod-core-contracts/v2/dtos/requests"
 	"github.com/edgexfoundry/go-mod-messaging/v2/messaging"
 	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
 	"log"
@@ -54,33 +55,33 @@ func pubEventClientZeroMq() {
 				testEvent.Origin = 123
 				err := testEvent.AddSimpleReading("Temperature", common.ValueTypeInt64, int64(i*8))
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.Temperature: %v\n", i, i*8)
+					fmt.Printf("Add reading error for %d.Temperature: %v\n", i, i*8)
 				}
 				err = testEvent.AddSimpleReading("Humidity", common.ValueTypeInt64, int64(i*9))
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.Humidity: %v\n", i, i*9)
+					fmt.Printf("Add reading error for %d.Humidity: %v\n", i, i*9)
 				}
 				err = testEvent.AddSimpleReading("b1", common.ValueTypeBool, i%2 == 0)
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.b1: %v\n", i, i%2 == 0)
+					fmt.Printf("Add reading error for %d.b1: %v\n", i, i%2 == 0)
 				}
 				err = testEvent.AddSimpleReading("i1", common.ValueTypeInt64, int64(i))
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.i1: %v\n", i, i)
+					fmt.Printf("Add reading error for %d.i1: %v\n", i, i)
 				}
 				err = testEvent.AddSimpleReading("f1", common.ValueTypeFloat64, float64(i)/2.0)
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.f1: %v\n", i, float64(i)/2.0)
+					fmt.Printf("Add reading error for %d.f1: %v\n", i, float64(i)/2.0)
 				}
 				err = testEvent.AddSimpleReading("ui64", common.ValueTypeUint64, uint64(10796529505058023104))
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.ui64: %v\n", i, uint64(10796529505058023104))
+					fmt.Printf("Add reading error for %d.ui64: %v\n", i, uint64(10796529505058023104))
 				}
 
 				fmt.Printf("readings: %v\n", testEvent.Readings)
 				data, err := json.Marshal(testEvent)
 				if err != nil {
-					fmt.Errorf("unexpected error MarshalEvent %v", err)
+					fmt.Printf("unexpected error MarshalEvent %v", err)
 				} else {
 					fmt.Println(string(data))
 				}
@@ -119,16 +120,18 @@ func pubToAnother() {
 		testEvent.Origin = 123
 		err := testEvent.AddSimpleReading("Temperature", common.ValueTypeInt64, int64(20))
 		if err != nil {
-			fmt.Errorf("Add reading error for Temperature: %v\n", 20)
+			fmt.Printf("Add reading error for Temperature: %v\n", 20)
 		}
 		err = testEvent.AddSimpleReading("Humidity", common.ValueTypeInt64, int64(30))
 		if err != nil {
-			fmt.Errorf("Add reading error for Humidity: %v\n", 20)
+			fmt.Printf("Add reading error for Humidity: %v\n", 20)
 		}
 
-		data, err := json.Marshal(testEvent)
+		req := requests.NewAddEventRequest(testEvent)
+
+		data, err := json.Marshal(req)
 		if err != nil {
-			fmt.Errorf("unexpected error MarshalEvent %v", err)
+			fmt.Printf("unexpected error marshal request %v", err)
 		} else {
 			fmt.Println(string(data))
 		}
@@ -164,21 +167,21 @@ func pubArrayMessage() {
 		testEvent.Origin = 123
 		err := testEvent.AddSimpleReading("ba", common.ValueTypeBoolArray, []bool{true, true, false})
 		if err != nil {
-			fmt.Errorf("Add reading error for ba: %v\n", []bool{true, true, false})
+			fmt.Printf("Add reading error for ba: %v\n", []bool{true, true, false})
 		}
 		err = testEvent.AddSimpleReading("ia", common.ValueTypeInt32Array, []int32{30, 40, 50})
 		if err != nil {
-			fmt.Errorf("Add reading error for ia: %v\n", []int32{30, 40, 50})
+			fmt.Printf("Add reading error for ia: %v\n", []int32{30, 40, 50})
 		}
 		err = testEvent.AddSimpleReading("fa", common.ValueTypeFloat64Array, []float64{3.14, 3.1415, 3.1415926})
 		if err != nil {
-			fmt.Errorf("Add reading error for fa: %v\n", []float64{3.14, 3.1415, 3.1415926})
+			fmt.Printf("Add reading error for fa: %v\n", []float64{3.14, 3.1415, 3.1415926})
 		}
 		testEvent.Readings[len(testEvent.Readings)-1].Value = "[3.14, 3.1415, 3.1415926]"
 
 		data, err := json.Marshal(testEvent)
 		if err != nil {
-			fmt.Errorf("unexpected error MarshalEvent %v", err)
+			fmt.Printf("unexpected error MarshalEvent %v", err)
 		} else {
 			fmt.Println(string(data))
 		}
@@ -215,16 +218,16 @@ func pubToMQTT(host string) {
 		testEvent.Origin = 123
 		err := testEvent.AddSimpleReading("Temperature", common.ValueTypeInt64, int64(20))
 		if err != nil {
-			fmt.Errorf("Add reading error for Temperature: %v\n", 20)
+			fmt.Printf("Add reading error for Temperature: %v\n", 20)
 		}
 		err = testEvent.AddSimpleReading("Humidity", common.ValueTypeInt64, int64(30))
 		if err != nil {
-			fmt.Errorf("Add reading error for Humidity: %v\n", 20)
+			fmt.Printf("Add reading error for Humidity: %v\n", 20)
 		}
 
 		data, err := json.Marshal(testEvent)
 		if err != nil {
-			fmt.Errorf("unexpected error MarshalEvent %v", err)
+			fmt.Printf("unexpected error MarshalEvent %v", err)
 		} else {
 			fmt.Println(string(data))
 		}
@@ -255,13 +258,13 @@ func pubMetaSource() {
 				testEvent.Origin = 13 * j
 				err := testEvent.AddSimpleReading("Temperature", common.ValueTypeInt64, j*8)
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.Temperature: %v\n", i, j*8)
+					fmt.Printf("Add reading error for %d.Temperature: %v\n", i, j*8)
 				}
 				testEvent.Readings[0].Origin = 24 * j
 				testEvent.Readings[0].DeviceName = "Temperature sensor"
 				err = testEvent.AddSimpleReading("Humidity", common.ValueTypeInt64, j*8)
 				if err != nil {
-					fmt.Errorf("Add reading error for %d.Humidity: %v\n", i, j*8)
+					fmt.Printf("Add reading error for %d.Humidity: %v\n", i, j*8)
 				}
 				testEvent.Readings[1].Origin = 34 * j
 				testEvent.Readings[1].DeviceName = "Humidity sensor"
@@ -270,7 +273,7 @@ func pubMetaSource() {
 
 				data, err := json.Marshal(testEvent)
 				if err != nil {
-					fmt.Errorf("unexpected error MarshalEvent %v", err)
+					fmt.Printf("unexpected error MarshalEvent %v", err)
 				} else {
 					fmt.Println(string(data))
 				}