fvt testcases for edgex

etc/sources/edgex.yaml

@@ -4,6 +4,7 @@ default:
   server: localhost
   port: 5570
   topic: events
+  serviceServer: http://localhost:8080
 
 #Override the global configurations
 demo_conf: #Conf_key

fvt_scripts/edgex/pub.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients/coredata"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients/urlclient/local"
+	"github.com/edgexfoundry/go-mod-core-contracts/models"
+	"github.com/edgexfoundry/go-mod-messaging/messaging"
+	"github.com/edgexfoundry/go-mod-messaging/pkg/types"
+	"log"
+	"math/rand"
+	"time"
+)
+var msgConfig1 = types.MessageBusConfig{
+	PublishHost: types.HostInfo{
+		Host:     "*",
+		Port:     5570,
+		Protocol: "tcp",
+	},
+}
+
+func pubEventClientZeroMq() {
+	msgConfig1.Type = messaging.ZeroMQ
+	if msgClient, err := messaging.NewMessageClient(msgConfig1); err != nil {
+		log.Fatal(err)
+	} else {
+		if ec := msgClient.Connect(); ec != nil {
+			log.Fatal(ec)
+		} else {
+			client := coredata.NewEventClient(local.New("test"))
+			r := rand.New(rand.NewSource(time.Now().UnixNano()))
+			for i := 0; i < 10; i++ {
+				temp := r.Intn(100)
+				humd := r.Intn(100)
+
+				var testEvent = models.Event{Device: "demo", Created: 123, Modified: 123, Origin: 123}
+				var testReading1 = models.Reading{Pushed: 123, Created: 123, Origin: 123, Modified: 123, Device: "test device name",
+					Name: "Temperature", Value: fmt.Sprintf("%d", temp)}
+				var testReading2 = models.Reading{Pushed: 123, Created: 123, Origin: 123, Modified: 123, Device: "test device name",
+					Name: "Humidity", Value: fmt.Sprintf("%d", humd)}
+				testEvent.Readings = append(testEvent.Readings, testReading1, testReading2)
+
+				data, err := client.MarshalEvent(testEvent)
+				if err != nil {
+					fmt.Errorf("unexpected error MarshalEvent %v", err)
+				} else {
+					fmt.Println(string(data))
+				}
+
+				env := types.NewMessageEnvelope([]byte(data), context.Background())
+				env.ContentType = "application/json"
+
+				if e := msgClient.Publish(env, "events"); e != nil {
+					log.Fatal(e)
+				} else {
+					fmt.Printf("Pub successful: %s\n", data)
+				}
+				time.Sleep(1 * time.Second)
+			}
+
+
+		}
+	}
+}
+
+func main() {
+	pubEventClientZeroMq()
+}
+

fvt_scripts/edgex/vd_server.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients"
+	"github.com/edgexfoundry/go-mod-core-contracts/models"
+	"log"
+	"net/http"
+)
+
+const (
+	testValueDesciptorDescription1 = "Temperature descriptor1"
+	testValueDesciptorDescription2 = "Humidity descriptor2"
+)
+
+var testValueDescriptor1 = models.ValueDescriptor{Id: "Temperature", Created: 123, Modified: 123, Origin: 123, Name: "Temperature",
+	Description: "test description", Min: -70, Max: 140, DefaultValue: 32, Formatting: "%d", Type:"STRING",
+	Labels: []string{"temp", "room temp"}, UomLabel: "F", MediaType: clients.ContentTypeJSON, FloatEncoding: "eNotation"}
+
+var testValueDescriptor2 = models.ValueDescriptor{Id: "Humidity", Created: 123, Modified: 123, Origin: 123, Name: "Humidity",
+	Description: "test description", Min: -70, Max: 140, DefaultValue: 32, Formatting: "%d", Type:"INT",
+	Labels: []string{"humi", "room humidity"}, UomLabel: "F", MediaType: clients.ContentTypeJSON, FloatEncoding: "eNotation"}
+
+func main() {
+	http.HandleFunc(clients.ApiValueDescriptorRoute, Hello)
+	if e := http.ListenAndServe(":8080", nil); e != nil {
+		log.Fatal(e)
+	}
+}
+
+func Hello(w http.ResponseWriter, req *http.Request) {
+	descriptor1 := testValueDescriptor1
+	descriptor1.Description = testValueDesciptorDescription1
+
+	descriptor2 := testValueDescriptor2
+	descriptor2.Description = testValueDesciptorDescription2
+	descriptors := []models.ValueDescriptor{descriptor1, descriptor2}
+
+	data, err := json.Marshal(descriptors)
+	if err != nil {
+		fmt.Errorf("marshaling error: %s", err.Error())
+	}
+	if _, err := fmt.Fprintf(w, "%s", data); err != nil {
+		log.Fatal(err)
+	}
+	//_, _ = w.Write(data)
+}
+
+

fvt_scripts/setup_env.sh

@@ -23,4 +23,14 @@ else
   done
 fi
 
-fvt_scripts/start_kuiper.sh
+fvt_scripts/start_kuiper.sh
+
+pids=`ps aux|grep "vdmocker" | grep "bin"|awk '{printf $2 " "}'`
+if [ "$pids" = "" ] ; then
+   echo "No value descriptor mockup server was started"
+else
+  for pid in $pids ; do
+    echo "kill value descriptor mockup server " $pid
+    kill -9 $pid
+  done
+fi

