From 02123df8cc15896f97c1580e49d8e2a991e37ddd Mon Sep 17 00:00:00 2001 From: sangtran Date: Thu, 3 Sep 2026 23:10:33 +0700 Subject: Moving all legacy quacker source code from github to cgit --- README | 125 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 README (limited to 'README') diff --git a/README b/README new file mode 100644 index 0000000..c758b17 --- /dev/null +++ b/README @@ -0,0 +1,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) -- cgit v1.3.1