solana_kit_subscribable

Subscribable and observable patterns for the Solana Kit Dart SDK -- a publish/subscribe event system with named channels, Dart Stream bridging, and event demultiplexing.

Note

New Dart-facing APIs should prefer exposing Streams directly. DataPublisher, WritableDataPublisher, and createDataPublisher() remain available as deprecated compatibility APIs. Prefer Stream<T>, StreamController<T>, and ChannelStreamController for new Dart code.

This is the Dart port of @solana/subscribable from the Solana TypeScript SDK.

Installation

Install the package directly:

dart pub add solana_kit_subscribable

If your app uses several Solana Kit packages together, you can also depend on the umbrella package instead:

dart pub add solana_kit

Inside this monorepo, Dart workspace resolution uses the local package automatically.

Documentation

Package page: https://pub.dev/packages/solana_kit_subscribable
API reference: https://pub.dev/documentation/solana_kit_subscribable/latest/
Workspace docs: https://openbudgetfun.github.io/solana_kit/
Package catalog entry: https://openbudgetfun.github.io/solana_kit/reference/package-catalog#solana_kit_subscribable
Source code: https://github.com/openbudgetfun/solana_kit/tree/main/packages/solana_kit_subscribable

For architecture notes, getting-started guides, and cross-package examples, start with the workspace docs site and then drill down into the package README and API reference.

Usage

Preferred: expose Dart Streams

If you are designing a new Dart API, prefer returning Stream<T> directly. Use the DataPublisher primitives in this package when you need to adapt to existing Solana Kit internals or TypeScript-shaped channel publishers.

Stream-native channel controllers

Use ChannelStreamController when you need named channels internally while still exposing Dart Streams to callers.

import 'package:solana_kit_subscribable/solana_kit_subscribable.dart';

Future<void> main() async {
  final channels = ChannelStreamController();

  final subscription = channels.stream<String>('data').listen((message) {
    print('Got message: $message');
  });

  channels.add('data', 'hello');
  // Prints: Got message: hello

  await subscription.cancel();
  await channels.close();
}

Deprecated compatibility: data publishers

The deprecated createDataPublisher() factory returns a WritableDataPublisher that supports both subscribing to and publishing data on named channels. Use this only when maintaining compatibility with DataPublisher-based APIs.

A single publisher supports multiple named channels, and each channel can have multiple subscribers.

import 'package:solana_kit_subscribable/solana_kit_subscribable.dart';

void main() {
  final publisher = createDataPublisher();

  // Subscribe to different channels.
  publisher.on('message', (data) {
    print('Message: $data');
  });

  publisher.on('error', (data) {
    print('Error: $data');
  });

  publisher.on('message', (data) {
    print('Also got: $data');
  });

  publisher.publish('message', 'hello');
  // Prints:
  //   Message: hello
  //   Also got: hello

  publisher.publish('error', 'something failed');
  // Prints:
  //   Error: something failed
}

Combining data and error Streams

The createStreamFromDataAndErrorStreams function creates a broadcast stream that forwards values from a data stream and errors from an error stream. createStreamFromDataPublisher remains available as a compatibility bridge from DataPublisher.

import 'package:solana_kit_subscribable/solana_kit_subscribable.dart';

void main() {
  final publisher = createDataPublisher();

  final stream = createStreamFromDataPublisher<String>(
    StreamFromDataPublisherConfig(
      dataChannelName: 'notification',
      dataPublisher: publisher,
      errorChannelName: 'error',
    ),
  );

  stream.listen(
    (message) => print('Got: $message'),
    onError: (Object error) => print('Error: $error'),
  );

  // Messages are forwarded to the stream.
  publisher.publish('notification', 'update 1');
  // Prints: Got: update 1

  publisher.publish('notification', 'update 2');
  // Prints: Got: update 2

  // Errors are forwarded as stream errors.
  publisher.publish('error', StateError('connection lost'));
  // Prints: Error: Bad state: connection lost
}

Async iterable from a data publisher

The createAsyncIterableFromDataPublisher function creates a single-subscription Stream that closely matches the TypeScript async iterable behavior. It supports abort signals for lifecycle control.

import 'dart:async';

import 'package:solana_kit_subscribable/solana_kit_subscribable.dart';

