feat(processor): new merge stage

docs/pages/stages/index.md

@@ -7,7 +7,7 @@ icon: lucide/library
 Goccia ships stages in three categories:
 
 - [Ingress](ingress/index.md): create messages from external sources.
-- [Processor](processor/index.md): transform, filter, fan out, or reorder messages.
+- [Processor](processor/index.md): transform, filter, fan in, fan out, or reorder messages.
 - [Egress](egress/index.md): deliver messages to external destinations.
 
 ## Message Interfaces
@@ -39,6 +39,7 @@ cfg := processor.NewGenericConfig(goccia.StageRunningModePool)
 `StageRunningModeSingle` runs one executor. `StageRunningModePool` enables the
 generic worker pool and exposes `cfg.Stage.Pool` for worker counts, queue sizes,
 and auto-scaling. Stages with custom runners, such as ingress stages, `tee`,
-`rob`, `sink`, file egress, and TCP egress, document their own execution model.
+`merge`, `rob`, `sink`, file egress, and TCP egress, document their own
+execution model.
 
 For lifecycle details, see [Stages](../concepts/stages.md).

docs/pages/stages/processor/index.md

@@ -11,6 +11,7 @@ Processor stages read from an input connector and write to an output connector.
 | Generic | [Generic](generic.md) | `processor.NewGenericStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | User-defined transform |
 | Generic | [Filter](filter.md) | `processor.NewFilterStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Drop messages by predicate |
 | Generic | [Tee](tee.md) | `processor.NewTeeStage` | Single runner | Clone one stream to many outputs |
