1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
QUACKER
=======
Type-safe, composable concurrency patterns with backpressure for Go.
Quacker provides composable, channel-native concurrency primitives.
Backpressure propagates naturally through Go's channel semantics —
no custom protocols, no framework lock-in.
INSTALLATION
------------
go get git.sangtran.io.vn/quacker.git
Requirements: Go 1.25+
PACKAGES
--------
pipeline
Stream processing with backpressure. Compose stages into pipelines.
Each stage transforms a channel into a channel.
Example:
ctx, cancel := context.WithCancelCause(ctx)
out := pipeline.Run(ctx, input,
pipeline.Map(3, func(ctx context.Context, e Event) (Event, bool) {
enriched, err := enrich(ctx, e)
if err != nil {
cancel(err)
return e, false
}
return enriched, true
}),
pipeline.Filter(func(e Event) bool { return e.Priority > 0 }),
pipeline.Sink(5, func(ctx context.Context, e Event) {
store(ctx, e)
}),
)
for range out {}
if err := context.Cause(ctx); err != nil {
log.Fatal(err)
}
Stages: Map, Filter, ForEach, Buffer, Tee, Take, Sink, Batch.
fanout
1-to-N distribution. Distribute a single stream to multiple
consumers. RoundRobin for load balancing, Broadcast for pub/sub.
Example:
fo, _ := fanout.NewFanOut[int](
fanout.WithWorkerCount(3),
fanout.WithStrategy(fanout.Broadcast),
)
fo.Run(ctx, input)
for _, ch := range fo.Outputs() {
go consume(ch)
}
fanin
N-to-1 merge. Merge multiple channels into a single stream.
Example:
fi, _ := fanin.NewFanIn[int]()
fi.Add(stream1)
fi.Add(stream2)
fi.Add(stream3)
merged := fi.Run(ctx)
workerpool
Fixed-size worker pool with push-based task queuing, metrics,
and graceful shutdown.
Example:
pool, _ := workerpool.NewWorkerPool[MyTask](ctx,
workerpool.WithNumWorkers(4),
workerpool.WithTaskQueueSize(100),
)
pool.Start()
pool.Push(task)
pool.StopAndWait()
HOW THEY COMPOSE
----------------
The primitives connect through channels — Go's universal connector:
+-- pipeline --+
source --> fanout --+-- pipeline --+-- fanin --> pipeline --> sink
+-- pipeline --+
Example:
fo.Run(ctx, source)
fast := pipeline.Run(ctx, fo.Outputs()[0], pipeline.Map(10, enrichFast))
slow := pipeline.Run(ctx, fo.Outputs()[1], pipeline.Map(2, enrichSlow))
fi.Add(fast)
fi.Add(slow)
merged := fi.Run(ctx)
out := pipeline.Run(ctx, merged, pipeline.Sink(5, store))
DESIGN
------
- Backpressure is free. Channels block when full. No custom protocol.
- Errors use stdlib. context.WithCancelCause + context.Cause.
- Shutdown is clear. Close input for drain; cancel context for stop.
- Small interfaces. Observer interfaces are 2-3 methods, optional.
- Zero dependencies. Core logic is stdlib only (golang.org/x/sync).
LICENSE
-------
MIT (see LICENSE)
|
