go_channels 0.1.1

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.