The Reactive Extensions for Dart

ReactiveX

Last update: Dec 20, 2022

Related tags

Templates rxdart

Overview

RxDart

About

RxDart extends the capabilities of Dart Streams and StreamControllers.

Dart comes with a very decent Streams API out-of-the-box; rather than attempting to provide an alternative to this API, RxDart adds functionality from the reactive extensions specification on top of it.

RxDart does not provide its Observable class as a replacement for Dart Streams. Instead, it offers several additional Stream classes, operators (extension methods on the Stream class), and Subjects.

If you are familiar with Observables from other languages, please see the Rx Observables vs. Dart Streams comparison chart for notable distinctions between the two.

Upgrading from RxDart 0.22.x to 0.23.x

RxDart 0.23.x moves away from the Observable class, utilizing Dart 2.6's new extension methods instead. This requires several small refactors that can be easily automated -- which is just what we've done!

Please follow the instructions on the rxdart_codemod package to automatically upgrade your code to support RxDart 0.23.x.

How To Use RxDart

For Example: Reading the Konami Code

import 'package:rxdart/rxdart.dart';

void main() {
  const konamiKeyCodes = <int>[
    KeyCode.UP,
    KeyCode.UP,
    KeyCode.DOWN,
    KeyCode.DOWN,
    KeyCode.LEFT,
    KeyCode.RIGHT,
    KeyCode.LEFT,
    KeyCode.RIGHT,
    KeyCode.B,
    KeyCode.A,
  ];

  final result = querySelector('#result')!;

  document.onKeyUp
      .map((event) => event.keyCode)
      .bufferCount(10, 1) // An extension method provided by rxdart
      .where((lastTenKeyCodes) => const IterableEquality<int>().equals(lastTenKeyCodes, konamiKeyCodes))
      .listen((_) => result.innerHtml = 'KONAMI!');
}

API Overview

RxDart adds functionality to Dart Streams in three ways:

Stream Classes - create Streams with specific capabilities, such as combining or merging many Streams.
Extension Methods - transform a source Stream into a new Stream with different capabilities, such as throttling or buffering events.
Subjects - StreamControllers with additional powers

Stream Classes

The Stream class provides different ways to create a Stream: Stream.fromIterable or Stream.periodic. RxDart provides additional Stream classes for a variety of tasks, such as combining or merging Streams!

You can construct the Streams provided by RxDart in two ways. The following examples are equivalent in terms of functionality:

Instantiating the Stream class directly.
- Example: final mergedStream = MergeStream([myFirstStream, mySecondStream]);
Using static factories from the Rx class, which are useful for discovering which types of Streams are provided by RxDart. Under the hood, these factories call the corresponding Stream constructor.
- Example: final mergedStream = Rx.merge([myFirstStream, mySecondStream]);

List of Classes / Static Factories

CombineLatestStream (combine2, combine3... combine9) / Rx.combineLatest2...Rx.combineLatest9
ConcatStream / Rx.concat
ConcatEagerStream / Rx.concatEager
DeferStream / Rx.defer
ForkJoinStream (join2, join3... join9) / Rx.forkJoin2...Rx.forkJoin9
FromCallableStream / Rx.fromCallable
MergeStream / Rx.merge
NeverStream / Rx.never
RaceStream / Rx.race
RangeStream / Rx.range
RepeatStream / Rx.repeat
RetryStream / Rx.retry
RetryWhenStream / Rx.retryWhen
SequenceEqualStream / Rx.sequenceEqual
SwitchLatestStream / Rx.switchLatest
TimerStream / Rx.timer
UsingStream / Rx.using
ZipStream (zip2, zip3, zip4, ..., zip9) / Rx.zip...Rx.zip9
If you're looking for an Interval equivalent, check out Dart's Stream.periodic for similar behavior.

Extension Methods

The extension methods provided by RxDart can be used on any Stream. They convert a source Stream into a new Stream with additional capabilities, such as buffering or throttling events.

Example

Stream.fromIterable([1, 2, 3])
  .throttleTime(Duration(seconds: 1))
  .listen(print); // prints 3

List of Extension Methods

Subjects

Dart provides the StreamController class to create and manage a Stream. RxDart offers two additional StreamControllers with additional capabilities, known as Subjects:

BehaviorSubject - A broadcast StreamController that caches the latest added value or error. When a new listener subscribes to the Stream, the latest value or error will be emitted to the listener. Furthermore, you can synchronously read the last emitted value.
ReplaySubject - A broadcast StreamController that caches the added values. When a new listener subscribes to the Stream, the cached values will be emitted to the listener.

Rx Observables vs Dart Streams

In many situations, Streams and Observables work the same way. However, if you're used to standard Rx Observables, some features of the Stream API may surprise you. We've included a table below to help folks understand the differences.

Additional information about the following situations can be found by reading the Rx class documentation.

Situation	Rx Observables	Dart Streams
An error is raised	Observable Terminates with Error	Error is emitted and Stream continues
Cold Observables	Multiple subscribers can listen to the same cold Observable, and each subscription will receive a unique Stream of data	Single subscriber only
Hot Observables	Yes	Yes, known as Broadcast Streams
Is {Publish, Behavior, Replay}Subject hot?	Yes	Yes
Single/Maybe/Completable ?	Yes	Yes, uses rxdart_ext Single (`Completable = Single<void>, Maybe<T> = Single<T?>`)
Support back pressure	Yes	Yes
Can emit null?	Yes, except RxJava	Yes
Sync by default	Yes	No
Can pause/resume a subscription*?	No	Yes

Examples

Web and command-line examples can be found in the example folder.