fvt_scripts/start_pubmsg.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+go build -o pubmsg fvt_scripts/edgex/pub.go
+export BUILD_ID=dontKillMe
+nohup ./pubmsg > log/pubmsg.out 2>&1 &

fvt_scripts/start_vdmock.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+
+
+echo "starting edgex value descriptor mockup server."
+go build -o vdmocker fvt_scripts/edgex/vd_server.go
+export BUILD_ID=dontKillMe
+nohup ./vdmocker > log/vdmocker.out 2>&1 &

xstream/extensions/edgex_source.go

@@ -1,8 +1,12 @@
 package extensions
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients/coredata"
+	"github.com/edgexfoundry/go-mod-core-contracts/clients/urlclient/local"
 	"github.com/edgexfoundry/go-mod-core-contracts/models"
 	"github.com/edgexfoundry/go-mod-messaging/messaging"
 	"github.com/edgexfoundry/go-mod-messaging/pkg/types"
@@ -13,17 +17,19 @@ import (
 )
 
 type EdgexZMQSource struct {
-	client messaging.MessageClient
+	client     messaging.MessageClient
+	vdc        coredata.ValueDescriptorClient
 	device     string
 	topic      string
 	valueDescs map[string]string
 }
 
 type EdgexConfig struct {
-	Protocol string `json:"protocol"`
-	Server   string `json:"server"`
-	Port     int    `json:"port"`
-	Topic    string `json:"topic"`
+	Protocol      string `json:"protocol"`
+	Server        string `json:"server"`
+	Port          int    `json:"port"`
+	Topic         string `json:"topic"`
+	ServiceServer string `json:"serviceServer"`
 }
 
 func (es *EdgexZMQSource) Configure(device string, props map[string]interface{}) error {
@@ -43,6 +49,13 @@ func (es *EdgexZMQSource) Configure(device string, props map[string]interface{})
 		es.topic = tpc.(string)
 	}
 
+	if serviceServer, ok := props["serviceServer"]; ok {
+		es.vdc = coredata.NewValueDescriptorClient(local.New(serviceServer.(string) + clients.ApiValueDescriptorRoute))
+		es.valueDescs = make(map[string]string)
+	} else {
+		return fmt.Errorf("The service server cannot be empty.")
+	}
+
 	mbconf := types.MessageBusConfig{SubscribeHost: types.HostInfo{Protocol: cfg.Protocol, Host: cfg.Server, Port: cfg.Port}, Type: messaging.ZeroMQ}
 	if client, err := messaging.NewMessageClient(mbconf); err != nil {
 		return err
@@ -50,6 +63,7 @@ func (es *EdgexZMQSource) Configure(device string, props map[string]interface{})
 		es.client = client
 		return nil
 	}
+
 }
 
 func (es *EdgexZMQSource) Open(ctx api.StreamContext, consumer chan<- api.SourceTuple, errCh chan<- error) {
@@ -121,7 +135,7 @@ func (es *EdgexZMQSource) Open(ctx api.StreamContext, consumer chan<- api.Source
 	}
 }
 
-func (es *EdgexZMQSource) getValue(r models.Reading, logger api.Logger) (interface{}, error){
+func (es *EdgexZMQSource) getValue(r models.Reading, logger api.Logger) (interface{}, error) {
 	t, err := es.getType(r.Name, logger)
 	if err != nil {
 		return nil, err
@@ -163,28 +177,30 @@ func (es *EdgexZMQSource) getValue(r models.Reading, logger api.Logger) (interfa
 	}
 }
 
-func (s *EdgexZMQSource) getType(id string, logger api.Logger) (string, error) {
-	return "string", nil
-	//if t, ok := s.valueDescs[id]; ok {
-	//	return t, nil
-	//}
-	//else {
-
-		//url := s.coreDataUrl + "/valuedescriptor/name/" + id
-		//logger.Debugf("get %s", url)
-		//response, err := http.Get(url)
-		//if err != nil {
-		//	return "", fmt.Errorf("fail to get type for value %s: %v", id, err)
-		//}
-		//defer response.Body.Close()
-		//logger.Debugf("response %s", response.Body)
-		//vd := models.ValueDescriptor{}
-		//if err := json.NewDecoder(response.Body).Decode(&vd); err != nil {
-		//	return "", fmt.Errorf("fail to decode value descriptor value %s: %v", id, err)
-		//}
-		//s.valueDescs[id] = vd.Type
-		//return vd.Type, nil
-	//}
+func (es *EdgexZMQSource) fetchAllDataDescriptors() error {
+	if vdArr, err := es.vdc.ValueDescriptors(context.Background()); err != nil {
+		return err
+	} else {
+		for _, vd := range vdArr {
+			es.valueDescs[vd.Id] = vd.Type
+		}
+	}
+	return nil
+}
+
+func (es *EdgexZMQSource) getType(id string, logger api.Logger) (string, error) {
+	if t, ok := es.valueDescs[id]; ok {
+		return t, nil
+	} else {
+		if e := es.fetchAllDataDescriptors(); e != nil {
+			return "", e
+		}
+		if t, ok := es.valueDescs[id]; ok {
+			return t, nil
+		} else {
+			return "", fmt.Errorf("cannot find type info for %s in value descriptor.", id)
+		}
+	}
 }
 
 func (es *EdgexZMQSource) Close(ctx api.StreamContext) error {