Skip OPTIONS request for input binding if direction is specified (dap…

…r#6608)

* skip OPTIONS request for input binding if direction is specified

Signed-off-by: yaron2 <schneider.yaron@live.com>

* linter

Signed-off-by: yaron2 <schneider.yaron@live.com>

---------

Signed-off-by: yaron2 <schneider.yaron@live.com>

pkg/runtime/runtime.go

@@ -1721,6 +1721,21 @@ func isBindingOfDirection(direction string, metadata []commonapi.NameValuePair)
 	return !directionFound
 }
 
+func isBindingOfExplicitDirection(direction string, metadata map[string]string) bool {
+	for k, v := range metadata {
+		if strings.EqualFold(k, bindingDirection) {
+			directions := strings.Split(v, ",")
+			for _, d := range directions {
+				if strings.TrimSpace(strings.ToLower(d)) == direction {
+					return true
+				}
+			}
+		}
+	}
+
+	return false
+}
+
 func (a *DaprRuntime) initInputBinding(c componentsV1alpha1.Component) error {
 	if !isBindingOfDirection(inputBinding, c.Spec.Metadata) {
 		return nil
@@ -3476,10 +3491,18 @@ func (a *DaprRuntime) startReadingFromBindings() (err error) {
 	a.inputBindingsCtx, a.inputBindingsCancel = context.WithCancel(a.ctx)
 
 	for name, binding := range a.compStore.ListInputBindings() {
-		isSubscribed, err := a.isAppSubscribedToBinding(name)
-		if err != nil {
-			return err
+		var isSubscribed bool
+		m := binding.GetComponentMetadata()
+
+		if isBindingOfExplicitDirection(inputBinding, m) {
+			isSubscribed = true
+		} else {
+			isSubscribed, err = a.isAppSubscribedToBinding(name)
+			if err != nil {
+				return err
+			}
 		}
+
 		if !isSubscribed {
 			log.Infof("app has not subscribed to binding %s.", name)
 			continue

pkg/runtime/runtime_test.go

@@ -4479,7 +4479,7 @@ func (b *mockBinding) Close() error {
 }
 
 func (b *mockBinding) GetComponentMetadata() map[string]string {
-	return map[string]string{}
+	return b.metadata
 }
 
 func TestInvokeOutputBindings(t *testing.T) {
@@ -6341,3 +6341,108 @@ func TestIsBindingOfDirection(t *testing.T) {
 		assert.False(t, f)
 	})
 }
+
+func TestIsBindingOfExplicitDirection(t *testing.T) {
+	t.Run("no direction in metadata input binding", func(t *testing.T) {
+		m := map[string]string{}
+		r := isBindingOfExplicitDirection("input", m)
+
+		assert.False(t, r)
+	})
+
+	t.Run("no direction in metadata output binding", func(t *testing.T) {
+		m := map[string]string{}
+		r := isBindingOfExplicitDirection("input", m)
+
+		assert.False(t, r)
+	})
+
+	t.Run("direction is input binding", func(t *testing.T) {
+		m := map[string]string{
+			"direction": "input",
+		}
+		r := isBindingOfExplicitDirection("input", m)
+
+		assert.True(t, r)
+	})
+
+	t.Run("direction is output binding", func(t *testing.T) {
+		m := map[string]string{
+			"direction": "output",
+		}
+		r := isBindingOfExplicitDirection("output", m)
+
+		assert.True(t, r)
+	})
+
+	t.Run("direction is not output binding", func(t *testing.T) {
+		m := map[string]string{
+			"direction": "input",
+		}
+		r := isBindingOfExplicitDirection("output", m)
+
+		assert.False(t, r)
+	})
+
+	t.Run("direction is not input binding", func(t *testing.T) {
+		m := map[string]string{
+			"direction": "output",
+		}
+		r := isBindingOfExplicitDirection("input", m)
+
+		assert.False(t, r)
+	})
+
+	t.Run("direction is both input and output binding", func(t *testing.T) {
+		m := map[string]string{
+			"direction": "output, input",
+		}
+
+		r := isBindingOfExplicitDirection("input", m)
+		assert.True(t, r)
+
+		r2 := isBindingOfExplicitDirection("output", m)
+
+		assert.True(t, r2)
+	})
+}
+
+func TestStartReadingFromBindings(t *testing.T) {
+	t.Run("OPTIONS request when direction is not specified", func(t *testing.T) {
+		rt := NewTestDaprRuntime(modes.KubernetesMode)
+		mockAppChannel := new(channelt.MockAppChannel)
+
+		mockAppChannel.On("InvokeMethod", mock.Anything, mock.Anything).Return(invokev1.NewInvokeMethodResponse(200, "OK", nil), nil)
+		rt.appChannel = mockAppChannel
+		defer stopRuntime(t, rt)
+
+		m := &mockBinding{}
+
+		rt.compStore.AddInputBinding("test", m)
+		err := rt.startReadingFromBindings()
+
+		assert.NoError(t, err)
+		assert.Len(t, mockAppChannel.Calls, 1)
+	})
+
+	t.Run("No OPTIONS request when direction is specified", func(t *testing.T) {
+		rt := NewTestDaprRuntime(modes.KubernetesMode)
+		mockAppChannel := new(channelt.MockAppChannel)
+
+		mockAppChannel.On("InvokeMethod", mock.Anything, mock.Anything).Return(invokev1.NewInvokeMethodResponse(200, "OK", nil), nil)
+		rt.appChannel = mockAppChannel
+		defer stopRuntime(t, rt)
+
+		m := &mockBinding{
+			metadata: map[string]string{
+				"direction": "input",
+			},
+		}
+
+		rt.compStore.AddInputBinding("test", m)
+		err := rt.startReadingFromBindings()
+
+		assert.NoError(t, err)
+		assert.Len(t, mockAppChannel.Calls, 0)
+	})
+}