Web Examples

In order to run the web examples, please follow these steps:

Clone this repo and enter the directory
Run dart pub get
Run dart pub run build_runner serve example
Navigate to http://localhost:8080/web/index.html in your browser

Command Line Examples

In order to run the command line example, please follow these steps:

Clone this repo and enter the directory
Run pub get
Run dart example/example.dart 10

Flutter Example

Install Flutter

To run the flutter example, you must have Flutter installed. For installation instructions, view the online documentation.

Run the app

Open up an Android Emulator, the iOS Simulator, or connect an appropriate mobile device for debugging.
Open up a terminal
cd into the example/flutter/github_search directory
Run flutter doctor to ensure you have all Flutter dependencies working.
Run flutter packages get
Run flutter run

Notable References

Changelog

Refer to the Changelog to get all release notes.

Comments

Whats the motivation for creating ValueWrapper and breaking backwards compatibility with ValueStream?

ValueWrapper seems to do nothing other than wrap another value, which ValueStream also did.

So why create it? Was it really necessary to break everyone's code due to this seemingly useless class? I'm genuinely curious as to why this was added.

So our code is going from _valueStream.value to _valueStream.valueWrapper.value.

opened by larssn 43

Deadlocks in tests after upgrading rxdart

Took another stab at upgrading off rxdart 0.23.1, but unfortunately still getting some failing tests. Here's a repro/comparison (could probably be minimized further but not sure which parts are relevant).

import 'package:meta/meta.dart';
import 'package:moor/ffi.dart';
import 'package:moor/moor.dart';
import 'package:rxdart/rxdart.dart';
import 'package:test/test.dart';

class TestDb extends GeneratedDatabase {
  TestDb() : super(SqlTypeSystem.withDefaults(), VmDatabase.memory());
  @override final List<TableInfo> allTables = const [];
  @override final int schemaVersion = 1;
}

Future<void> testCase({
  @required Stream<int> Function() createOuter,
  @required Stream<int> Function() createInner,
}) async {
  final log = <int>[];
  final timeout = Duration(milliseconds: 100);

  final a = createOuter();
  final b = a.switchMap((_) => createInner());

  b.listen(log.add);
  await b.first.then(log.add)
    .timeout(timeout, onTimeout: () => fail('1st should complete'));
  expect(log, [2, 2]);

  b.listen(log.add);
  await b.first.then(log.add)
    .timeout(timeout, onTimeout: () => fail('2nd should complete'));
  expect(log, [2, 2, 2, 2]);
}

void main() {
  group('rxdart upgrade', () {
  
    test("moor", () async {
      final db = TestDb();
      Stream<int> selectInt(int i) => db
        .customSelect('select $i a')
        .map((row) => row.read<int>('a'))
        .watchSingle();
      await testCase(
        createOuter: () => selectInt(1),
        createInner: () => selectInt(2),
      );
    });

    test("rxdart", () async {
      final outer = BehaviorSubject<int>();
      final tc = testCase(
        createOuter: () => outer,
        createInner: () {
          final inner = BehaviorSubject<int>();
          Future.delayed(Duration(milliseconds: 10)).then((_) => inner.add(2));
          return inner;
        }
      );
      await Future.delayed(Duration(milliseconds: 10));
      outer.add(1);
      await tc;
    });
  });
}

rxdart 0.23.1: Both tests pass
rxdart 0.24.0 and 0.24.1: Both tests fail with "2nd should complete"
rxdart 0.25.0, 0.26.0, and 0.27.0: Only the moor test fails
All the tests were done with moor 4.3.1, but I also spot checked moor 3.4.0 + rxdart 0.27.0 and I get the same result.

Since this began after upgrading rxdart I've started by creating an issue here, but @simolus3 might be able to help determine if this is a moor bug.

#477 and #500 might have some context from the last time we tried to upgrade off rxdart 0.23.1.

I looked around for similar issues since last time and came #511 looks similar. The 1st repro in that issue still times out with rxdart 0.27.0, even though the 2nd repro looks fixed.

opened by Mike278 33

CombineLatest doesn't really work out
combineLatest2() doesn't really work out it return latest value and not merge two values to one.

Exemple:

Stream get submitValid => Observable.combineLatest2(email, password, (e, p) => true);

Enter an email value that is valid eg. "ed@" and then enter a password that is valid. My code is set to validate at length >= 5 so "qwerty" becomes valid on the 't'

delete the password until the field is now empty on screen and go back to the email section from there.

delete the email until the char before '@' to make it invalid.

re-enter the '@' character and hey presto the button is now enabled but the password field is showing red error message on screen and the value of it is still empty yet the button is enabled to be clicked.
opened by iamarnas 33
Testing of BehaviorSubject with seed or first `null` value
I can't seem to test properly that BehaviorSubject emits null value first, either with add method or with seed value. I'm not sure if this is I'm doing wrong, but here's what I have:

test("Can expect initial non-null value", () { final subject = BehaviorSubject<String>(seedValue: "value"); expect(subject, emits("value")); subject.close(); }); test("Can expect initial null value", () { final subject = BehaviorSubject<String>(seedValue: null); expect(subject, emits(null)); subject.close(); });

For the non-null value the test passes. For the null value, the assertion fails because the stream is closed. If the stream is not closed, the test timeouts.

I feel something is off with null values being pushed as initial or first values to the subject, for example this:

test("Print initial values", () { BehaviorSubject<String>(seedValue: null) ..listen(print) ..add(null) ..add(null) ..close(); });

will print null only twice. When I replace nulls with instances, the value is printed three times.

