adjust the structure of Input and Output

Signed-off-by: laminar <fangtian@kubesphere.io>

Author: tpiperatgod

context/context.go

@@ -256,21 +256,33 @@ type SyncRequest struct {
 
 type Input struct {
 	Uri           string            `json:"uri,omitempty"`
-	Component     string            `json:"component"`
+	ComponentName string            `json:"componentName"`
 	ComponentType string            `json:"componentType"`
-	Type          ResourceType      `json:"type"`
 	Metadata      map[string]string `json:"metadata,omitempty"`
 }
 
+// GetType will be called after the context has been parsed correctly,
+// therefore we do not have to handle the error return of getBuildingBlockType()
+func (i *Input) GetType() ResourceType {
+	bbt, _ := getBuildingBlockType(i.ComponentType)
+	return bbt
+}
+
 type Output struct {
 	Uri           string            `json:"uri,omitempty"`
-	Component     string            `json:"component"`
+	ComponentName string            `json:"componentName"`
 	ComponentType string            `json:"componentType"`
-	Type          ResourceType      `json:"type"`
 	Metadata      map[string]string `json:"metadata,omitempty"`
 	Operation     string            `json:"operation,omitempty"`
 }
 
+// GetType will be called after the context has been parsed correctly,
+// therefore we do not have to handle the error return of getBuildingBlockType()
+func (o *Output) GetType() ResourceType {
+	bbt, _ := getBuildingBlockType(o.ComponentType)
+	return bbt
+}
+
 type FunctionOut struct {
 	mu       sync.Mutex
 	Code     int               `json:"code"`
@@ -348,12 +360,12 @@ func (ctx *FunctionContext) Send(outputName string, data []byte) ([]byte, error)
 		payloadBytes = ie.GetCloudEventJSON()
 	}
 
-	switch output.Type {
+	switch output.GetType() {
 	case OpenFuncTopic:
-		err = ctx.daprClient.PublishEvent(context.Background(), output.Component, output.Uri, payload)
+		err = ctx.daprClient.PublishEvent(context.Background(), output.ComponentName, output.Uri, payload)
 	case OpenFuncBinding:
 		in := &dapr.InvokeBindingRequest{
-			Name:      output.Component,
+			Name:      output.ComponentName,
 			Operation: output.Operation,
 			Data:      payloadBytes,
 			Metadata:  output.Metadata,
@@ -684,22 +696,18 @@ func parseContext() (*FunctionContext, error) {
 
 	if !ctx.HasInputs() {
 		for name, in := range ctx.Inputs {
-			switch in.Type {
-			case OpenFuncBinding, OpenFuncTopic:
-				break
-			default:
-				return nil, fmt.Errorf("invalid input type %s: %s", name, in.Type)
+			if _, err := getBuildingBlockType(in.ComponentType); err != nil {
+				klog.Errorf("failed to get building block type for input %s: %v", name, err)
+				return nil, err
 			}
 		}
 	}
 
 	if !ctx.HasOutputs() {
 		for name, out := range ctx.Outputs {
-			switch out.Type {
-			case OpenFuncBinding, OpenFuncTopic:
-				break
-			default:
-				return nil, fmt.Errorf("invalid output type %s: %s", name, out.Type)
+			if _, err := getBuildingBlockType(out.ComponentType); err != nil {
+				klog.Errorf("failed to get building block type for output %s: %v", name, err)
+				return nil, err
 			}
 		}
 	}
@@ -784,3 +792,17 @@ func traceable(t string) bool {
 	// determine if the tracing metadata can be added.
 	return bindingQueueComponents[t]
 }
+
+func getBuildingBlockType(componentType string) (ResourceType, error) {
+	typeSplit := strings.Split(componentType, ".")
+	if len(typeSplit) > 1 {
+		t := typeSplit[0]
+		switch ResourceType(t) {
+		case OpenFuncBinding, OpenFuncTopic:
+			return ResourceType(t), nil
+		default:
+			return "", fmt.Errorf("unknown component type: %s", t)
+		}
+	}
+	return "", errors.New("invalid component type")
+}

context/innerevent_test.go

@@ -17,39 +17,36 @@ var funcCtx = `{
   "inputs": {
     "cron": {
       "uri": "cron_input",
-      "type": "bindings",
-      "component": "cron_input"
+      "componentName": "cron_input",
+      "componentType": "bindings.cron"
     },
     "eventbus": {
       "uri": "default",
-      "type": "pubsub",
-      "component": "nats_eventbus"
+      "componentName": "nats_eventbus",
+      "componentType": "pubsub.natsstreaming"
     }
   },
   "outputs": {
     "echo": {
       "uri": "echo",
       "operation": "create",
-      "component": "echo",
+      "componentName": "echo",
       "componentType": "bindings.kafka",
       "metadata": {
         "path": "echo",
         "Content-Type": "application/json; charset=utf-8"
-      },
-      "type": "bindings"
+      }
     },
     "target": {
       "uri": "sample",
       "operation": "create",
-      "component": "kafka-server",
-      "componentType": "pubsub.kafka",
-      "type": "pubsub"
+      "componentName": "kafka-server",
+      "componentType": "pubsub.kafka"
     },
     "target2": {
       "uri": "cron_output",
-      "component": "cron_output",
-      "componentType": "bindings.cron",
-      "type": "bindings"
+      "componentName": "cron_output",
+      "componentType": "bindings.cron"
     }
   }
 }`

framework/framework_test.go

@@ -176,8 +176,8 @@ func TestAsyncBindingsFunction(t *testing.T) {
   "inputs": {
     "cron": {
       "uri": "test",
-      "type": "bindings",
-      "component": "test"
+      "componentName": "test",
+      "componentType": "bindings.Kafka"
     }
   }
 }`
@@ -259,8 +259,8 @@ func TestAsyncPubsubTopic(t *testing.T) {
   "inputs": {
     "sub": {
       "uri": "my_topic",
-      "type": "pubsub",
-      "component": "msg"
+      "componentName": "msg",
+      "componentType": "pubsub.kafka"
     }
   }
 }`

runtime/async/async.go

@@ -90,9 +90,9 @@ func (r *Runtime) RegisterOpenFunction(
 		// Serving function with inputs
 		if ctx.HasInputs() {
 			for name, input := range ctx.GetInputs() {
-				switch input.Type {
+				switch input.GetType() {
 				case ofctx.OpenFuncBinding:
-					input.Uri = input.Component
+					input.Uri = input.ComponentName
 					funcErr = r.handler.AddBindingInvocationHandler(input.Uri, func(c context.Context, in *dapr.BindingEvent) (out []byte, err error) {
 						rm := runtime.NewRuntimeManager(ctx, prePlugins, postPlugins)
 						rm.FuncContext.SetEvent(name, in)
@@ -109,7 +109,7 @@ func (r *Runtime) RegisterOpenFunction(
 					})
 				case ofctx.OpenFuncTopic:
 					sub := &dapr.Subscription{
-						PubsubName: input.Component,
+						PubsubName: input.ComponentName,
 						Topic:      input.Uri,
 					}
 					funcErr = r.handler.AddTopicEventHandler(sub, func(c context.Context, e *dapr.TopicEvent) (retry bool, err error) {
@@ -137,7 +137,7 @@ func (r *Runtime) RegisterOpenFunction(
 						}
 					})
 				default:
-					return fmt.Errorf("invalid input type: %s", input.Type)
+					return fmt.Errorf("invalid input type: %s", input.GetType())
 				}
 				if funcErr != nil {
 					// When the function throws an exception,