GitLocker: The Coding Marketplace

Description:

rxdart ext

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)

Supported Dart SDK version: >=2.12.0 <4.0.0 #
RxDart compatibility #

rxdart
rxdart_ext

^0.27.2
0.1.2

^0.27.3
0.1.3 → 0.2.0

^0.27.4
0.2.1 → 0.2.2

^0.27.5
0.2.3 → 0.2.9

^0.28.0
^0.3.0

The latest version of rxdart_ext always works fine with the latest version of rxdart.

Tip
For example: when using rxdart: ^0.27.4 → we can use rxdart_ext: ^v where v in 0.2.1 → 0.2.9
(i.e. all rows since ^0.27.4 row in the above table).
But in some cases there is any conflict between rxdart version and rxdart_ext version,
you must use stricter version, e.g. rxdart: ^0.27.4 → rxdart_ext: ^v where v in 0.2.1 → 0.2.2
(same row in the above table)

API - Documentation #
1. Single #
A Stream which emits single event, either data or error, and then close with a done-event.
Success case: ------------(*)|------
data done

Failure case: ------------(x)|------
error done
copied to clipboard

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

Single is suitable for one-shot operations (likes Future but lazy - executes when listening), eg. making API request, reading local storage, ...
import 'package:http/http.dart' as http;

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

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 (returns a Single instead of Stream)

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

2. Operators for Stream #

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

3. StateStream #
A Stream that provides synchronous access to the last emitted item,
and two consecutive values are not equal.
The equality between previous data event and current data event is determined by StateStream.equals.
This Stream always has no error.

Broadcast

StateSubject
StateConnectableStream

publishState
shareState

Single-subscription

toStateStream

Example
Useful for Flutter BLoC pattern - StreamBuilder, expose broadcast state stream to UI, can synchronous access to the last emitted item, and distinct until changed

✅ Distinct: distinct until changed.
✅ Value: can synchronous access to the last emitted item.
✅ NotReplay: not replay the latest value.
✅ Connectable: broadcast stream - can be listened to multiple time.

Stream (dart:core)
^
|
|
|--------------------------------------------|
| |
| |
ValueStream (rxdart) |
^ |
| |
| |
NotReplayValueStream (rxdart_ext) |
^ ConnectableStream (rxdart)
| ^
| |
StateStream (rxdart_ext) |
^ |
| |
|------------ -----------|
| |
| |
StateConnectableStream (rxdart_ext)
copied to clipboard
class UiState { ... }

final Stream<UiState> state$ = ...;

final StateConnectableStream<UiState> state$ = state$.publishState(UiState.initial());
final connection = state$.connect();

StreamBuilder<UiState>(
initialData: state$.value,
stream: state$,
builder: (context, snapshot) {
final UiState state = snapshot.requireData;

return ...;
},
);
copied to clipboard
See also flutter_bloc_pattern/RxStreamBuilder,
it can be used with StateStream perfectly and more easily (don't require initialData and don't need to call snapshot.requireData).
final StateStream<UiState> state$;

RxStreamBuilder<UiState>(
stream: state$,
builder: (context, UiState state) {
// use state directly
return ...;
}
);
copied to clipboard
4. NotReplayValueStream #
A Stream that provides synchronous access to the last emitted item, but not replay the latest value.

Broadcast

ValueSubject
NotReplayValueConnectableStream

publishValueNotReplay
shareValueNotReplay

Single-subscription

ValueStreamController
toNotReplayValueStream

5. Utils #
DisposableMixin
A mixin that makes it easy to dispose streams without having to store and close 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'),
);
}
}
copied to clipboard
License #
MIT License

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