Overall seems like the existing null value isn't pushed to new subscribers. I haven't found it documented that for null it'd behave differently, and also I haven't found any information about special handling of null for RxDart in general.
opened by lwasyl 30

Combine connected query

Stream matchCombine;

@override
  void initState() {
    super.initState();
    matchCombine = _firestore
        .collection('mat')
        .snapshots()
        .map((convert) {
      return convert.documents.map((f) {

        Observable<Matches> match = f.reference
            .snapshots()
            .map<Matches>((document) => Matches.fromMap(document.data));

        Observable<User> user = _firestore
            .collection("u")
            .document()
            .snapshots()
            .map<User>((document) => User.fromMap(document.data));

        Observable<Message> message = _firestore
            .collection('mat')
            .document(f.documentID)
            .collection("c")
            .orderBy('date', descending: true)
            .limit(1)
            .snapshots()
            .expand((snapShot) => snapShot.documents)
            .map<Message>((document) => Message.fromMap(document.data));

        return Observable.combineLatest3(match, user, message,
                (matches, user, message) => CombinedStream(matches, user, message));
      });
    }).map((convert) {
      return Observable.combineLatestList(convert);
    });
  }

How can i combine this. I've searched everywhere but couldn't find a solution. Where am I doing wrong

opened by cnkygmr 25

Extension Methods phase 1
Summary

Introduce *Extension extension classes for each Stream / StreamTransformer that provides a transform method.

Move all Rx operators from Observable to ObservableExtensions

Remove all "wrapper" methods from Observable

Default to Stream Implementations

Keep constructors on Observable class (for now?)

Remove Exported Futures & Tests

AsObservable Future no longer provides utility, since all streams will have Rx methods

Min and Max are easily inlined

Breaking Changes

All methods that previously returned an Observable<T> now return a plain Stream<T>

All Future implementations removed.

Discussion

I'd love to hear feedback on this change -- good or bad! This PR will result in some breaking changes, but I've tried to make them as painless as possible for this first phase. If these changes are too drastic, please let me know and we can discuss!

Rationale

In practice, this PR should make it much easier to use Rx transformations, since they will be available to all streams by importing rxdart.

import 'package:rxdart/rxdart.dart'; // Uses the `max` extension method // Prints 3 Stream.fromIterable([1, 2, 3]).max().then(print); // Uses the `onErrorReturn` extenion method // Prints "Recovery Value" Stream<String>.error(Exception()).onErrorReturn("Recovery Value").listen(print);

Background

RxDart is built upon Stream primitives. All operators are implemented as StreamTransformer classes, such as ScanStreamTransformer. All constructors are implement as regular Stream classes, such as RangeStream.

The Observable class itself is simply a wrapper around a Stream that makes use of these primitives to add additional methods to the Stream it wraps. This includes the scan method, or the Observable.range constructor.

With the advent of Dart 2.6, we can take advantage of Static Extension Methods to add the rx transformation methods onto the Stream class, without requiring end-users to wrap their Streams in an Observable!

This means we can remove almost all methods that proxy up to the Stream class, and move the Rx operators to an extension.
opened by brianegan 24
More methods

Hi,

awesome work in progress with this library!

Just wondering when you intend to add more methods, like: "distinctBy", "groupBy", "bufferByTime", "delay", "swithMap", "retryWhen" as they would really help!

Thank you and good job again!
in progress

opened by qwilbird 19
shareValue() only works if source is declared as a field

Hey folks,

It appears that shareValue() only works for sources declared as fields. If the source is either a method or just a getter, the sync value returns null. You can find a quick gist featuring this behavior at https://gist.github.com/RollyPeres/b39034b1abcb55eda0605eae072b3be1 Tested on rxdart: ^0.23.1.

Would like to hear your thoughts @frankpepermans , @brianegan . Cheers!

opened by narcodico 16
Add `mapNotNull` operator
Sometimes is interesting to perform complex operations inside map and for some values you just want to ignore. Today we need to methods to acomplish this: map and a where afterwards, like:

enum Result { IS_EVEN, IS_MULTIPLE_OF_3 } void main(List<String> arguments) async { var stream = Stream.fromIterable([0, 1, 2, 3, 4, 5]); stream.map((value) { if (value.isEven) { return Result.IS_EVEN; } if (value % 3 == 0) { return Result.IS_MULTIPLE_OF_3; } return null; }).where((it) => it != null); }

To avoid an addition of a new trasnformer onto the stream chain we could just implement an mapNotNull method, that will ignore the elements that returned null through the convert lambda.
opened by shinayser 16
Extension methods phase 2
Now that we have all the extension methods in place, it's time to talk about a larger breaking change: Removing the Observable class.

I understand there's some hesitation around this change, but I'd ask folks to read the following information and take a look at the code changes -- personally, I think these changes will make your code and RxDart code much leaner and less confusing in the long run (as can be seen by the number of lines removed vs added!), but I'd love to hear feedback as to why you might disagree!

Rationale

Many folks might not know that the Observable class is just a wrapper class. It does not exhibit any special behavior compared to Streams, since Observable is a Stream! With extension methods, it's not actually needed at all :)

However, many folks believe the Observable class we provide will work exactly like other Rx Observables, which terminate on error and can be subscribed to more than once. Given this is a common source of confusion, I propose we remove the Observable class altogether.

Background

In reality, the Observable class serves only two purposes:

To wrap the source Stream provided to the constructor. In the past, this allowed us to add additional methods to the wrapped Stream.

To provide factory constructors and static factories. These were used to create Streams provided by the standard library or RxDart, ensuring they were wrapped in an Observable.

With extension methods, users no longer need to wrap Streams in an Observable to use the additional methods provided by RxDart!

