rxdart_ext 0.4.0

Some extension methods and classes built on top of RxDart - RxDart extension.

rxdart_ext

Author: Petrus Nguyễn Thái Học

Some extension methods and classes built on top of RxDart - RxDart extension.

Liked some of my work? Buy me a coffee (or more likely a beer)

RxDart and Dart SDK compatibility

rxdart_ext	rxdart	Dart SDK
0.1.2	^0.27.2	>=2.12.0 <4.0.0
0.1.3 → 0.2.0	^0.27.3	>=2.12.0 <4.0.0
0.2.1 → 0.2.2	^0.27.4	>=2.12.0 <4.0.0
0.2.3 → 0.2.9	^0.27.5	>=2.12.0 <4.0.0
^0.3.0	^0.28.0	>=2.12.0 <4.0.0
^0.4.0	^0.28.0	^3.0.0

rxdart_ext 0.4.x supports rxdart ^0.28.0 and requires Dart ^3.0.0.

NOTE: 0.4.0 introduces Dart 3 class modifiers. NotReplayValueStream, StateStream and MutableStateStream can be implemented, but cannot be extended outside their defining libraries. See CHANGELOG.md for all inheritance restrictions and migration notes.

Add the package with dart pub add rxdart_ext (or flutter pub add rxdart_ext for Flutter projects), then import:

import 'package:rxdart_ext/rxdart_ext.dart';

The main library also exports RxDart. You can import individual libraries such as package:rxdart_ext/single.dart or package:rxdart_ext/state_stream.dart when you only need part of the package.

API - Documentation

1. Single

A Stream which emits a single event, either data or error, and then closes with a done-event.

Success case: ------------(*)|------
                         data done

Failure case: ------------(x)|------
                        error done

NOTE: Single extends Stream, so all operators and transformers for Stream are available for Single as well.

Single is suitable for one-shot operations (like Future), eg. making an API request, reading local storage, ...

Single can also model a deferred/lazy computation. Use Single.fromCallable or Single.defer to start the computation only when listening. Single.fromFuture wraps a future that has already been created.

The computation's lifetime can be managed through its StreamSubscription. Calling cancel() stops listening and propagates cancellation upstream, so the source can stop pending work and release its resources. This makes it easier to manage the computation, eg. when a BLoC or widget is disposed. Use await subscription.cancel() to wait for the source's cleanup.

Cancellation of the underlying work depends on the source. When wrapping a plain Future, including one returned by Single.fromCallable, cancelling the subscription stops result delivery but does not abort the future's work.

For example, assuming your app defines baseUrl and User.fromJson:

import 'dart:convert';
import 'dart:io';

import 'package:http/http.dart' as http;
import 'package:rxdart_ext/rxdart_ext.dart';

Single<User> fetchUser(String id) {
  return Single.fromCallable(() => http.get(Uri.parse('$baseUrl/users/$id')))
      .flatMapSingle<String>((res) => res.statusCode == HttpStatus.ok
          ? Single.value(res.body)
          : Single.error(Exception('Cannot fetch user with id=$id')))
      .map((body) => User.fromJson(jsonDecode(body) as Map<String, dynamic>));
}

• Create Single

  • Factory constructors.

    • Single.unsafeFromStream
    • Single.value
    • Single.error
    • Single.fromFuture
    • Single.fromCallable
    • Single.timer
    • Single.defer
    • Single.retry

  • Static methods provided by RxSingles class

    • RxSingles.zip2
    • RxSingles.forkJoin2 ..RxSingles.forkJoin9 and RxSingles.forkJoinList
    • RxSingles.using

  • Convert others to Single via extensions.

    • Stream.singleOrError
    • Future.asSingle
    • (FutureOr<T> Function()).asSingle

• Operators for Single (return a Single instead of Stream)

  • asNullable
  • flatMapSingle
  • flatMapEitherSingle
  • asyncExpandSingle
  • switchMapSingle
  • exhaustMapSingle
  • debug
  • delay
  • doOnCancel
  • doOnData
  • doOnError
  • doOnListen
  • onErrorReturn
  • onErrorReturnWith
  • onErrorResumeSingle
  • onErrorResumeNextSingle
  • mapTo
  • toEitherSingle
  • asVoid

2. Operators for Stream

• asNullable
• debug, collect
• distinctUniqueBy
• distinctBy
• doOn
• doneOnError
• flatMapBatches
• flatMapBatchesSingle
• ignoreErrors
• mapNotNull (moved to rxdart 0.27.4 as standard operator: mapNotNull)
• toSingleSubscriptionStream
• asVoid
• whereNotNull (moved to rxdart 0.27.4 as standard operator: whereNotNull)

asNullable() adapts a Stream<T> to Stream<T?> without wrapping the source stream. Useful when an operator needs a nullable seed value:

final source$ = Stream.fromIterable([1, 2, 3]);
final StateConnectableStream<int?> state$ = source$.asNullable().publishState(null);

