go_channels 0.1.1
go_channels: ^0.1.1 copied to clipboard
Go-style concurrency for Dart: typed channels, a faithful select over many channel operations, and structured task scopes with cooperative cancellation.
go_channels #
Go-style concurrency for Dart: typed channels, a faithful select over many
channel operations, and structured task scopes with cooperative cancellation.
Dart's dart:async gives you futures and streams, but not the primitives Go
and Kotlin developers reach for: a typed channel, a select that waits on
several operations at once, or a scope where one task's failure cancels its
siblings. go_channels adds those, in plain Dart, on a single isolate.
import 'package:go_channels/go_channels.dart';
final ch = Channel<int>(capacity: 1);
await ch.send(1);
print(await ch.receive()); // 1
Channels #
A Channel<T> passes values between asynchronous tasks. Unbuffered channels
(the default) rendezvous: a send completes only when a receiver takes the
value. Buffered channels hold up to capacity values before a send blocks.
The whole distinction is when send completes, which is what decides whether
the sender feels backpressure the instant the receiver stalls:

final jobs = Channel<String>(); // unbuffered
final queue = Channel<String>(capacity: 32); // buffered
await for (final job in jobs.stream) { // ranges until the channel closes
handle(job);
}
Closing a channel lets receivers drain what is left, then observe closure.
receiveOr mirrors Go's v, ok := <-ch:
final (value, ok) = await ch.receiveOr();
if (!ok) print('channel closed');
select #
select waits on several channel operations and runs exactly one, like Go's
select. If more than one branch is ready, it picks one at random for fairness.
final label = await select<String>((s) {
s.onReceive(jobs, (job, ok) => ok ? 'job: $job' : 'jobs closed');
s.onSend(results, 42, () => 'sent a result');
s.onTimeout(const Duration(seconds: 1), () => 'timed out');
});
Add onDefault to make the whole select non-blocking:
await select<void>((s) {
s.onReceive(events, (e, ok) => handle(e));
s.onDefault(() {}); // returns immediately if nothing is ready
});
Structured task scopes #
withTaskScope runs a group of tasks and returns only once all of them finish.
If any task fails, the scope's token is cancelled so the siblings can stop, and
the first error is rethrown. Nothing spawned inside outlives the scope.
final results = await withTaskScope((scope) async {
final a = scope.spawn((_) => fetchA());
final b = scope.spawn((_) => fetchB());
return [await a, await b];
});
waitAll is the common case in one call:
final pages = await waitAll([
(_) => fetch('/a'),
(_) => fetch('/b'),
(_) => fetch('/c'),
]);
Cancellation #
Dart futures cannot be forcibly killed, so cancellation is cooperative: a task
observes its CancelToken and stops itself. Check isCancelled, call
throwIfCancelled, or await whenCancelled inside a select.
await withTimeout(const Duration(seconds: 5), (token) async {
while (!token.isCancelled) {
await doOneChunk();
}
});
A note on parallelism #
go_channels coordinates asynchronous tasks on one isolate. It does not add
parallelism by itself: use it to structure concurrent work, and combine it with
isolates when you need more than one core. Raw coordination overhead is low
(about 0.5M send/receive round-trips per second on an Apple M-series core;
see benchmark/throughput_benchmark.dart).
Dart is landing shared-memory multithreading (Isolate.runShared, tracked in
dart-lang/sdk#56841). As that
stabilizes, go_channels will offer a shared-memory execution path behind a
capability check, so the same channel and select code can run across threads.
Status #
Version 0.1.0. The API is small and may change before 1.0. Issues and feedback are welcome.