Therefore, at this point, the only purpose of the Observable class is to provide Stream factories. However, even these these Stream factories no longer need to wrap the Streams they create in an Observable class! For example, you no longer need to use Observable.fromIterable, you could just use Stream.fromIterable and the functionality will be identical.

Solution

Therefore, I suggest we remove the Observable class entirely and move all factories to a new Rx class. Remove all constructors that simply create Streams from the Standard library, e.g. remove Observable.fromIterable and replace it with Stream.fromIterable

From listening to feedback from all you fine folks (thanks to all who wrote us!), it seems people want the factories to stick around. My proposal: Move all factories to a new class called Rx and make them all static. This will mean creating streams provided by this library will still be nice and concise, yet easily discoverable.

Before and After Comparisons

|Before|After| |------|-----| |Observable.just(1)|Stream.value(1)| |Observable.error(Exception())|Stream.error(Exception())| |Observable.fromIterable([1, 2, 3])|Stream.fromIterable([1,2,3])| |Observable.combineLatest2|Rx.combineLatest2| |Observable<int>.concat|Rx.concat<int>|

Migration Strategy

Since this would require a lot of changes, I plan on writing a codemod (thanks to robrbecker for the suggestion!). The codemod would go through your codebase and update all necessary constructors.

I also plan on doing a search for blog posts and articles on RxDart and plan on contacting authors asking them if they could update the code snippets to the latest version.

List of Changes

Remove the Observable class. With extensions, you no longer need to wrap Streams in an Observable!

Convert all factories to static constructors to aid in discoverability of Stream classes

Move all factories to an Rx class.

Remove Observable.just, use Stream.value

Remove Observable.error, use Stream.error

Remove all tests that check base Stream methods

Subjects and *Observable classes extend Stream instead of base Observable

Rename *Observable to *Stream to reflect the fact they're just Streams.

ValueObservable -> ValueStream

ReplayObservable -> ReplayStream

ConnectableObservable -> ConnectableStream

ValueConnectableObservable -> ValueConnectableStream

ReplayConnectableObservable -> ReplayConnectableStream
opened by brianegan 16
Release a 1.x

Dart doesn't handle 0.x.x the same as 1.x.x. It makes updating consumers a chore when a lot of them target ^0.14.0 and this releases 0.15.x for a minor update.

Any chance we can get a 1.x to make consumption easier?
under evaluation

opened by johnbland-wf 15

StreamSubscription pause()/resume() with PublishSubject.stream.listen not working

// I have a code 
  Stream<BleScannerState> get state => _statePublishSubject.stream;

// and
 _bleScannerSubscription = _bleScanner.state.listen(...)

// but in code when I try to do
 _bleScannerSubscription?.pause();

// and later  
_bleScannerSubscription?.resume();

// I still get isPaused == true
final isPaused = _bleScannerSubscription?.isPaused;

opened by ObranS 0

StackOverflowError forwarding_stream.dart in _MultiControllerSink.add

This exception was reported to our sentry instance by some of our end user devices. Seems to me like it is caused by some internal rxdart logic, hence why I'm reporting it here. I have not had luck reproducing this - it's hard to do when the stacktrace doesn't even tell you where to look. I'm hoping you can help me out here, I'd like to know what conditions are needed to get this to occur.

We are using rxdart 0.27.7.

Stacktrace (Click to show)

StackOverflowError: Stack Overflow
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  ?, in ...
  ?, in ...
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_controller.dart", in _StreamController._add
  File "stream_controller.dart", in _StreamController.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _BufferingStreamSubscription._add
  File "stream_impl.dart", in _MultiStreamController.addSync
  File "forwarding_stream.dart", in _MultiControllerSink.add
  File "zone.dart", in _RootZone.runUnaryGuarded
  File "stream_impl.dart", in _BufferingStreamSubscription._sendData
  File "stream_impl.dart", in _DelayedData.perform
  File "stream_impl.dart", in _PendingEvents.handleNext
  File "stream_impl.dart", in _PendingEvents.schedule.<fn>
  File "schedule_microtask.dart", in _microtaskLoop
  File "schedule_microtask.dart", in _startMicrotaskLoop

opened by davidpanic 0

Addition of Copyright Notice

Hello! According to the Appendix of Apache License 2.0, if you want to license your software under this License you should "attach the boilerplate notice, with the fields enclosed by brackets "[]" replaced with your own identifying information". This condition is not met now.

Сould you add a COPYRIGHT NOTICE FILE in the appropriate form, including the year of the software development and your name and surname? You could also apply the Apache License to your source-code files by attaching the notice as a comment at the top of each file. Thank you in advance! Apache.org says: Include a copy of the Apache License, typically in a file called LICENSE, in your work, and consider also including a NOTICE file. It is also valuable to tag each of your source-code files in case they become detached from the LICENSE file. To apply the Apache License to your source-code files, one approach is to attach the following notice to as a comment at the top of each file. Replace the copyright templates with your own identifying information:

Copyright [yyyy] [name of copyright owner] Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

opened by lubovtorina 0

BehaviorSubject doesn't provide value via 'first' in test environment

The dartpad example: https://dartpad.dev/?id=02878d4135fdf329059f95fb9f7a72c4

Description

The issue only appears in widget test environment ('testWidgets'). If widget utilize the 'stream.first' within itself or bloc, not really matters, and stream is BehaviorSubject, the Future provided by 'first' never finishes.

Example

The widget in the example below just shows text 'Value = $number' with the number from assigned stream.

In the real environment the last added value to the stream is shown on the screen. But in the test we see 'Value = -1' because timeout fires. Without timeout widget never receives any value.