// The Single version preserves the Single type and the original instance.
final Single<int?> single = Single.value(1).asNullable();

3. StateStream

A Stream that provides synchronous access to the latest value, and skips consecutive equal values. The equality between the previous value and the current value is determined by StateStream.equals, using == by default. This Stream always has no error and does not replay the latest value to new listeners.

• Broadcast
  • StateSubject
  • StateConnectableStream
    • publishState
    • shareState
• Single-subscription
  • toStateStream

StateSubject also implements MutableStateStream, so you can assign value directly or use update, getAndUpdate and updateAndGet. getAndUpdate returns the previous value; updateAndGet returns the new value.

Example

Useful for Flutter BLoC pattern - StreamBuilder: expose a broadcast state stream to the UI, with synchronous access to the latest value and distinct until changed behavior.

• ✅ Distinct: distinct until changed.
• ✅ Value: synchronous access to the latest value.
• ✅ NotReplay: does not replay the latest value to new listeners.
• ✅ Connectable: publishState creates a broadcast stream that starts when connect() is called.

The arrows below show subtype relationships:

                                Stream (dart:async)
                                   ^
                                   |
                                   |
            |--------------------------------------------|
            |                                            |
            |                                            |
        ValueStream (rxdart)                             |
            ^                                            |
            |                                            |
            |                                            |
    NotReplayValueStream (rxdart_ext)                    |
            ^                                    ConnectableStream (rxdart)
            |                                            ^
            |                                            |
       StateStream (rxdart_ext)                          |
            ^                                            |
            |                                            |
            |------------                     -----------|
                        |                     |
                        |                     |
                     StateConnectableStream (rxdart_ext)

In a Flutter State<ExamplePage>, keep the stream and its connection as fields:

import 'dart:async';

import 'package:flutter/widgets.dart';
import 'package:rxdart_ext/rxdart_ext.dart';

class UiState { ... }

class _ExamplePageState extends State<ExamplePage> {
  late final StateConnectableStream<UiState> state$;
  late final StreamSubscription<UiState> connection;

  @override
  void initState() {
    super.initState();
    final Stream<UiState> source$ = ...;
    state$ = source$.publishState(UiState.initial());
    connection = state$.connect();
  }

  @override
  void dispose() {
    unawaited(connection.cancel());
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return StreamBuilder<UiState>(
      initialData: state$.value,
      stream: state$,
      builder: (context, snapshot) {
        final UiState state = snapshot.requireData;

        return ...;
      },
    );
  }
}

Create the connectable stream once in initState() and cancel connection in dispose(). State.dispose() is synchronous, so use unawaited for the cleanup future returned by connection.cancel(). Pass state$.value as initialData because the stream does not replay it. This avoids a duplicate initial event that could trigger an unnecessary rebuild. Use shareState when you want the stream to connect automatically on the first listener and disconnect when the last listener cancels.

See also flutter_bloc_pattern/RxStreamBuilder, it can be used with the same StateStream more easily (no need for initialData or snapshot.requireData).

NOTE: flutter_bloc_pattern 3.0.0 depends on rxdart_ext ^0.3.0. With rxdart_ext 0.4.x, the example below requires a compatible flutter_bloc_pattern release.

// Inside build(), using the state$ field from the example above:

RxStreamBuilder<UiState>(
  stream: state$,
  builder: (context, UiState state) {
    // use state directly
    return ...;
  },
);

4. NotReplayValueStream

A Stream that provides synchronous access to the latest value, but does not replay it to new listeners. ValueSubject and ValueStreamController require a seed value, so their streams have a value even before the first event.

• Broadcast
  • ValueSubject
  • NotReplayValueConnectableStream
    • publishValueNotReplay
    • shareValueNotReplay
• Single-subscription
  • ValueStreamController
  • toNotReplayValueStream

5. Utils

DisposableMixin

A mixin that makes it easy to cancel stream subscriptions without having to store a StreamSubscription variable.

Typical usage is as follows:

class DisposableExample with DisposableMixin {
  DisposableExample({
    required Stream<DateTime> dateTimeStream,
  }) {
    dateTimeStream.takeUntil(dispose$).listen(
          (value) => print('Disposable example: $value'),
        );
  }
}

Call dispose() when the owner is no longer needed. This emits on dispose$, so takeUntil(dispose$) cancels the subscription. If you override dispose(), call super.dispose().

For example, in the State of a Flutter DisposablePage widget:

import 'package:flutter/widgets.dart';

class _DisposablePageState extends State<DisposablePage> {
  late final DisposableExample _disposableExample;

  @override
  void initState() {
    super.initState();
    _disposableExample = DisposableExample(
      dateTimeStream: Stream<DateTime>.periodic(
        const Duration(seconds: 1),
        (_) => DateTime.now(),
      ),
    );
  }

  @override
  void dispose() {
    _disposableExample.dispose();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) => const SizedBox.shrink();
}

License

MIT License

Copyright (c) 2020-2022 Petrus Nguyễn Thái Học