Connect

A highly powerful connector utility that allows you to dynamically stream and pipe multiple asynchronous streams directly into a single reactive Signal.

The concept is inspired by Angular Signals integration with RxJS streams.

Start with an existing mutable Signal and call connect(signal) to create a connector instance.

1. Chaining Streams#

You can bind multiple streams to feed the same destination signal. The connector will handle the subscription management for all streams seamlessly.

final counter = signal(0);
final connector = connect(counter);

final fastStream = Stream.periodic(Duration(seconds: 1), (i) => i);
final slowStream = Stream.periodic(Duration(seconds: 5), (i) => i * 10);

// Values from both streams will be piped into the counter signal!
connector.from(fastStream).from(slowStream);

2. The Shift Operator (<<)#

For a more concise and beautiful visual flow, you can use the shift operator (<<) to chain streams:

final s = signal(0);
final c = connect(s);

c << fastStream << slowStream;

3. Lifecycle and Disposal#

To avoid memory leaks, make sure to dispose the connector when it is no longer needed. Disposing the connector will automatically cancel all underlying active stream subscriptions.

connector.dispose(); // Cancels all stream subscriptions safely

Constructors#

Properties#

Methods#

final counter = signal(0);
final c = connect(counter);

final s1 = Stream.value(1);
final s2 = Stream.value(2);

c.from(s1).from(s2);

c.dispose();

connect#

The idea for connect comes from Anguar Signals with RxJS:

Start with a signal and then use the connect method to create a connector. Streams will feed Signal value.

final s = signal(0);
final c = connect(s);

to#

Add streams to the connector.

final s = signal(0);
final c = connect(s);

final s1 = Stream.value(1);
final s2 = Stream.value(2);

c.from(s1).from(s2); // These can be chained

dispose#

Cancel all subscriptions.

final s = signal(0);
final c = connect(s);

final s1 = Stream.value(1);
final s2 = Stream.value(2);

c.from(s1).from(s2);
// or
c << s1 << s2

c.dispose(); // This will cancel all subscriptions