void main() {
  testWidgets('Proof', (tester) async {
    final BehaviorSubject<int> subject = BehaviorSubject.seeded(99);
    final valueFinder = find.text('Value = 99');
    final noneFinder = find.text('None');
    await tester.pumpWidget(
      MaterialApp(
        home: SomeWidget(stream: subject),
      ),
    );
    print('Looking for None');
    expect(valueFinder, findsNothing);
    expect(noneFinder, findsOneWidget);
    await tester.pumpAndSettle();
    print('Looking for Value = 99');
    expect(valueFinder, findsOneWidget);
    expect(noneFinder, findsNothing);
  });
}

class SomeWidget extends StatelessWidget {
  const SomeWidget({Key? key, required this.stream}) : super(key: key);

  final Stream<int> stream;

  @override
  Widget build(BuildContext context) => FutureBuilder<int>(
        future: stream.first
            .timeout(const Duration(milliseconds: 10))
            .onError<TimeoutException>((e, st) => -1),
        builder: (context, snapshot) => Center(
          child: Text(
            snapshot.hasData ? 'Value = ${snapshot.data}' : 'None',
          ),
        ),
      );
}

Note

According to this issue, if we wrap pumpWidget within runAsync, it actually works. However, it doesn't feel right. It is more like a workaround.

The question is, can this issue be resolved? If not, or if you believe this is not an issue, could you clarify why such behavior is legit.

Thanks in advance.

opened by agavrilko 0

Stream tests lack time manipulation

RxJava has the ability to inject a scheduler to some streams, this way we can manipulate time in tests. If I have an interval of 5 minutes in my code, I can't test it without a very long test, or by changing my interval in the test to some milliseconds, which won't simulate 100% what's happening in the app.

The same way that Flutter tests allows us to manipulate time for testing animations, without having to actually wait for them, rxdart should give us similar controls, as other rx libraries already do.

opened by feinstein 6

Releases(0.27.7)

0.27.7(Nov 16, 2022)
https://pub.dev/packages/rxdart/versions/0.27.7

Fixed

Subject

Only call onAdd and onError if the subject is not closed. This ensures BehaviorSubject and ReplaySubject do not update their values after they have been closed.

Subject.stream now returns a read-only Stream. Previously, Subject.stream was identical to the Subject, so we could add events to it, for example: (subject.stream as Sink<T>).add(event). This behavior is now disallowed, and will throw a TypeError if attempted. Use Subject.sink/Subject itself for adding events.

Change return type of ReplaySubject<T>.stream to ReplayStream<T>.

Internal refactoring of Subject.addStream.

What's Changed

fix(subject): only call onAdd and onError if the subject is not closed by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/698

refactor(subject): Subject.stream now returns a read-only Stream by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/699

refactor(subject): addStream by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/700

chore(publish): prepare for v0.27.7 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/701

Full Changelog: https://github.com/ReactiveX/rxdart/compare/0.27.6...0.27.7
Source code(tar.gz)
Source code(zip)
0.27.6(Nov 11, 2022)
https://pub.dev/packages/rxdart/versions/0.27.6

Rx.using/UsingStream: resourceFactory can now return a Future. This allows for asynchronous resource creation.

Rx.range/RangeStream: ensure RangeStream is only listened to once.

What's Changed

fix(range): make sure RangeStream can only be listened to once by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/690

fix(range): ensure RangeStream is only listened to once by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/694

refactor(Rx.using): resourceFactory can return a Future by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/696

chore(publish): prepare for v0.27.6 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/697

Full Changelog: https://github.com/ReactiveX/rxdart/compare/0.27.5...0.27.6
Source code(tar.gz)
Source code(zip)
0.27.5(Jul 16, 2022)
https://pub.dev/packages/rxdart/versions/0.27.5

Bug fixes

Fix issue #683: Throws runtime type error when using extension methods on a Stream<R> but its type annotation is Stream<T>, R is a subtype of T (covariance issue with StreamTransformer).

Stream<num> s1 = Stream<int>.fromIterable([1, 2, 3]); // throws "type 'SwitchMapStreamTransformer<num, num>' is not a subtype of type 'StreamTransformer<int, num>' of 'streamTransformer'" s1.switchMap((v) => Stream.value(v)); Stream<int?> s2 = Stream<int>.fromIterable([1, 2, 3]); // throws "type 'SwitchMapStreamTransformer<int?, int?>' is not a subtype of type 'StreamTransformer<int, int?>' of 'streamTransformer'" s2.switchMap((v) => Stream.value(v));

Extension methods were previously implemented via stream.transform(streamTransformer), now via streamTransformer.bind(stream) to avoid this issue.

Fix concatEager: activeSubscription should be changed to next subscription.

Code refactoring

Change return type of pairwise to Stream<List<T>>.

What's Changed

Fix ConcatEagerStream bug by @Stitch-Taotao in https://github.com/ReactiveX/rxdart/pull/682

Fix #683 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/684

refactor(pairwise): change return type of pairwise to Stream<List> by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/685

Update CHANGELOG.md by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/686

prepare for v0.27.5 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/688

New Contributors

@Stitch-Taotao made their first contribution in https://github.com/ReactiveX/rxdart/pull/682

Full Changelog: https://github.com/ReactiveX/rxdart/compare/0.27.4...0.27.5
Source code(tar.gz)
Source code(zip)
0.27.4(May 29, 2022)
https://pub.dev/packages/rxdart/versions/0.27.4

Bug fixes

withLatestFrom should iterate over Iterable<Stream>s only once when the stream is listened to.

