Merge pull request #34 from blinklabs-io/feat/filter

feat: basic filter support

Author: agaffney

cmd/snek/main.go

@@ -19,6 +19,7 @@ import (
 	"net/http"
 	"os"
 
+	_ "github.com/blinklabs-io/snek/filter"
 	_ "github.com/blinklabs-io/snek/input"
 	"github.com/blinklabs-io/snek/internal/config"
 	"github.com/blinklabs-io/snek/internal/logging"
@@ -112,6 +113,12 @@ func main() {
 	}
 	pipe.AddInput(input)
 
+	// Configure filters
+	for _, filterEntry := range plugin.GetPlugins(plugin.PluginTypeFilter) {
+		filter := plugin.GetPlugin(plugin.PluginTypeFilter, filterEntry.Name)
+		pipe.AddFilter(filter)
+	}
+
 	// Configure output
 	output := plugin.GetPlugin(plugin.PluginTypeOutput, cfg.Output)
 	if output == nil {

filter/event/event.go

@@ -0,0 +1,85 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package event
+
+import (
+	"github.com/blinklabs-io/snek/event"
+)
+
+type Event struct {
+	errorChan  chan error
+	inputChan  chan event.Event
+	outputChan chan event.Event
+	filterType string
+}
+
+// New returns a new Event object with the specified options applied
+func New(options ...EventOptionFunc) *Event {
+	e := &Event{
+		errorChan:  make(chan error),
+		inputChan:  make(chan event.Event, 10),
+		outputChan: make(chan event.Event, 10),
+	}
+	for _, option := range options {
+		option(e)
+	}
+	return e
+}
+
+// Start the event filter
+func (e *Event) Start() error {
+	go func() {
+		// TODO: pre-process filter params to be more useful for direct comparison
+		for {
+			evt, ok := <-e.inputChan
+			// Channel has been closed, which means we're shutting down
+			if !ok {
+				return
+			}
+			// Drop events if we have a type filter configured and the event doesn't match
+			if e.filterType != "" {
+				if evt.Type != e.filterType {
+					continue
+				}
+			}
+			// Send event along
+			e.outputChan <- evt
+		}
+	}()
+	return nil
+}
+
+// Stop the event filter
+func (e *Event) Stop() error {
+	close(e.inputChan)
+	close(e.outputChan)
+	close(e.errorChan)
+	return nil
+}
+
+// ErrorChan returns the filter error channel
+func (e *Event) ErrorChan() chan error {
+	return e.errorChan
+}
+
+// InputChan returns the input event channel
+func (e *Event) InputChan() chan<- event.Event {
+	return e.inputChan
+}
+
+// OutputChan returns the output event channel
+func (e *Event) OutputChan() <-chan event.Event {
+	return e.outputChan
+}

filter/event/option.go

@@ -0,0 +1,24 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package event
+
+type EventOptionFunc func(*Event)
+
+// WithType specfies the event type to filter on
+func WithType(eventType string) EventOptionFunc {
+	return func(e *Event) {
+		e.filterType = eventType
+	}
+}

filter/event/plugin.go

@@ -0,0 +1,51 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package event
+
+import (
+	"github.com/blinklabs-io/snek/plugin"
+)
+
+var cmdlineOptions struct {
+	eventType string
+}
+
+func init() {
+	plugin.Register(
+		plugin.PluginEntry{
+			Type:               plugin.PluginTypeFilter,
+			Name:               "event",
+			Description:        "filters events based on top-level event attributes",
+			NewFromOptionsFunc: NewFromCmdlineOptions,
+			Options: []plugin.PluginOption{
+				{
+					Name:         "type",
+					Type:         plugin.PluginOptionTypeString,
+					Description:  "specifies event type to filter on",
+					DefaultValue: "",
+					Dest:         &(cmdlineOptions.eventType),
+					CustomFlag:   "type",
+				},
+			},
+		},
+	)
+}
+
+func NewFromCmdlineOptions() plugin.Plugin {
+	p := New(
+		WithType(cmdlineOptions.eventType),
+	)
+	return p
+}

filter/filter.go

@@ -0,0 +1,19 @@
+// Copyright 2023 Blink Labs, LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filter
+
+import (
+	_ "github.com/blinklabs-io/snek/filter/event"
+)

input/chainsync/chainsync.go

@@ -168,15 +168,15 @@ func (c *ChainSync) setupConnection() error {
 }
 
 func (c *ChainSync) handleRollBackward(point ocommon.Point, tip ochainsync.Tip) error {
-	evt := event.New("rollback", time.Now(), NewRollbackEvent(point))
+	evt := event.New("chainsync.rollback", time.Now(), NewRollbackEvent(point))
 	c.eventChan <- evt
 	return nil
 }
 
 func (c *ChainSync) handleRollForward(blockType uint, blockData interface{}, tip ochainsync.Tip) error {
 	switch v := blockData.(type) {
 	case ledger.Block:
-		evt := event.New("block", time.Now(), NewBlockEvent(v, c.includeCbor))
+		evt := event.New("chainsync.block", time.Now(), NewBlockEvent(v, c.includeCbor))
 		c.eventChan <- evt
 	case ledger.BlockHeader:
 		var blockSlot uint64
@@ -202,10 +202,10 @@ func (c *ChainSync) handleRollForward(blockType uint, blockData interface{}, tip
 		if err != nil {
 			return err
 		}
-		blockEvt := event.New("block", time.Now(), NewBlockEvent(block, c.includeCbor))
+		blockEvt := event.New("chainsync.block", time.Now(), NewBlockEvent(block, c.includeCbor))
 		c.eventChan <- blockEvt
 		for _, transaction := range block.Transactions() {
-			txEvt := event.New("transaction", time.Now(), NewTransactionEvent(block, transaction, c.includeCbor))
+			txEvt := event.New("chainsync.transaction", time.Now(), NewTransactionEvent(block, transaction, c.includeCbor))
 			c.eventChan <- txEvt
 		}
 	}

pipeline/pipeline.go

@@ -23,14 +23,17 @@ import (
 
 type Pipeline struct {
 	inputs     []plugin.Plugin
+	filters    []plugin.Plugin
 	outputs    []plugin.Plugin
+	filterChan chan event.Event
 	outputChan chan event.Event
 	errorChan  chan error
 	doneChan   chan bool
 }
 
 func New() *Pipeline {
 	p := &Pipeline{
+		filterChan: make(chan event.Event),
 		outputChan: make(chan event.Event),
 		errorChan:  make(chan error),
 		doneChan:   make(chan bool),
@@ -42,6 +45,10 @@ func (p *Pipeline) AddInput(input plugin.Plugin) {
 	p.inputs = append(p.inputs, input)
 }
 
+func (p *Pipeline) AddFilter(filter plugin.Plugin) {
+	p.filters = append(p.filters, filter)
+}
+
 func (p *Pipeline) AddOutput(output plugin.Plugin) {
 	p.outputs = append(p.outputs, output)
 }
@@ -57,11 +64,35 @@ func (p *Pipeline) Start() error {
 		if err := input.Start(); err != nil {
 			return fmt.Errorf("failed to start input: %s", err)
 		}
-		// Start background process to send input events to combined output channel
-		go p.chanCopyLoop(input.OutputChan(), p.outputChan)
+		// Start background process to send input events to combined filter channel
+		go p.chanCopyLoop(input.OutputChan(), p.filterChan)
 		// Start background error listener
 		go p.errorChanWait(input.ErrorChan())
 	}
+	// Start filters
+	for idx, filter := range p.filters {
+		if err := filter.Start(); err != nil {
+			return fmt.Errorf("failed to start input: %s", err)
+		}
+		if idx == 0 {
+			// Start background process to send events from combined filter channel to first filter plugin
+			go p.chanCopyLoop(p.filterChan, filter.InputChan())
+		} else {
+			// Start background process to send events from previous filter plugin to current filter plugin
+			go p.chanCopyLoop(p.filters[idx-1].OutputChan(), filter.InputChan())
+		}
+		if idx == len(p.filters)-1 {
+			// Start background process to send events from last filter to combined output channel
+			go p.chanCopyLoop(filter.OutputChan(), p.outputChan)
+		}
+		// Start background error listener
+		go p.errorChanWait(filter.ErrorChan())
+	}
+	if len(p.filters) == 0 {
+		// Start background process to send events from combined filter channel to combined output channel if
+		// there are no filter plugins
+		go p.chanCopyLoop(p.filterChan, p.outputChan)
+	}
 	// Start outputs
 	for _, output := range p.outputs {
 		if err := output.Start(); err != nil {
@@ -78,6 +109,7 @@ func (p *Pipeline) Start() error {
 func (p *Pipeline) Stop() error {
 	close(p.doneChan)
 	close(p.errorChan)
+	close(p.filterChan)
 	close(p.outputChan)
 	// Stop inputs
 	for _, input := range p.inputs {

plugin/option.go

@@ -36,13 +36,19 @@ type PluginOption struct {
 	Name         string
 	Type         PluginOptionType
 	CustomEnvVar string
+	CustomFlag   string
 	Description  string
 	DefaultValue interface{}
 	Dest         interface{}
 }
 
-func (p *PluginOption) AddToFlagSet(fs *flag.FlagSet, flagPrefix string) error {
-	flagName := fmt.Sprintf("%s%s", flagPrefix, p.Name)
+func (p *PluginOption) AddToFlagSet(fs *flag.FlagSet, pluginType string, pluginName string) error {
+	var flagName string
+	if p.CustomFlag != "" {
+		flagName = fmt.Sprintf("%s-%s", pluginType, p.CustomFlag)
+	} else {
+		flagName = fmt.Sprintf("%s-%s-%s", pluginType, pluginName, p.Name)
+	}
 	switch p.Type {
 	case PluginOptionTypeString:
 		fs.StringVar(p.Dest.(*string), flagName, p.DefaultValue.(string), p.Description)

plugin/register.go

@@ -24,6 +24,7 @@ type PluginType int
 const (
 	PluginTypeInput  PluginType = 1
 	PluginTypeOutput PluginType = 2
+	PluginTypeFilter PluginType = 3
 )
 
 func PluginTypeName(pluginType PluginType) string {
@@ -32,6 +33,8 @@ func PluginTypeName(pluginType PluginType) string {
 		return "input"
 	case PluginTypeOutput:
 		return "output"
+	case PluginTypeFilter:
+		return "filter"
 	default:
 		return ""
 	}
@@ -53,9 +56,8 @@ func Register(pluginEntry PluginEntry) {
 
 func PopulateCmdlineOptions(fs *flag.FlagSet) error {
 	for _, plugin := range pluginEntries {
-		flagPrefix := fmt.Sprintf("%s-%s-", PluginTypeName(plugin.Type), plugin.Name)
 		for _, option := range plugin.Options {
-			if err := option.AddToFlagSet(fs, flagPrefix); err != nil {
+			if err := option.AddToFlagSet(fs, PluginTypeName(plugin.Type), plugin.Name); err != nil {
 				return err
 			}
 		}