solana_kit_subscribable 0.5.0

Subscribable and observable pattern for the Solana Kit Dart SDK.

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.