Fix analyzer warnings when using Dart 2.16.0.

Features

Add mapNotNull/MapNotNullStreamTransformer.

Add whereNotNull/WhereNotNullStreamTransformer.

Documentation

Fix grammar errors in code examples (thanks to @fzyzcjy).

Update RxMarbles URL for RaceStream (thanks to @Péter Ferenc Gyarmati).

What's Changed

test(flatMap): fix flatMap tests by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/658

fix(withLatestFrom): should iterate over Iterable<Stream> once by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/662

fix: analyzer when using Dart 2.16.0 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/663

Fix grammar errors in code examples by @fzyzcjy in https://github.com/ReactiveX/rxdart/pull/664

docs: update rxmarbles URL for RaceStream by @peter-gy in https://github.com/ReactiveX/rxdart/pull/669

Fix ci error pub: command not found by @beeth0ven in https://github.com/ReactiveX/rxdart/pull/673

ci: remove deprecated packages flag by @beeth0ven in https://github.com/ReactiveX/rxdart/pull/674

Create flutter-example.yml by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/676

Update flutter-example.yml by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/677

Add mapNotNull and whereNotNull by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/548

Prepare for v0.27.4 by @hoc081098 in https://github.com/ReactiveX/rxdart/pull/679

New Contributors

@fzyzcjy made their first contribution in https://github.com/ReactiveX/rxdart/pull/664

@peter-gy made their first contribution in https://github.com/ReactiveX/rxdart/pull/669

@beeth0ven made their first contribution in https://github.com/ReactiveX/rxdart/pull/673

Full Changelog: https://github.com/ReactiveX/rxdart/compare/0.27.3...0.27.4
Source code(tar.gz)
Source code(zip)
0.27.3(Nov 21, 2021)
https://pub.dev/packages/rxdart/versions/0.27.3

Bug fixes

flatMap now creates inner Streams lazily.

combineLatest, concat, concatEager, forkJoin, merge, race, zip iterate over Iterable<Stream>s only once when the stream is listened to.

Disallow mixing autoConnect, connect and refCount together, only one of them should be used.

Features

Introduce AbstractConnectableStream, base class for the ConnectableStream implementations.

Improve CompositeSubscription (thanks to @BreX900)

CompositeSubscription's dispose, clear, and remove methods now return a completion future.

Fixed an issue where a stream not present in CompositeSubscription was canceled.

Added the ability not to cancel the stream when it is removed from CompositeSubscription.

CompositeSubscription implements StreamSubscription.

CompositeSubscription.add will throw a StateError instead of a String if this composite was disposed.

Documentation

Fix Connectable examples.

Update Web example to null safety.

Fix Flutter example: SearchResultItem.fromJson type error (thanks to @WenYeh)

Code refactoring

Simplify takeLast implementation.

Migrate from pedantic to lints and flutter_lints.

Refactor BehaviorSubject, ReplaySubject implementations by using "Sentinel object"s instead of ValueWrappers.

New Contributors

@wayne900204 made their first contribution in https://github.com/ReactiveX/rxdart/pull/627

@BreX900 made their first contribution in https://github.com/ReactiveX/rxdart/pull/633

Full Changelog: https://github.com/ReactiveX/rxdart/compare/0.27.2...0.27.3
Source code(tar.gz)
Source code(zip)
0.27.2(Sep 3, 2021)
https://pub.dev/packages/rxdart/versions/0.27.2

Bug fixes

onErrorReturnWith now does not drop the remaining data events after the first error.

Disallow changing handlers of ConnectableStreamSubscription.

Features

Add delayWhen operator.

Add optional parameter maxConcurrent to flatMap.

groupBy

Rename GroupByStream to GroupedStream.

Add optional parameter durationSelector, which used to determine how long each group should exist.

ignoreElements

Remove @deprecated annotation (ignoreElements should not be marked as deprecated).

Change return type to Stream<Never>.

Documentation

Update to PublishSubject's docs (thanks to @AlexanderJohr).

Code refactoring

Refactoring Stream Transformers, using Stream.multi internally.

Source code(tar.gz)
Source code(zip)
0.27.1(May 30, 2021)
https://pub.dev/packages/rxdart/versions/0.27.1

Bugfix: ForkJoinStream throws Null check operator used on a null value when using nullable-type.

Bugfix: delay operator

Pause and resume properly.

Cancel all timers after it has been cancelled.

Source code(tar.gz)
Source code(zip)
0.27.0(May 7, 2021)
https://pub.dev/packages/rxdart/versions/0.27.0

BREAKING: ValueStream

Remove ValueStreamExtensions.

ValueStream.valueWrapper becomes

value.

valueOrNull.

hasValue.

ValueStream.errorAndStackTrace becomes

error.

errorOrNull.

hasError.

stackTrace.

Add skipLast/SkipLastStreamTransformer (thanks @HannibalKcc).

Update scan: change seed to required param.

Add StackTrace param to recoveryFn when using OnErrorResumeStreamTransformer/onErrorResume/onErrorReturnWith.

Internal refactoring ConnectableStream.

Source code(tar.gz)
Source code(zip)
0.26.0(Feb 27, 2021)
Stable, null-safe release.

Add takeLast (thanks @ThomasKliszowski).

Rework for retry/retryWhen:

Removed RetryError.

retry: emits all errors if retry fails.

retryWhen: emits original error, and error from factory if they are not identical.

streamFactory now accepts non-nullable StackTrace argument.

Update ValueStream.requireValue and ValueStream.requireError: throws actual error or a StateError, instead of throwing "Null check operator used on a null value" error.