void main() async {
  final publisher = createDataPublisher();
  final abortCompleter = Completer<void>();

  final stream = createAsyncIterableFromDataPublisher<String>(
    dataPublisher: publisher,
    dataChannelName: 'message',
    errorChannelName: 'error',
    abortSignal: abortCompleter.future,
  );

  // Publish some messages first (they will be queued until listened to).
  publisher.publish('message', 'first');
  publisher.publish('message', 'second');

  // Listen with await for.
  var count = 0;
  await for (final message in stream) {
    print('Got: $message');
    count++;
    if (count >= 2) {
      abortCompleter.complete(); // Stop the stream.
    }
  }
  // Prints:
  //   Got: first
  //   Got: second
}

Demultiplexing events

The demultiplexDataPublisher function splits a single source channel into multiple derived channels using a message transformer. The source subscription is lazy -- it only starts when the first subscriber appears and stops when the last one unsubscribes.

import 'package:solana_kit_subscribable/solana_kit_subscribable.dart';

void main() {
  final sourcePublisher = createDataPublisher();

  // Create a demultiplexed publisher that routes by subscriber ID.
  final demuxed = demultiplexDataPublisher<Map<String, Object?>>(
    sourcePublisher: sourcePublisher,
    sourceChannelName: 'message',
    messageTransformer: (message) {
      final id = message['subscriberId'] as String;
      return ('notification-for:$id', message);
    },
  );

  // Subscribe to notifications for specific subscriber IDs.
  final unsub1 = demuxed.on('notification-for:abc', (data) {
    print('Subscriber abc got: $data');
  });

  demuxed.on('notification-for:xyz', (data) {
    print('Subscriber xyz got: $data');
  });

  // Publish to the source -- messages are routed to the right subscribers.
  sourcePublisher.publish('message', {
    'subscriberId': 'abc',
    'value': 42,
  });
  // Prints: Subscriber abc got: {subscriberId: abc, value: 42}

  sourcePublisher.publish('message', {
    'subscriberId': 'xyz',
    'value': 99,
  });
  // Prints: Subscriber xyz got: {subscriberId: xyz, value: 99}

  // When all subscribers unsubscribe, the source subscription is cancelled.
  unsub1();
}

API Reference

Interfaces

Interface	Description
`ChannelStreamController`	Stream-native named-channel controller for compatibility adapters that still need string-keyed channels.
`DataPublisher`	Deprecated compatibility API. Subscribe to named channels via `on(channelName, subscriber)`, returning `UnsubscribeFn`.
`WritableDataPublisher`	Deprecated compatibility API. Extends `DataPublisher` with `publish(channelName, data)` for emitting events.

Factory functions

Function	Description
`createDataPublisher()`	Deprecated compatibility factory for `WritableDataPublisher` with named channel support.
`createStreamFromDataAndErrorStreams<T>({dataStream, errorStream})`	Creates a broadcast `Stream<T>` from separate data and error streams.
`createStreamFromDataPublisher<T>(config)`	Compatibility bridge that creates a broadcast `Stream<T>` from a `DataPublisher`.
`createAsyncIterableFromDataPublisher<T>({...})`	Creates a single-subscription `Stream<T>` with abort signal support.
`demultiplexStream<TSource, TDestination>({...})`	Splits a source stream into one derived channel stream with lazy subscription.
`demultiplexDataPublisher<T>({sourcePublisher, sourceChannelName, messageTransformer})`	Compatibility bridge that splits one channel into many derived channels.

Type aliases

Type	Description
`UnsubscribeFn`	`void Function()` -- returned by `on()` to unsubscribe a listener.
`Subscriber<T>`	`void Function(T data)` -- a function that receives published data.
`MessageTransformer<T>`	`(String, Object?)? Function(T)` -- transforms a source message into a channel/message pair, or `null` to drop.

Configuration classes

Class	Description
`StreamFromDataPublisherConfig`	Configuration for `createStreamFromDataPublisher`: `dataChannelName`, `dataPublisher`, `errorChannelName`.

Example

Use example/main.dart as a runnable starting point for solana_kit_subscribable.

Import path: package:solana_kit_subscribable/solana_kit_subscribable.dart
This section is centrally maintained with mdt to keep package guidance aligned.
After updating shared docs templates, run docs:update from the repo root.

Maintenance

Validate docs in CI and locally with docs:check.
Keep examples focused on one workflow and reference package README sections for deeper API details.