Initial version

Author: floatdrop

.github/workflows/ci.yaml

@@ -0,0 +1,21 @@
+name: CI
+
+on: [push]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24'
+      - name: Install dependencies
+        run: go get .
+      - name: Build
+        run: go build -v ./...
+      - name: Test with the Go CLI
+        run: go test
+        

README.md

@@ -0,0 +1,20 @@
+# demux
+
+[![CI](https://github.com/floatdrop/demux/actions/workflows/ci.yaml/badge.svg)](https://github.com/floatdrop/demux/actions/workflows/ci.yaml)
+[![Go Report Card](https://goreportcard.com/badge/github.com/floatdrop/demux)](https://goreportcard.com/report/github.com/floatdrop/demux)
+[![Go Reference](https://pkg.go.dev/badge/github.com/floatdrop/demux.svg)](https://pkg.go.dev/github.com/floatdrop/demux)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+
+## Installation
+
+```bash
+go get github.com/floatdrop/demux
+```
+
+## Contributing
+
+Contributions are welcome! Please feel free to submit a Pull Request.
+
+## License
+
+This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

dynamic.go

@@ -0,0 +1,29 @@
+package demux
+
+// Dynamic creates dynamic demultiplexer that routes items from 'in' based on keys returned by 'keyFunc'.
+// For each unique key, a new goroutine is spawned running 'consumeFunc'.
+// Each consumeFunc receives a channel that delivers values matching its key.
+func Dynamic[T any, K comparable](
+	in <-chan T,
+	keyFunc func(T) K,
+	consumeFunc func(K, <-chan T),
+) {
+	outChans := make(map[K]chan T)
+	defer func() {
+		for _, ch := range outChans {
+			close(ch)
+		}
+	}()
+
+	for t := range in {
+		key := keyFunc(t)
+		ch, exists := outChans[key]
+		if !exists {
+			ch = make(chan T)
+			outChans[key] = ch
+
+			go consumeFunc(key, ch)
+		}
+		ch <- t
+	}
+}

dynamic_test.go

@@ -0,0 +1,110 @@
+package demux_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/floatdrop/demux"
+)
+
+func TestDynamic_BasicDemuxing(t *testing.T) {
+	type input struct {
+		id   int
+		data string
+	}
+
+	in := make(chan input)
+	var mu sync.Mutex
+	result := make(map[int][]string)
+
+	go demux.Dynamic(in,
+		func(i input) int { return i.id },
+		func(k int, ch <-chan input) {
+			var items []string
+			for item := range ch {
+				items = append(items, item.data)
+			}
+			mu.Lock()
+			result[k] = items
+			mu.Unlock()
+		})
+
+	// Send data
+	go func() {
+		in <- input{id: 1, data: "a"}
+		in <- input{id: 2, data: "b"}
+		in <- input{id: 1, data: "c"}
+		in <- input{id: 2, data: "d"}
+		close(in)
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+
+	// Assertions
+	if len(result) != 2 {
+		t.Fatalf("expected 2 keys, got %d", len(result))
+	}
+	if len(result[1]) != 2 || result[1][0] != "a" || result[1][1] != "c" {
+		t.Errorf("unexpected data for key 1: %v", result[1])
+	}
+	if len(result[2]) != 2 || result[2][0] != "b" || result[2][1] != "d" {
+		t.Errorf("unexpected data for key 2: %v", result[2])
+	}
+}
+
+func TestDynamic_SingleKey(t *testing.T) {
+	in := make(chan int)
+	var collected []int
+	done := make(chan struct{})
+
+	go demux.Dynamic(
+		in,
+		func(i int) string { return "only" },
+		func(k string, ch <-chan int) {
+			for i := range ch {
+				collected = append(collected, i)
+			}
+			close(done)
+		})
+
+	go func() {
+		in <- 10
+		in <- 20
+		in <- 30
+		close(in)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Timeout waiting for consumer")
+	}
+
+	expected := []int{10, 20, 30}
+	for i, val := range expected {
+		if collected[i] != val {
+			t.Errorf("expected %d at index %d, got %d", val, i, collected[i])
+		}
+	}
+}
+
+func TestDynamic_NoInput(t *testing.T) {
+	in := make(chan string)
+	called := false
+
+	go demux.Dynamic(
+		in,
+		func(s string) int { return 0 },
+		func(k int, ch <-chan string) {
+			called = true
+		},
+	)
+
+	close(in)
+
+	time.Sleep(100 * time.Millisecond)
+	if called {
+		t.Error("consumeFunc should not be called for empty input")
+	}
+}

go.mod

@@ -0,0 +1,3 @@
+module github.com/floatdrop/demux
+
+go 1.24.4

static.go

@@ -0,0 +1,18 @@
+package demux
+
+// Static creates Static demultipler that routes each element to channel in channels map by key
+// computed by keyFunc.
+func Static[T any, K comparable](in <-chan T, keyFunc func(T) K, channels map[K]chan<- T) {
+	defer func() {
+		for _, ch := range channels {
+			close(ch)
+		}
+	}()
+
+	for t := range in {
+		k := keyFunc(t)
+		if ch, ok := channels[k]; ok {
+			ch <- t
+		}
+	}
+}

static_test.go

@@ -0,0 +1,90 @@
+package demux_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/floatdrop/demux"
+)
+
+func TestStatic_RoutesToCorrectChannels(t *testing.T) {
+	in := make(chan int)
+	outA := make(chan int, 2)
+	outB := make(chan int, 2)
+
+	channels := map[string]chan<- int{
+		"A": outA,
+		"B": outB,
+	}
+
+	keyFunc := func(i int) string {
+		if i%2 == 0 {
+			return "A"
+		}
+		return "B"
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		demux.Static(in, keyFunc, channels)
+	}()
+
+	// Send data
+	go func() {
+		in <- 1
+		in <- 2
+		in <- 3
+		in <- 4
+		close(in)
+	}()
+
+	wg.Wait()
+
+	// Collect output
+	var aValues, bValues []int
+
+	done := make(chan struct{})
+	go func() {
+		for v := range outA {
+			aValues = append(aValues, v)
+		}
+		for v := range outB {
+			bValues = append(bValues, v)
+		}
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Test timed out")
+	}
+
+	// Validate
+	expectedA := []int{2, 4}
+	expectedB := []int{1, 3}
+
+	if !equal(aValues, expectedA) {
+		t.Errorf("expected A channel values %v, got %v", expectedA, aValues)
+	}
+	if !equal(bValues, expectedB) {
+		t.Errorf("expected B channel values %v, got %v", expectedB, bValues)
+	}
+}
+
+// Helper for slice comparison
+func equal[T comparable](a, b []T) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i, v := range a {
+		if v != b[i] {
+			return false
+		}
+	}
+	return true
+}