Source code(tar.gz)
Source code(zip)
0.26.0-nullsafety.1(Jan 8, 2021)
Breaking change: ValueStream

Add valueWrapper to ValueStream.

Change value, hasValue, error and hasError to extension getters.

Fixed some API example documentation (thanks @HannibalKcc).

throttle/throttleTime have been optimised for performance.

Updated Flutter example to work with the latest Flutter stable.

Source code(tar.gz)
Source code(zip)
0.25.0(Dec 6, 2020)
Sync behavior when using publishValueSeeded.

ValueStream, ReplayStream: exposes stackTrace along with the error:

Change ValueStream.error to ValueStream.errorAndStackTrace.

Change ReplayStream.errors to ReplayStream.errorAndStackTraces.

Merge Notification.error and Notification.stackTrace into Notification.errorAndStackTrace.

Bugfix: debounce/debounceTime unnecessarily kept too many elements in queue.

Source code(tar.gz)
Source code(zip)
0.26.0-nullsafety.0(Nov 24, 2020)
Migrate this package to null safety.

Sdk constraints: >=2.12.0-0 <3.0.0 based on beta release guidelines.

Source code(tar.gz)
Source code(zip)
0.25.0-beta3(Oct 16, 2020)
Bugfix: switchMap doesn't close after the last inner Stream closes.

Docs: updated URL for "Single-Subscription vs. Broadcast Streams" doc (thanks Aman Gupta).

Add FromCallableStream/Rx.fromCallable: allows you to create a Stream from a callable function.

Override BehaviorSubject's built-in operators to correct replaying the latest value of BehaviorSubject.

Bugfix: Source StreamSubscription doesn't cancel when cancelling refCount, zip, merge, concat StreamSubscription.

Forward done event of upstream to ConnectableStream.

Source code(tar.gz)
Source code(zip)
0.25.0-beta2(Sep 24, 2020)
Internal refactoring Stream Transformers.

Fixed RetryStream example documentation.

Error thrown from DeferStream factory will now be caught and converted to Stream.error.

doOnError now have strong type signature: Stream<T> doOnError(void Function(Object, StackTrace) onError).

Updated ForkJoinStream:

When any Stream emits an error, listening still continues unless cancelOnError: true on the downstream.

Pause and resume Streams properly.

Added UsingStream.

Updated TimerStream: Pause and resume Timer when pausing and resuming StreamSubscription.

Source code(tar.gz)
Source code(zip)
0.25.0-beta(Aug 2, 2020)
stream transformations on a ValueStream will also return a ValueStream, instead of a standard broadcast Stream

throttle can now be both leading and trailing

better handling of empty Lists when using operators that accept a List as input

error & hasError added to BehaviorSubject

various docs updates

note that this is a beta release, mainly because the behavior of transform has been adjusted (see first bullet) if all goes well, we'll release a proper 0.25.0 release soon

Source code(tar.gz)
Source code(zip)
0.24.1(May 15, 2020)
Fix for BehaviorSubject, no longer emits null when using addStream and expecting an Error as first event (thanks yuvalr1)

min/max have been optimised for performance

Further refactors on our Transformers

Source code(tar.gz)
Source code(zip)
0.24.0(Apr 14, 2020)
Fix throttle no longer outputting the current buffer onDone

Adds endWith and endWithMany

Fix when using pipe and an Error, Subjects would throw an Exception that couldn't be caught using onError

Updates links for docs (thanks @renefloor)

Fix links to correct marbles diagram for debounceTime (thanks @wheater)

Fix flakiness of withLatestFrom test Streams

Update to docs (@wheater)

Fix withLatestFrom not pause/resume/cancelling underlying Streams

Support sync behavior for Subjects

Add addTo extension for StreamSubscription, use it to easily add a subscription to a CompositeSubscription

Fix mergeWith and zipWith will return a broadcast Stream, if the source Stream is also broadcast

Fix concatWith will return a broadcast Stream, if the source Stream is also broadcast (thanks @jarekb123)

Adds pauseAll, resumeAll, ... to CompositeSubscription

Additionally, fixes some issues introduced with 0.24.0-dev.1

Source code(tar.gz)
Source code(zip)
0.24.0-dev.1(Mar 10, 2020)
Breaking: as of this release, we've refactored the way Stream transformers are set up. Previous releases had some incorrect behavior when using certain operators, for example:

startWith (startWithMany, startWithError) would incorrectly replay the starting event(s) when using a broadcast Stream at subscription time.

doOnX was not always producing the expected results:

doOnData did not output correct sequences on streams that were transformed multiple times in sequence.

doOnCancel now acts in the same manner onCancel works on regular subscriptions, i.e. it will now be called when all active subscriptions on a Stream are cancelled.

doOnListen will now call the first time the Stream is subscribed to, and will only call again after all subscribers have cancelled, before a new subscription starts.

To properly fix this up, a new way of transforming Streams was introduced.
Operators as of now use Stream.eventTransformed and we've refactored all operators to implement Sink instead.

Adds takeWileInclusive operator (thanks to @hoc081098)

We encourage everyone to give the dev release(s) a spin and report back if anything breaks. If needed, a guide will be written to help migrate from the old behavior to the new behavior in certain common use cases.

Keep in mind that we tend to stick as close as we can to how normal Dart Streams work!

Source code(tar.gz)
Source code(zip)
0.23.1(Dec 13, 2019)
Fix broken links in README to point to new api docs for extension methods

Source code(tar.gz)
Source code(zip)
0.23.0(Dec 13, 2019)
Extension Methods replace Observable class!

Please upgrade existing code by using the rxdart_codemod package

