Conditional filter

activity/filter/README.md

@@ -21,7 +21,7 @@ flogo install github.com/project-flogo/stream/activity/filter
       "name": "type",
       "type": "string",
       "required": true,
-      "allowed" : ["non-zero"]
+      "allowed" : ["non-zero","conditional"]
     },
     {
       "name": "proceedOnlyOnEmit",
@@ -32,6 +32,10 @@ flogo install github.com/project-flogo/stream/activity/filter
     {
       "name": "value",
       "type": "any"
+    },
+    {
+      "name": "condition",
+      "type": "boolean"
     }
   ],
   "output": [
@@ -51,14 +55,15 @@ flogo install github.com/project-flogo/stream/activity/filter
 #### Settings:
 | Setting     | Required | Description |
 |:------------|:---------|:------------|
-| type              | true   | The type of filter to apply (ex. non-zero)
+| type              | true   | The type of filter to apply (ex. non-zero, conditional)
 | proceedOnlyOnEmit | false  | Indicates that the next activity should proceed, true by default
 _note_ : if using this activity in a flow, proceedOnlyOnEmit should be set to false
 
 #### Input:
 | Name     | Description |
 |:------------|:---------|
 | value    | The input value
+| condition| The filter condition when filter type is 'conditional'
 
 #### Output:
 | Name     | Description |

activity/filter/activity.go

@@ -8,19 +8,21 @@ import (
 )
 
 const (
-	ivValue    = "value"
-	ovFiltered = "filtered"
-	ovValue    = "value"
+	ivValue     = "value"
+	ivCondition = "condition"
+	ovFiltered  = "filtered"
+	ovValue     = "value"
 )
 
 //we can generate json from this! - we could also create a "validate-able" object from this
 type Settings struct {
-	Type              string `md:"type,allowed(non-zero)"`
+	Type              string `md:"type,allowed(non-zero,conditional)"`
 	ProceedOnlyOnEmit bool
 }
 
 type Input struct {
-	Value interface{} `md:"value"`
+	Value     interface{} `md:"value"`
+	Condition bool        `md:"condition"`
 }
 
 type Output struct {
@@ -45,6 +47,8 @@ func New(ctx activity.InitContext) (activity.Activity, error) {
 
 	if s.Type == "non-zero" {
 		act.filter = &NonZeroFilter{}
+	} else if s.Type == "conditional" {
+		act.filter = &ConditionalFilter{}
 	} else {
 		return nil, fmt.Errorf("unsupported filter: '%s'", s.Type)
 	}
@@ -72,8 +76,9 @@ func (a *Activity) Eval(ctx activity.Context) (done bool, err error) {
 	proceedOnlyOnEmit := a.proceedOnlyOnEmit
 
 	in := ctx.GetInput(ivValue)
+	condition := ctx.GetInput(ivCondition).(bool)
 
-	filteredOut := filter.FilterOut(in)
+	filteredOut := filter.FilterOut(in, condition)
 
 	done = !(proceedOnlyOnEmit && filteredOut)
 
@@ -90,5 +95,5 @@ func (a *Activity) Eval(ctx activity.Context) (done bool, err error) {
 }
 
 type Filter interface {
-	FilterOut(val interface{}) bool
+	FilterOut(val interface{}, condition bool) bool
 }

activity/filter/activity_test.go

@@ -49,6 +49,42 @@ func TestEval(t *testing.T) {
 	assert.Equal(t, 0, value)
 }
 
+func TestEvalWithCondition(t *testing.T) {
+
+	settings := &Settings{Type: "conditional", ProceedOnlyOnEmit: true}
+	//mf := mapper.NewFactory(resolve.GetBasicResolver())
+	iCtx := test.NewActivityInitContext(settings, nil)
+
+	act, err := New(iCtx)
+	assert.Nil(t, err)
+
+	tc := test.NewActivityContext(act.Metadata())
+
+	tc.SetInput(ivValue, 1)
+	tc.SetInput(ivCondition, true)
+	done, err := act.Eval(tc)
+	assert.Nil(t, err)
+	assert.False(t, done)
+
+	filtered := tc.GetOutput(ovFiltered).(bool)
+	value := tc.GetOutput(ovValue)
+
+	assert.True(t, filtered)
+	assert.Equal(t, 1, value)
+
+	tc.SetInput(ivValue, 1)
+	tc.SetInput(ivCondition, false)
+	done, err = act.Eval(tc)
+	assert.Nil(t, err)
+	assert.True(t, done)
+
+	filtered = tc.GetOutput(ovFiltered).(bool)
+	value = tc.GetOutput(ovValue)
+
+	assert.False(t, filtered)
+	assert.Equal(t, 1, value)
+}
+
 func TestPOOEFalse(t *testing.T) {
 
 	settings := &Settings{Type: "non-zero", ProceedOnlyOnEmit: false}

activity/filter/descriptor.json

@@ -10,7 +10,7 @@
       "name": "type",
       "type": "string",
       "required": true,
-      "allowed" : ["non-zero"]
+      "allowed" : ["non-zero", "conditional"]
     },
     {
       "name": "proceedOnlyOnEmit",
@@ -21,6 +21,10 @@
     {
       "name": "value",
       "type": "any"
+    },
+    {
+      "name": "condition",
+      "type": "boolean"
     }
   ],
   "output": [

activity/filter/filters.go

@@ -3,7 +3,10 @@ package filter
 type NonZeroFilter struct {
 }
 
-func (*NonZeroFilter) FilterOut(val interface{}) bool {
+type ConditionalFilter struct {
+}
+
+func (*NonZeroFilter) FilterOut(val interface{}, condition bool) bool {
 	return !IsNonZero(val)
 }
 
@@ -33,3 +36,7 @@ func IsNonZero(val interface{}) bool {
 	//todo handle unsupported type
 	return true
 }
+
+func (*ConditionalFilter) FilterOut(val interface{}, condition bool) bool {
+	return condition
+}