use result caching and add xor condition

lib/processor/condition/constructor.go

@@ -57,6 +57,7 @@ type Config struct {
 	Not      NotConfig     `json:"not" yaml:"not"`
 	Or       OrConfig      `json:"or" yaml:"or"`
 	Resource string        `json:"resource" yaml:"resource"`
+	Xor      XorConfig     `json:"xor" yaml:"xor"`
 }
 
 // NewConfig returns a configuration struct fully populated with default values.
@@ -68,6 +69,7 @@ func NewConfig() Config {
 		Not:      NewNotConfig(),
 		Or:       NewOrConfig(),
 		Resource: "",
+		Xor:      NewXorConfig(),
 	}
 }
 

lib/processor/condition/resource.go

@@ -114,8 +114,7 @@ func (c *Resource) Check(msg types.Message) bool {
 		c.log.Errorf("Failed to obtain condition resource '%v': %v", c.name, err)
 		return false
 	}
-	// TODO: Result caching.
-	return cond.Check(msg)
+	return msg.LazyCondition(c.name, cond)
 }
 
 //------------------------------------------------------------------------------

lib/processor/condition/xor.go

@@ -0,0 +1,93 @@
+// Copyright (c) 2018 Ashley Jeffs
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software or associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, or/or sell
+// copies of the Software, or to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright orice or this permission orice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT or LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE or NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package condition
+
+import (
+	"fmt"
+
+	"github.com/Jeffail/benthos/lib/types"
+	"github.com/Jeffail/benthos/lib/util/service/log"
+	"github.com/Jeffail/benthos/lib/util/service/metrics"
+)
+
+//------------------------------------------------------------------------------
+
+func init() {
+	Constructors["xor"] = TypeSpec{
+		constructor: NewXor,
+		description: `
+Xor is a condition that returns the logical XOR of its children conditions,
+meaning it only resolves to true if _exactly_ one of its children conditions
+resolves to true.`,
+	}
+}
+
+//------------------------------------------------------------------------------
+
+// XorConfig is a configuration struct containing fields for the Xor condition.
+type XorConfig []Config
+
+// NewXorConfig returns a XorConfig with default values.
+func NewXorConfig() XorConfig {
+	return XorConfig{}
+}
+
+//------------------------------------------------------------------------------
+
+// Xor is a condition that returns the logical xor of all children.
+type Xor struct {
+	children []Type
+}
+
+// NewXor returns an Xor processor.
+func NewXor(
+	conf Config, mgr types.Manager, log log.Modular, stats metrics.Type,
+) (Type, error) {
+	children := []Type{}
+	for _, childConf := range conf.Xor {
+		child, err := New(childConf, mgr, log, stats)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create child '%v': %v", childConf.Type, err)
+		}
+		children = append(children, child)
+	}
+	return &Xor{
+		children: children,
+	}, nil
+}
+
+//------------------------------------------------------------------------------
+
+// Check attempts to check a message part against a configured condition.
+func (c *Xor) Check(msg types.Message) bool {
+	hadTrue := false
+	for _, child := range c.children {
+		if child.Check(msg) {
+			if hadTrue {
+				return false
+			}
+			hadTrue = true
+		}
+	}
+	return hadTrue
+}
+
+//------------------------------------------------------------------------------

lib/processor/condition/xor_test.go

@@ -0,0 +1,136 @@
+// Copyright (c) 2018 Ashley Jeffs
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package condition
+
+import (
+	"os"
+	"testing"
+
+	"github.com/Jeffail/benthos/lib/types"
+	"github.com/Jeffail/benthos/lib/util/service/log"
+	"github.com/Jeffail/benthos/lib/util/service/metrics"
+)
+
+func TestXorCheck(t *testing.T) {
+	testLog := log.NewLogger(os.Stdout, log.LoggerConfig{LogLevel: "NONE"})
+	testMet := metrics.DudType{}
+
+	testMsg := types.NewMessage([][]byte{
+		[]byte("foo"),
+	})
+
+	passConf := NewConfig()
+	passConf.Content.Operator = "contains"
+	passConf.Content.Arg = "foo"
+
+	failConf := NewConfig()
+	failConf.Content.Operator = "contains"
+	failConf.Content.Arg = "bar"
+
+	tests := []struct {
+		name string
+		arg  []Config
+		want bool
+	}{
+		{
+			name: "one pass",
+			arg: []Config{
+				passConf,
+			},
+			want: true,
+		},
+		{
+			name: "two pass",
+			arg: []Config{
+				passConf,
+				passConf,
+			},
+			want: false,
+		},
+		{
+			name: "one fail",
+			arg: []Config{
+				failConf,
+			},
+			want: false,
+		},
+		{
+			name: "two fail",
+			arg: []Config{
+				failConf,
+				failConf,
+			},
+			want: false,
+		},
+		{
+			name: "first fail",
+			arg: []Config{
+				failConf,
+				passConf,
+			},
+			want: true,
+		},
+		{
+			name: "second fail",
+			arg: []Config{
+				passConf,
+				failConf,
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conf := NewConfig()
+			conf.Type = "xor"
+			conf.Xor = tt.arg
+
+			c, err := New(conf, nil, testLog, testMet)
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			if got := c.Check(testMsg); got != tt.want {
+				t.Errorf("Xor.Check() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestXorBadOperator(t *testing.T) {
+	testLog := log.NewLogger(os.Stdout, log.LoggerConfig{LogLevel: "NONE"})
+	testMet := metrics.DudType{}
+
+	cConf := NewConfig()
+	cConf.Type = "content"
+	cConf.Content.Operator = "NOT_EXIST"
+
+	conf := NewConfig()
+	conf.Type = "xor"
+	conf.Xor = []Config{
+		cConf,
+	}
+
+	_, err := NewXor(conf, nil, testLog, testMet)
+	if err == nil {
+		t.Error("expected error from bad operator")
+	}
+}