Remove the Observable class. With extensions, you no longer need to wrap Streams in a [Stream]!

Convert all factories to static constructors to aid in discoverability of Stream classes

Move all factories to an Rx class.

Remove Observable.just, use Stream.value

Remove Observable.error, use Stream.error

Remove all tests that check base Stream methods

Subjects and *Observable classes extend Stream instead of base Observable

Rename *Observable to *Stream to reflect the fact they're just Streams.

ValueObservable -> ValueStream

ReplayObservable -> ReplayStream

ConnectableObservable -> ConnectableStream

ValueConnectableObservable -> ValueConnectableStream

ReplayConnectableObservable -> ReplayConnectableStream

All transformation methods removed from Observable class

Transformation methods are now Extensions of the Stream class

Any Stream can make use of the transformation methods provided by RxDart

Observable class remains in place with factory methods to create different types of Streams

Removed deprecated ofType method, use whereType instead

Deprecated concatMap, use standard Stream asyncExpand.

Removed AsObservableFuture, MinFuture, MaxFuture, and WrappedFuture

This removes asObservable method in chains

Use default asStream method from the base Future class instead.

min and max now implemented directly on the Stream class

Source code(tar.gz)
Source code(zip)
0.23.0-dev.1(Nov 19, 2019)
Feedback on this change appreciated as this is a dev release before 0.23.0 stable!

All transformation methods removed from Observable class

Transformation methods are now Extensions of the Stream class

Any Stream can make use of the transformation methods provided by RxDart

Observable class remains in place with factory methods to create different types of Streams

Removed deprecated ofType method, use whereType instead

Deprecated concatMap, use standard Stream asyncExpand.

Removed AsObservableFuture, MinFuture, MaxFuture, and WrappedFuture

This removes asObservable method in chains

Use default asStream method from the base Future class instead.

min and max now implemented directly on the Stream class

Source code(tar.gz)
Source code(zip)
0.22.6(Nov 6, 2019)
Bugfix: When listening multiple times to aBehaviorSubject that starts with an Error, it emits duplicate events.

Linter: public_member_api_docs is now used, we have added extra documentation where required.

Source code(tar.gz)
Source code(zip)
0.22.5(Oct 31, 2019)
Bugfix: DeferStream created Stream too early

Bugfix: TimerStream created Timer too early

Source code(tar.gz)
Source code(zip)
0.22.4(Oct 17, 2019)
SwitchMap controller is no longer closed improperly in some scenarios

Source code(tar.gz)
Source code(zip)
0.22.3(Oct 4, 2019)

The newly added whereType operator, which replaces onType, was failing in Flutter production builds. The test method for Type matching has been altered.
Source code(tar.gz)
Source code(zip)
0.22.1(Jul 23, 2019)
Fixes following issues:

Erroneous behavior with scan and BehaviorSubject.

Bug where flatMap would cancel inner subscriptions in pause/resume.

Updates to make the current "pedantic" analyzer happy.

Source code(tar.gz)
Source code(zip)
0.22.0(May 3, 2019)
This version includes refactoring for the backpressure operators:

Breaking Change: debounce is now split into debounce and debounceTime.

Breaking Change: sample is now split into sample and sampleTime.

Breaking Change: throttle is now split into throttle and throttleTime.

Source code(tar.gz)
Source code(zip)
0.21.0(Feb 20, 2019)
Breaking Change: BehaviorSubject now has a separate factory constructor seeded() This allows you to seed this Subject with a null value.

Breaking Change: BehaviorSubject will now emit an Error, if the last event was also an Error. Before, when an Error occurred before a listen, the subscriber would not be notified of that Error. To refactor, simply change all occurences of BehaviorSubject(seedValue: value) to BehaviorSubject.seeded(value)

Added the groupBy operator

Bugix: doOnCancel: will now await the cancel result, if it is a Future.

Removed: bufferWithCount, windowWithCount, tween Please use bufferCount and windowCount, tween is removed, because it never was an official Rx spec.

Updated Flutter example to work with the latest Flutter stable.

Source code(tar.gz)
Source code(zip)
0.20.0(Dec 12, 2018)
Breaking Change: bufferCount had buggy behavior when using startBufferEvery (was skip previously) If you were relying on bufferCount with skip greater than 1 before, then you may have noticed erroneous behavior.

Breaking Change: repeat is no longer an operator which simply repeats the last emitted event n-times, instead this is now an Observable factory method which takes a StreamFactory and a count parameter. This will cause each repeat cycle to create a fresh Observable sequence.

mapTo is a new operator, which works just like map, but instead of taking a mapper Function, it takes a single value where each event is mapped to.

Bugfix: switchIfEmpty now correctly calls onDone

combineLatest and zip can now take any amount of Streams:

combineLatest2-9 & zip2-9 functionality unchanged, but now use a new path for construction.

adds combineLatest and zipLatest which allows you to pass through an Iterable<Stream> and a combiner that takes a List when any source emits a change.

adds combineLatestList / zipList which allows you to take in an Iterable<Stream> and emit a Observable<List> with the values. Just a convenience factory if all you want is the list!

Constructors are provided by the Stream implementation directly

Bugfix: Subjects that are transformed will now correctly return a new Observable where isBroadcast is true (was false before)

Remove deprecated operators which were replaced long ago: bufferWithCount, windowWithCount, amb, flatMapLatest

Source code(tar.gz)
Source code(zip)
0.13.1(Jun 30, 2017)
Fix error with FlatMapLatest where it was not properly cancelled in some scenarios

Remove additional async methods on Stream handlers unless they're shown to solve a problem

Source code(tar.gz)
Source code(zip)