update triggers to invoke latest RunAction command (#152)

* update triggers to invoke latest RunAction command

* activity-return,reply,mapper: add support for running in tester service

Author: Frank Martinez

.gitignore

@@ -28,6 +28,10 @@ action_metadata.go
 activity_metadata.go
 trigger_metadata.go
 
+action_metadata.go
+activity_metadata.go
+trigger_metadata.go
+
 # Compiled Object files, Static and Dynamic libs (Shared Objects)
 *.o
 *.a

trigger/cli/trigger.go

@@ -1,7 +1,6 @@
 package cli
 
 import (
-	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -143,14 +142,24 @@ func (t *CliTrigger) Invoke(actionId string, handlerCfg *trigger.HandlerConfig,
 	startAttrs, _ := t.metadata.OutputsToAttrs(data, false)
 
 	act := action.Get(actionId)
-	ctx := trigger.NewContextWithData(context.Background(), &trigger.ContextData{Attrs: startAttrs, HandlerCfg: handlerCfg})
-	_, replyData, err := t.runner.Run(ctx, act, actionId, nil)
+
+	ctx := trigger.NewInitialContext(startAttrs, handlerCfg)
+	results, err := t.runner.RunAction(ctx, act, nil)
 
 	if err != nil {
 		log.Debugf("CLI Trigger Error: %s", err.Error())
 		return "", err
 	}
 
+	var replyData interface{}
+
+	if len(results) != 0 {
+		dataAttr, ok := results["data"]
+		if ok {
+			replyData = dataAttr.Value
+		}
+	}
+
 	if replyData != nil {
 		data, err := json.Marshal(replyData)
 		if err != nil {

trigger/cli/trigger.json

@@ -7,12 +7,18 @@
   "title": "CLI Trigger",
   "description": "Simple CLI Trigger",
   "homepage": "https://github.com/TIBCOSoftware/flogo-contrib/tree/master/trigger/cli",
-  "outputs": [
+  "output": [
     {
       "name": "args",
       "type": "array"
     }
   ],
+  "reply": [
+    {
+      "name": "data",
+      "type": "object"
+    }
+  ],
   "handler": {
     "settings": [
       {

trigger/coap/trigger.go

@@ -2,7 +2,6 @@ package coap
 
 import (
 	"bytes"
-	"context"
 	"fmt"
 	"net"
 	"net/url"
@@ -228,10 +227,9 @@ func newActionHandler(rt *CoapTrigger, resource *CoapResource) coap.Handler {
 		//rh.addr2 = addr
 		//rh.conn = conn
 
-		action := action.Get(actionId)
-
-		context := trigger.NewContext(context.Background(), startAttrs)
-		_, _, err := rt.runner.Run(context, action, actionId, nil)
+		act := action.Get(actionId)
+		ctx := trigger.NewInitialContext(startAttrs, nil)
+		_, err := rt.runner.RunAction(ctx, act, nil)
 
 		if err != nil {
 			//todo determining if 404 or 500

trigger/lambda/trigger.go

@@ -1,7 +1,6 @@
 package lambda
 
 import (
-	"context"
 	"encoding/json"
 	"flag"
 
@@ -99,8 +98,18 @@ func Invoke() (interface{}, error) {
 	}
 
 	act := action.Get(actionId)
-	ctx := trigger.NewContextWithData(context.Background(), &trigger.ContextData{Attrs: startAttrs, HandlerCfg: singleton.config.Handlers[0]})
-	_, replyData, err := singleton.runner.Run(ctx, act, actionId, nil)
+
+	ctx := trigger.NewInitialContext(startAttrs, singleton.config.Handlers[0])
+	results, err := singleton.runner.RunAction(ctx, act, nil)
+
+	var replyData interface{}
+
+	if len(results) != 0 {
+		dataAttr, ok := results["data"]
+		if ok {
+			replyData = dataAttr.Value
+		}
+	}
 
 	if err != nil {
 		log.Debugf("Lambda Trigger Error: %s", err.Error())

trigger/lambda/trigger.json

@@ -9,7 +9,7 @@
   "homepage": "https://github.com/TIBCOSoftware/flogo-contrib/tree/master/trigger/lambda",
   "settings": [
   ],
-  "outputs": [
+  "output": [
     {
       "name": "logStreamName",
       "type": "string"
@@ -30,5 +30,11 @@
       "name": "evt",
       "type": "object"
     }
+  ],
+  "reply": [
+    {
+      "name": "data",
+      "type": "object"
+    }
   ]
 }

trigger/mqtt/trigger.go

@@ -1,7 +1,6 @@
 package mqtt
 
 import (
-	"context"
 	"encoding/json"
 	"strconv"
 	"time"
@@ -141,14 +140,26 @@ func (t *MqttTrigger) RunAction(actionURI string, payload string) {
 	//todo handle error
 	startAttrs, _ := t.metadata.OutputsToAttrs(req.Data, false)
 
-	action := action.Get(actionURI)
-	context := trigger.NewContext(context.Background(), startAttrs)
-	_, replyData, err := t.runner.Run(context, action, actionURI, nil)
+	act := action.Get(actionURI)
+
+	//add handlerCfg to handle return mappings
+	ctx := trigger.NewInitialContext(startAttrs, nil)
+	results, err := t.runner.RunAction(ctx, act, nil)
+
 	if err != nil {
 		log.Error("Error starting action: ", err.Error())
 	}
 	log.Debugf("Ran action: [%s]", actionURI)
 
+	var replyData interface{}
+
+	if len(results) != 0 {
+		dataAttr, ok := results["data"]
+		if ok {
+			replyData = dataAttr.Value
+		}
+	}
+
 	if replyData != nil {
 		data, err := json.Marshal(replyData)
 		if err != nil {

trigger/mqtt/trigger.json

@@ -41,12 +41,18 @@
       "type": "boolean"
     }
   ],
-  "outputs": [
+  "output": [
     {
       "name": "message",
       "type": "string"
     }
   ],
+  "reply": [
+    {
+      "name": "data",
+      "type": "object"
+    }
+  ],
   "handler": {
     "settings": [
       {

trigger/rest/trigger.go

@@ -1,7 +1,6 @@
 package rest
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -156,8 +155,22 @@ func newActionHandler(rt *RestTrigger, actionId string, handlerCfg *trigger.Hand
 		startAttrs, _ := rt.metadata.OutputsToAttrs(data, false)
 
 		act := action.Get(actionId)
-		ctx := trigger.NewContextWithData(context.Background(), &trigger.ContextData{Attrs: startAttrs, HandlerCfg: handlerCfg})
-		replyCode, replyData, err := rt.runner.Run(ctx, act, actionId, nil)
+		ctx := trigger.NewInitialContext(startAttrs, handlerCfg)
+		results, err := rt.runner.RunAction(ctx, act, nil)
+
+		var replyData interface{}
+		var replyCode int
+
+		if len(results) != 0 {
+			dataAttr, ok := results["data"]
+			if ok {
+				replyData = dataAttr.Value
+			}
+			codeAttr, ok := results["code"]
+			if ok {
+				replyCode = codeAttr.Value.(int)
+			}
+		}
 
 		if err != nil {
 			log.Debugf("REST Trigger Error: %s", err.Error())

trigger/rest/trigger.json

@@ -13,7 +13,7 @@
       "required": true
     }
   ],
-  "outputs": [
+  "output": [
     {
       "name": "params",
       "type": "params"

trigger/timer/timer.go

@@ -111,7 +111,8 @@ func (t *TimerTrigger) scheduleOnce(endpoint *trigger.HandlerConfig) {
 
 		if act != nil {
 			log.Debugf("Running action: %s", endpoint.ActionId)
-			_, _, err := t.runner.Run(context.Background(), act, endpoint.ActionId, nil)
+
+			_, err := t.runner.RunAction(context.Background(), act, nil )
 
 			if err != nil {
 				log.Error("Error running action: ", err.Error())