fix: plugins own their error channel

Signed-off-by: Chris Gianelloni <wolf31o2@blinklabs.io>

Author: wolf31o2

filter/chainsync/chainsync.go

@@ -158,30 +158,36 @@ func TestChainSync_Stop(t *testing.T) {
 	if err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}
-	// Check if channels are closed
-	select {
-	case <-c.inputChan:
-	default:
-		t.Fatalf("expected inputChan to be closed")
+	// Check if channels are nil after stop
+	if c.inputChan != nil {
+		t.Fatalf("expected inputChan to be nil after stop")
 	}
-	select {
-	case <-c.outputChan:
-	default:
-		t.Fatalf("expected outputChan to be closed")
+	if c.outputChan != nil {
+		t.Fatalf("expected outputChan to be nil after stop")
 	}
 }
 
 func TestChainSync_InputChan(t *testing.T) {
 	c := New()
+	err := c.Start()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	defer c.Stop()
 	if c.InputChan() == nil {
-		t.Fatalf("expected non-nil inputChan")
+		t.Fatalf("expected non-nil inputChan after Start()")
 	}
 }
 
 func TestChainSync_OutputChan(t *testing.T) {
 	c := New()
+	err := c.Start()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	defer c.Stop()
 	if c.OutputChan() == nil {
-		t.Fatalf("expected non-nil outputChan")
+		t.Fatalf("expected non-nil outputChan after Start()")
 	}
 }
 

filter/chainsync/chainsync_test.go

@@ -50,11 +50,24 @@ func TestPluginChannels(t *testing.T) {
 	// Create a new plugin instance
 	p := NewFromCmdlineOptions()
 
+	// Start the plugin (channels are created in Start())
+	err := p.Start()
+	assert.NoError(t, err, "Plugin should start without errors")
+	defer p.Stop()
+
 	// Verify that the input channel is not nil
-	assert.NotNil(t, p.InputChan(), "Input channel should not be nil")
+	assert.NotNil(
+		t,
+		p.InputChan(),
+		"Input channel should not be nil after Start()",
+	)
 
 	// Verify that the output channel is not nil
-	assert.NotNil(t, p.OutputChan(), "Output channel should not be nil")
+	assert.NotNil(
+		t,
+		p.OutputChan(),
+		"Output channel should not be nil after Start()",
+	)
 }
 
 func TestPluginEventProcessing(t *testing.T) {

filter/chainsync/plugin_test.go

@@ -16,25 +16,25 @@ package event
 
 import (
 	"slices"
+	"sync"
 
 	"github.com/blinklabs-io/adder/event"
 	"github.com/blinklabs-io/adder/plugin"
 )
 
 type Event struct {
-	errorChan   chan<- error
+	errorChan   chan error
 	inputChan   chan event.Event
 	outputChan  chan event.Event
+	doneChan    chan struct{}
+	stopOnce    sync.Once
 	logger      plugin.Logger
 	filterTypes []string
 }
 
 // New returns a new Event object with the specified options applied
 func New(options ...EventOptionFunc) *Event {
-	e := &Event{
-		inputChan:  make(chan event.Event, 10),
-		outputChan: make(chan event.Event, 10),
-	}
+	e := &Event{}
 	for _, option := range options {
 		option(e)
 	}
@@ -43,37 +43,58 @@ func New(options ...EventOptionFunc) *Event {
 
 // Start the event filter
 func (e *Event) Start() error {
+	e.errorChan = make(chan error)
+	e.inputChan = make(chan event.Event, 10)
+	e.outputChan = make(chan event.Event, 10)
+	e.doneChan = make(chan struct{})
+	e.stopOnce = sync.Once{}
 	go func() {
 		for {
-			evt, ok := <-e.inputChan
-			// Channel has been closed, which means we're shutting down
-			if !ok {
+			select {
+			case <-e.doneChan:
 				return
-			}
-			// Drop events if we have a type filter configured and the event doesn't match
-			if len(e.filterTypes) > 0 {
-				matched := slices.Contains(e.filterTypes, evt.Type)
-				if !matched {
-					continue
+			case 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 len(e.filterTypes) > 0 {
+					matched := slices.Contains(e.filterTypes, evt.Type)
+					if !matched {
+						continue
+					}
+				}
+				// Send event along
+				e.outputChan <- evt
 			}
-			// Send event along
-			e.outputChan <- evt
 		}
 	}()
 	return nil
 }
 
 // Stop the event filter
 func (e *Event) Stop() error {
-	close(e.inputChan)
-	close(e.outputChan)
+	e.stopOnce.Do(func() {
+		if e.doneChan != nil {
+			close(e.doneChan)
+		}
+		if e.inputChan != nil {
+			close(e.inputChan)
+		}
+		if e.outputChan != nil {
+			close(e.outputChan)
+		}
+		if e.errorChan != nil {
+			close(e.errorChan)
+		}
+	})
 	return nil
 }
 
-// SetErrorChan sets the error channel
-func (e *Event) SetErrorChan(ch chan<- error) {
-	e.errorChan = ch
+// ErrorChan returns the filter error channel
+func (e *Event) ErrorChan() chan error {
+	return e.errorChan
 }
 
 // InputChan returns the input event channel

filter/event/event.go

@@ -91,7 +91,7 @@ type ChainSync struct {
 	kupoClient         *kugo.Client
 	oConn              *ouroboros.Connection
 	eventChan          chan event.Event
-	errorChan          chan<- error
+	errorChan          chan error
 	status             *ChainSyncStatus
 	dialFamily         string
 	kupoUrl            string
@@ -128,7 +128,6 @@ type StatusUpdateFunc func(ChainSyncStatus)
 // New returns a new ChainSync object with the specified options applied
 func New(options ...ChainSyncOptionFunc) *ChainSync {
 	c := &ChainSync{
-		eventChan:       make(chan event.Event, 10),
 		intersectPoints: []ocommon.Point{},
 		status:          &ChainSyncStatus{},
 	}
@@ -142,6 +141,8 @@ func New(options ...ChainSyncOptionFunc) *ChainSync {
 
 // Start the chain sync input
 func (c *ChainSync) Start() error {
+	c.eventChan = make(chan event.Event, 10)
+	c.errorChan = make(chan error)
 	if err := c.setupConnection(); err != nil {
 		return err
 	}
@@ -166,14 +167,25 @@ func (c *ChainSync) Start() error {
 
 // Stop the chain sync input
 func (c *ChainSync) Stop() error {
-	err := c.oConn.Close()
-	close(c.eventChan)
+	var err error
+	if c.oConn != nil {
+		err = c.oConn.Close()
+		c.oConn = nil
+	}
+	if c.eventChan != nil {
+		close(c.eventChan)
+		c.eventChan = nil
+	}
+	if c.errorChan != nil {
+		close(c.errorChan)
+		c.errorChan = nil
+	}
 	return err
 }
 
-// SetErrorChan sets the error channel
-func (c *ChainSync) SetErrorChan(ch chan<- error) {
-	c.errorChan = ch
+// ErrorChan returns the input error channel
+func (c *ChainSync) ErrorChan() chan error {
+	return c.errorChan
 }
 
 // InputChan always returns nil

input/chainsync/chainsync.go

@@ -23,16 +23,14 @@ import (
 type CallbackFunc func(event.Event) error
 
 type EmbeddedOutput struct {
-	errorChan    chan<- error
+	errorChan    chan error
 	eventChan    chan event.Event
 	callbackFunc CallbackFunc
 	outputChan   chan event.Event
 }
 
 func New(options ...EmbeddedOptionFunc) *EmbeddedOutput {
-	e := &EmbeddedOutput{
-		eventChan: make(chan event.Event, 10),
-	}
+	e := &EmbeddedOutput{}
 	for _, option := range options {
 		option(e)
 	}
@@ -41,6 +39,8 @@ func New(options ...EmbeddedOptionFunc) *EmbeddedOutput {
 
 // Start the embedded output
 func (e *EmbeddedOutput) Start() error {
+	e.eventChan = make(chan event.Event, 10)
+	e.errorChan = make(chan error)
 	go func() {
 		for {
 			evt, ok := <-e.eventChan
@@ -66,16 +66,24 @@ func (e *EmbeddedOutput) Start() error {
 
 // Stop the embedded output
 func (e *EmbeddedOutput) Stop() error {
-	close(e.eventChan)
+	if e.eventChan != nil {
+		close(e.eventChan)
+		e.eventChan = nil
+	}
+	if e.errorChan != nil {
+		close(e.errorChan)
+		e.errorChan = nil
+	}
 	if e.outputChan != nil {
 		close(e.outputChan)
+		e.outputChan = nil
 	}
 	return nil
 }
 
-// SetErrorChan sets the error channel
-func (e *EmbeddedOutput) SetErrorChan(ch chan<- error) {
-	e.errorChan = ch
+// ErrorChan returns the input error channel
+func (e *EmbeddedOutput) ErrorChan() chan error {
+	return e.errorChan
 }
 
 // InputChan returns the input event channel

output/embedded/embedded.go

@@ -24,7 +24,7 @@ import (
 )
 
 type LogOutput struct {
-	errorChan    chan<- error
+	errorChan    chan error
 	eventChan    chan event.Event
 	logger       plugin.Logger
 	outputLogger *slog.Logger
@@ -33,8 +33,7 @@ type LogOutput struct {
 
 func New(options ...LogOptionFunc) *LogOutput {
 	l := &LogOutput{
-		eventChan: make(chan event.Event, 10),
-		level:     "info",
+		level: "info",
 	}
 	for _, option := range options {
 		option(l)
@@ -54,6 +53,8 @@ func New(options ...LogOptionFunc) *LogOutput {
 
 // Start the log output
 func (l *LogOutput) Start() error {
+	l.eventChan = make(chan event.Event, 10)
+	l.errorChan = make(chan error)
 	go func() {
 		for {
 			evt, ok := <-l.eventChan
@@ -79,13 +80,20 @@ func (l *LogOutput) Start() error {
 
 // Stop the log output
 func (l *LogOutput) Stop() error {
-	close(l.eventChan)
+	if l.eventChan != nil {
+		close(l.eventChan)
+		l.eventChan = nil
+	}
+	if l.errorChan != nil {
+		close(l.errorChan)
+		l.errorChan = nil
+	}
 	return nil
 }
 
-// SetErrorChan sets the error channel
-func (l *LogOutput) SetErrorChan(ch chan<- error) {
-	l.errorChan = ch
+// ErrorChan returns the input error channel
+func (l *LogOutput) ErrorChan() chan error {
+	return l.errorChan
 }
 
 // InputChan returns the input event channel