+| Generic | [Merge](merge.md) | `processor.NewMergeStage` | Single runner | Combine many streams into one output |
 | CSV | [Decoder](csv-decoder.md) | `processor.NewCSVDecoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Bytes to typed CSV rows |
 | CSV | [Encoder](csv-encoder.md) | `processor.NewCSVEncoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Typed CSV rows to bytes |
 | CAN | [CAN](can.md) | `processor.NewCANStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Raw CAN frames to decoded signals |

docs/pages/stages/processor/merge.md

@@ -0,0 +1,52 @@
+---
+icon: lucide/git-merge
+---
+
+# Merge Processor
+
+`MergeStage` combines multiple input connectors into one output connector.
+
+``` go
+cfg := processor.NewMergeConfig()
+stage := processor.NewMergeStage(
+	[]connector.MessageConnector[*MyMessage]{inA, inB, inC},
+	out,
+	cfg,
+)
+```
+
+## Messages
+
+### Input Message
+
+Accepted body type: `T`.
+
+Additional input interfaces: none required beyond `message.Body`. Every input
+connector must carry the same body type.
+
+### Output Message
+
+Produced body type: the same `T` body received from any input connector.
+
+Additional interfaces: preserved from `T`. The stage does not clone, copy, or
+transform the message. It forwards each incoming message envelope to the output
+connector.
+
+## Configuration
+
+| Field | Default | Description |
+| --- | --- | --- |
+| `OutputQueueSize` | `256` | Size of the internal fan-in ring buffer between input readers and the output bridge. |
+
+At least one input connector is required.
+
+## Internals
+
+`MergeStage` uses a custom single runner instead of the generic worker pool.
+The runner starts one reader goroutine per input connector. Each reader moves
+messages into an internal MPSC fan-in buffer, and one output bridge drains that
+buffer into the output connector.
+
+The stage keeps reading until every input connector is closed and drained. It
+then closes the internal buffer, waits for the output bridge to finish, and
+closes the output connector during stage close.

docs/zensical.toml

@@ -34,6 +34,7 @@ nav = [
       { "Generic" = "stages/processor/generic.md" },
       { "Filter" = "stages/processor/filter.md" },
       { "Tee" = "stages/processor/tee.md" },
+      { "Merge" = "stages/processor/merge.md" },
       { "CSV" = [
         { "Decoder" = "stages/processor/csv-decoder.md" },
         { "Encoder" = "stages/processor/csv-encoder.md" },

processor/merge.go

@@ -0,0 +1,185 @@
+package processor
+
+import (
+	"context"
+	"errors"
+	"sync"
+
+	"github.com/FerroO2000/goccia/connector"
+	"github.com/FerroO2000/goccia/internal/config"
+	"github.com/FerroO2000/goccia/internal/metrics"
+	"github.com/FerroO2000/goccia/internal/rb"
+	"github.com/FerroO2000/goccia/internal/stage"
+	"github.com/FerroO2000/goccia/internal/stage/env"
+)
+
+// ─── Config ─────────────────────────────────────────────────────────────────|
+
+// Default values for the merge stage configuration.
+const (
+	DefaultMergeConfigOutputQueueSize = 256
+)
+
+// MergeConfig structs contains the configuration for the merge stage.
+type MergeConfig struct {
+	// OutputQueueSize is the size of the fan-in ring buffer
+	// placed between the input connectors and the output connector.
+	OutputQueueSize int
+}
+
+// NewMergeConfig returns a new MergeConfig with default values.
+func NewMergeConfig() *MergeConfig {
+	return &MergeConfig{
+		OutputQueueSize: DefaultMergeConfigOutputQueueSize,
+	}
+}
+
+// Validate checks the configuration.
+func (c *MergeConfig) Validate(ac *config.AnomalyCollector) {
+	config.CheckNotNegative(ac, "OutputQueueSize", &c.OutputQueueSize, DefaultMergeConfigOutputQueueSize)
+	config.CheckNotZero(ac, "OutputQueueSize", &c.OutputQueueSize, DefaultMergeConfigOutputQueueSize)
+}
+
+// ─── Environment ────────────────────────────────────────────────────────────|
+
+type mergeEnv struct {
+	*env.BaseEnv[*MergeConfig, *metrics.EmptyMetrics]
+}
+
+func newMergeEnv(config *MergeConfig) *mergeEnv {
+	return &mergeEnv{
+		BaseEnv: env.NewProcessorEnv(config, metrics.NewEmptyMetrics()),
+	}
+}
+
+// ─── Runner ─────────────────────────────────────────────────────────────────|
+
+var _ stage.Runner[*mergeEnv] = (*mergeRunner[msgBody])(nil)
+
+type mergeRunner[T msgBody] struct {
+	*mergeEnv
+
+	inConnectors []msgConn[T]
+	outConnector msgConn[T]
+
+	fanIn *rb.RingBuffer[*msg[T]]
+
+	readersWg *sync.WaitGroup
+
+	runOutputBridgeDone chan struct{}
+	runDone             chan struct{}
+}
+
+func newMergeRunner[T msgBody](inConnectors []msgConn[T], outConnector msgConn[T]) *mergeRunner[T] {
+	return &mergeRunner[T]{
+		inConnectors: inConnectors,
+		outConnector: outConnector,
+
+		readersWg: &sync.WaitGroup{},
+
+		runOutputBridgeDone: make(chan struct{}),
+		runDone:             make(chan struct{}),
+	}
+}
+
+func (mr *mergeRunner[T]) SetEnvironment(env *mergeEnv) {
+	mr.mergeEnv = env
+}
+
+func (mr *mergeRunner[T]) Init(_ context.Context) error {
+	if len(mr.inConnectors) == 0 {
+		return errors.New("no input connector specified")
+	}
+
+	fanInCapacity := uint64(mr.Config.OutputQueueSize)
+	mr.fanIn = rb.NewRingBuffer[*msg[T]](fanInCapacity, rb.BufferKindMPSC)
+
+	return nil
+}
+
+func (mr *mergeRunner[T]) Run(ctx context.Context) {
+	defer close(mr.runDone)
+
+	mr.readersWg.Add(len(mr.inConnectors))
+	for _, inConnector := range mr.inConnectors {
+		go mr.readInput(ctx, inConnector)
+	}
+
+	go mr.runOutputBridge(context.WithoutCancel(ctx))
+
+	mr.readersWg.Wait()
+	mr.fanIn.Close()
+	<-mr.runOutputBridgeDone
+}
+
+func (mr *mergeRunner[T]) readInput(ctx context.Context, inConnector msgConn[T]) {
+	defer mr.readersWg.Done()
+
+	for {
+		msgIn, err := inConnector.Read(ctx)
+		if err != nil {
+			// This means the input connector is closed
+			// and there are no more messages in it
+			return
+		}
+
+		if err := mr.fanIn.Write(msgIn); err != nil {
+			msgIn.Destroy()
+		}
+	}
+}
+
+func (mr *mergeRunner[T]) runOutputBridge(ctx context.Context) {
+	defer close(mr.runOutputBridgeDone)
+
+	for {
+		msg, err := mr.fanIn.Read(ctx)
+		if err != nil {
+			return
+		}
+
+		if err := mr.outConnector.Write(msg); err != nil {
+			msg.Destroy()
+		}
+	}
+}
+
+func (mr *mergeRunner[T]) Close(_ context.Context) {
+	<-mr.runDone
+	mr.outConnector.Close()
+}
+
+func (mr *mergeRunner[T]) Inputs() []uintptr {
+	inputs := make([]uintptr, 0, len(mr.inConnectors))
+
+	for _, inConnector := range mr.inConnectors {
+		inputs = append(inputs, connector.GetConnectorID(inConnector))
+	}
+
+	return inputs
+}
+
+func (mr *mergeRunner[T]) Outputs() []uintptr {
+	return []uintptr{connector.GetConnectorID(mr.outConnector)}
+}
+
+// ─── Stage ──────────────────────────────────────────────────────────────────|
+
+// MergeStage is a stage that merges multiple input connectors into a single
+// output connector.
+// It is the counterpart of the tee stage.
+type MergeStage[T msgBody] struct {
+	*stage.ProcessorStage[T, T, *mergeEnv]
+}
+
+// NewMergeStage returns a new merge processor stage.
+func NewMergeStage[T msgBody](
+	inConnectors []msgConn[T], outConnector msgConn[T], config *MergeConfig,
+) *MergeStage[T] {
+
+	return &MergeStage[T]{
+		ProcessorStage: stage.NewProcessorStageFromRunner[T, T](
+			"merge", newMergeEnv(config), newMergeRunner(inConnectors, outConnector),
+		),
+	}
+}

processor/merge_test.go

@@ -0,0 +1,88 @@
+package processor
+
+import (
+	"testing"
+
+	"github.com/FerroO2000/goccia/connector"
+	"github.com/FerroO2000/goccia/internal/message"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func Test_MergeStage(t *testing.T) {
+	const (
+		connSize         = uint64(32)
+		inputCount       = 3
+		msgCountPerInput = 4
+	)
+
+	inputs := make([]msgConn[*dummyMsg], 0, inputCount)
+	expectedMessages := make(map[*msg[*dummyMsg]]struct{}, inputCount*msgCountPerInput)
+
+	for inputIdx := range inputCount {
+		in := connector.NewRingBuffer[*dummyMsg](connSize)
+		inputs = append(inputs, in)
+
+		for msgIdx := range msgCountPerInput {
+			msgIn := message.NewMessage(&dummyMsg{value: inputIdx*msgCountPerInput + msgIdx})
+			expectedMessages[msgIn] = struct{}{}
+
+			require.NoError(t, in.Write(msgIn))
+		}
+
+		in.Close()
+	}
+
+	out := connector.NewRingBuffer[*dummyMsg](connSize)
+
+	cfg := NewMergeConfig()
+	cfg.OutputQueueSize = 2
+
+	stage := NewMergeStage(inputs, out, cfg)
+	require.NoError(t, stage.Init(t.Context()))
+
+	stage.Run(t.Context())
+	stage.Close(t.Context())
+
+	for range inputCount * msgCountPerInput {
+		msgOut, err := out.Read(t.Context())
+		require.NoError(t, err)
+
+		assert.Contains(t, expectedMessages, msgOut)
+		delete(expectedMessages, msgOut)
+	}
+
+	_, err := out.Read(t.Context())
+	assert.ErrorIs(t, err, connector.ErrClosed)
+	assert.Empty(t, expectedMessages)
+}
+
+func Test_MergeRunner_InitRequiresInput(t *testing.T) {
+	out := connector.NewRingBuffer[*dummyMsg](1)
+	runner := newMergeRunner([]msgConn[*dummyMsg]{}, out)
+	runner.SetEnvironment(newMergeEnv(NewMergeConfig()))
+
+	require.EqualError(t, runner.Init(t.Context()), "no input connector specified")
+}
+
+func Test_MergeConfig_ValidateOutputQueueSize(t *testing.T) {
+	tests := map[string]int{
+		"zero":     0,
+		"negative": -1,
+	}
+
+	for name, outputQueueSize := range tests {
+		t.Run(name, func(t *testing.T) {
+			in := connector.NewRingBuffer[*dummyMsg](1)
+			out := connector.NewRingBuffer[*dummyMsg](1)
+
+			cfg := NewMergeConfig()
+			cfg.OutputQueueSize = outputQueueSize
+
+			stage := NewMergeStage([]msgConn[*dummyMsg]{in}, out, cfg)
+			require.NoError(t, stage.Init(t.Context()))
+
+			assert.Equal(t, DefaultMergeConfigOutputQueueSize, cfg.OutputQueueSize)
+		})
+	}
+}