reactivex/rxdart

The Reactive Extensions for Dart

rxdart's Issues

Improve Subject Test Coverage

Implement distinctUntilChanged

Any plans to implement distinctUntilChanged ? see example on rxjs

Problems compiling with DDC

Overview

rxdart does not seem to compile with the Dart Dev Compiler (DDC). I'm using Dart 1.24.3.

dart2js works

> pub build test

ddc does not

> pub build test --web-compiler=dartdevc

I also tried:

pub serve --web-compiler=dartdevc example/web

and I get runtime errors when playing with the examples.

To try to find problems, I tried updating the analysis_options.yaml file to turn off implicit-casts:

analyzer:
  strong-mode:
    implicit-casts: false
    implicit-dynamic: true

And there appear to be a handful that I am not comfortable fixing.

Question: What's the versioning policy for this repo?

Do you intend on following semantic versioning with this repo?

If so, are you open to the idea of adding something like this to the readme?

https://github.com/workiva/w_transport#versioning-and-stability

Convert min / max to return single value

In order to maintain consistency with other Rx implementations, such as RxJS, we should return a single value from the max / min operators.

• http://reactivex.io/documentation/operators/count.html
• https://github.com/ReactiveX/rxjs/blob/master/src/operator/max.ts

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!

New Name: Ideas & Discussion!

Given the feedback we heard in issue #95, we've decided it's best to go ahead and rename the repo! We should probably also create a Github org of a similar name.

Please submit each name as an individual comment and we can use 👍 to help guide us a bit with votes!

Improve "Coming from other Rx implementations" docs

We've got some docs in the Observable class, but we should create more docs for peeps to get started. Ideas:

• Creating Observables / Streams
• Creating / working with Futures for one-off operations, where streams aren't actually necessary (might be good for folks new to Dart who don't know about the async primitives)
• Working with single-subscription streams
• Documenting cancelOnError patterns

Type safe `combineLatest`?

Hey there, thanks so much for the Rx port to Dart! One question: I was using this lib + combineLatest in Strong Mode, but combineLatest simply takes an Iterable<Stream<dynamic>>, and a Function as the predicate, therefore losing all type information about the values within the predicate.

Now that we have proper method generics type inference in dart 1.21+, would you be open to a PR that makes this more Type Safe?

RxJava does this in the following way:
https://github.com/ReactiveX/RxJava/blob/010256b0e19bdb18f136f80a4dc48c0a53c3da41/src/main/java/rx/Observable.java#L686

Pseudo-proposal:

typedef R CombineTwoPredicate<T1, T2, R>(T1 t1, T2 t2);
typedef R CombineThreePredicate<T1, T2, T3, R>(T1 t1, T2 t2, T3 t3);

Observable.combineLatest(S1 s1, S2 s2, CombineTwoPredicate predicate)
Observable.combineThreeLatest(S1 s1, S2 s2, S3 s3, CombineThreePredicate predicate)

Again, thanks for your effort, and happy to help contribute! Lemme know whatcha think :)

Has a object which can hold onto multiple observable?

Just like CompositeDisposable in RxJava

License doesn't match rxjs

Can the license be changed to a permissive license like Apache or MIT? I was surprised to see the license is GPL, since this is derivative of rxjs, which itself has an Apache license.

The method 'doOnData' isn't defined for the class 'Observable<...>'.

I'm getting this error when I try to run the 'flutter' example. It's only 'doOnData' function that's giving this error. Other library functions are just fine.

RxDart doesn't successfully build with DartDev Compiler

While building our library with DDC, we noticed that rxDart was flagged as a problem area. I cloned this repo and attempted to build using the following command:

pub build --web-compiler=dartdevc --all --verbose --trace

Unfortunately, the build failed yielding no helpful information. It was upon deleting the example directory that the issues were found:

wf14358:rxdart dustinlessard$ pub build --web-compiler=dartdevc --all
Loading source assets... 
Building rxdart... 
Building dartdevc modules... (4.1s)
Error compiling dartdevc module:rxdart|lib/lib__shared_1.js

[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/buffer_with_count.dart, line 20, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/cast.dart, line 40, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/debounce.dart, line 17, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/default_if_empty.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/dematerialize.dart, line 25, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/distinct_unique.dart, line 24, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/do.dart, line 40, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/flat_map.dart, line 18, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/flat_map_latest.dart, line 20, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/ignore_elements.dart, line 13, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/interval.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/materialize.dart, line 17, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/of_type.dart, line 33, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/on_error_resume_next.dart, line 16, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/repeat.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/sample.dart, line 13, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/scan.dart, line 12, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/skip_until.dart, line 13, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/start_with.dart, line 10, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/start_with_many.dart, line 10, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/switch_if_empty.dart, line 27, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/take_until.dart, line 15, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/throttle.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/time_interval.dart, line 12, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/timestamp.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/window_with_count.dart, line 17, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:rxdart/src/transformers/with_latest_from.dart, line 17, col 7)

Please fix all errors before compiling (warnings are okay).

Error compiling dartdevc module:async|lib/lib__shared_0.js

[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:async/src/result/capture_transformer.dart, line 15, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:async/src/result/release_transformer.dart, line 12, col 7)

Please fix all errors before compiling (warnings are okay).

Error compiling dartdevc module:rxdart|lib/lib__futures.js

[error] Invalid override. The type of 'Observable.cast' ('<S>(TypeToken<S>) → Observable<S>') isn't a subtype of 'Stream<T>.cast' ('<R>() → Stream<R>'). (package:rxdart/src/observable.dart, line 1044, col 3)
[error] Invalid override. The type of 'Observable.firstWhere' ('((T) → bool, {defaultValue: () → Object}) → AsObservableFuture<dynamic>') isn't a subtype of 'Stream<T>.firstWhere' ('((T) → bool, {defaultValue: () → dynamic, orElse: () → T}) → Future<T>'). (package:rxdart/src/observable.dart, line 1300, col 3)
[error] Invalid override. The type of 'Observable.lastWhere' ('((T) → bool, {defaultValue: () → Object}) → AsObservableFuture<dynamic>') isn't a subtype of 'Stream<T>.lastWhere' ('((T) → bool, {defaultValue: () → dynamic, orElse: () → T}) → Future<T>'). (package:rxdart/src/observable.dart, line 1420, col 3)
[error] Invalid override. The type of 'Observable.singleWhere' ('((T) → bool) → AsObservableFuture<T>') isn't a subtype of 'Stream<T>.singleWhere' ('((T) → bool, {orElse: () → T}) → Future<T>'). (package:rxdart/src/observable.dart, line 1664, col 3)

Please fix all errors before compiling (warnings are okay).

Error compiling dartdevc module:async|lib/lib__async.js

[error] Missing concrete implementation of getter 'Timer.tick'. (package:async/src/restartable_timer.dart, line 11, col 7)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:async/src/single_subscription_transformer.dart, line 16, col 7)
[error] Invalid override. The type of 'TypeSafeStream.firstWhere' ('((T) → bool, {defaultValue: () → Object}) → Future<dynamic>') isn't a subtype of 'Stream<T>.firstWhere' ('((T) → bool, {defaultValue: () → dynamic, orElse: () → T}) → Future<T>'). (package:async/src/typed/stream.dart, line 56, col 3)
[error] Invalid override. The type of 'TypeSafeStream.lastWhere' ('((T) → bool, {defaultValue: () → Object}) → Future<dynamic>') isn't a subtype of 'Stream<T>.lastWhere' ('((T) → bool, {defaultValue: () → dynamic, orElse: () → T}) → Future<T>'). (package:async/src/typed/stream.dart, line 59, col 3)
[error] Invalid override. The type of 'TypeSafeStream.singleWhere' ('((T) → bool) → Future<T>') isn't a subtype of 'Stream<T>.singleWhere' ('((T) → bool, {orElse: () → T}) → Future<T>'). (package:async/src/typed/stream.dart, line 62, col 3)
[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (package:async/src/typed_stream_transformer.dart, line 23, col 7)
[warning] A function of type '(T) → void' can't be assigned to a location of type '(dynamic) → void'. (package:async/src/delegate/stream_subscription.dart, line 32, col 20)

Please fix all errors before compiling (warnings are okay).

Error compiling dartdevc module:test|lib/lib__test.js

[error] The expression here has a type of 'void', and therefore cannot be used. (package:test/src/runner/load_suite.dart, line 157, col 27)
[error] The expression here has a type of 'void', and therefore cannot be used. (package:test/src/runner/load_suite.dart, line 158, col 14)
[error] The operator '==' isn't defined for the class 'void'. (package:test/src/runner/load_suite.dart, line 158, col 23)
[error] The expression here has a type of 'void', and therefore cannot be used. (package:test/src/runner/load_suite.dart, line 158, col 49)
[error] The argument type 'List' can't be assigned to the parameter type 'FutureOr<List<T>>'. (package:test/src/util/stream_queue.dart, line 514, col 25)
[error] The argument type 'List' can't be assigned to the parameter type 'FutureOr<List<T>>'. (package:test/src/util/stream_queue.dart, line 519, col 25)

Please fix all errors before compiling (warnings are okay).

Error compiling dartdevc module:rxdart|test/test__transformers__transform_test.js

[error] Missing concrete implementations of 'StreamTransformer.cast' and 'StreamTransformer.retype'. (/private/var/folders/6v/y8vzm1ns6snb3cpz2nyb8wt00000gm/T/pub_yW63nj/test/transformers/transform_test.dart, line 18, col 7)

Please fix all errors before compiling (warnings are okay).

Build failed.

Any plans to have RxDart be more close to canonical implementation of Rx

I understand that some tradeoffs had to be made to implement Rx in Dart (as outlined in this doc https://www.dartdocs.org/documentation/rxdart/0.10.2/rx/Observable-class.html) but in my opinion the best thing about is lost in the process - parity of features/api with other Rx implementations. It makes my RxSwift or RxJS skills for example less applicable and transferable to RxDart and vice versa.
Are there any plans to make APIs closer to canonical Rx such as RxJS?

The Old JS Libraries

Hi Frank...

I checking you native implementation.... I nice and certainly cleaner in some ways without all the JS wrapping.

I just wanted to drop you a line to try to understand, if you had any issues to with 0.6.0 interop.

I am middle to move some JS stuff to Dart and mostly not a problem. Except promises. I know +jacob richman is working to harmonise Futures/Await and promises.

My real question is did you come across something that made you want to dump the wrapping.... or was it case of native Async Observable was cleaner approach....

Thanks in advance, hope the question is not too fluffy..

Create more examples

More examples are needed for this API.
I think one example this library really needs is something that explains the transition from Dart Streams (since this API isn't documented that well) to RxDart Obserbables.

Retry operator did not re-trigger onListen/onSubscribe side effect

As described in ReactiveX retry operator, current observable should be resubscribed once an error occurs.

But existing implementation does not do that.

Code coverage

Coverage is quite good, but the reporting is broken. We need to find an alternative for dart_coveralls, which seems broken as of sdk 1.16

compiling for flutter dart 2 fails

on master branch of Flutter...
but had to change version in pubspec to dev.22.0
env:

x-MacBook-Pro:rxdart apple$ flutter doctor -v
[✓] Flutter (on Mac OS X 10.13.3 17D47, locale en-DE, channel master)
    • Flutter version 0.1.1-pre.34 at /Users/apple/flutter
    • Framework revision d745684393 (4 hours ago), 2018-02-13 14:16:26 +0100
    • Engine revision f5a4a93787
    • Tools Dart version 2.0.0-dev.22.0
    • Engine Dart version 2.0.0-edge.32396138ff733cbe8d9ac32178e6cffccdf80644

[✓] Android toolchain - develop for Android devices (Android SDK 27.0.3)
    • Android SDK at /Users/apple/Library/Android/sdk
    • Android NDK at /Users/apple/Library/Android/sdk/ndk-bundle
    • Platform android-27, build-tools 27.0.3
    • ANDROID_HOME = /Users/apple/Library/Android/sdk
    • Java binary at: /Applications/Android Studio.app/Contents/jre/jdk/Contents/Home/bin/java
    • Java version OpenJDK Runtime Environment (build 1.8.0_152-release-915-b08)

[✓] iOS toolchain - develop for iOS devices (Xcode 9.2)
    • Xcode at /Applications/Xcode.app/Contents/Developer
    • Xcode 9.2, Build version 9C40b
    • ios-deploy 1.9.2
    • CocoaPods version 1.4.0

[✓] Android Studio (version 3.0)
    • Android Studio at /Applications/Android Studio.app/Contents
    • Java version OpenJDK Runtime Environment (build 1.8.0_152-release-915-b08)

[✓] VS Code (version 1.20.0)
    • VS Code at /Applications/Visual Studio Code.app/Contents
    • Dart Code extension version 2.8.1

[✓] Connected devices
    • iPhone 8 Plus • 10FA54AF-3CDB-414B-92B9-9F0D15DA861D • ios • iOS 11.2 (simulator)

make file commands:

cd rxdart/example/flutter/github_search && flutter packages get
cd rxdart/example/flutter/github_search && flutter run

output:

cd rxdart/example/flutter/github_search && flutter packages get
Running "flutter packages get" in github_search...    2.1s
cd rxdart/example/flutter/github_search && flutter run
Launching lib/main.dart on iPhone 8 Plus in debug mode...
Upgrading project.pbxproj of Runner.app' to include the 'flutter_assets' directory
Automatic upgrade of project.pbxproj failed.
 To manually upgrade, open ios/Runner.xcodeproj/project.pbxproj:
 Add the following line in the "PBXBuildFile" section
                2D5378261FAA1A9400D5DBA9 /* flutter_assets in Resources */ = {isa = PBXBuildFile; fileRef = 2D5378251FAA1A9400D5DBA9 /* flutter_assets */; };
 Add the following line in the "PBXFileReference" section
                2D5378251FAA1A9400D5DBA9 /* flutter_assets */ = {isa = PBXFileReference; lastKnownFileType = folder; name = flutter_assets; path = Flutter/flutter_assets; sourceTree = SOURCE_ROOT; };
 Add the following line in the "children" list of the "Flutter" group in the "PBXGroup" section
                                2D5378251FAA1A9400D5DBA9 /* flutter_assets */,
 Add the following line in the "files" list of "Resources" in the "PBXResourcesBuildPhase" section
                                2D5378261FAA1A9400D5DBA9 /* flutter_assets in Resources */,
Could not build the application for the simulator.
Error launching application on iPhone 8 Plus.
make: *** [run-flu] Error 2

Merge operator has potential bug

See following code

  test('rx.Observable.merge does not deal properly', () async {
     // This will fail:
     // const list<num> expectedOutput = const <num> [1, 2, 3, 4, 5, 6];
     const list<num> expectedOutput = const <num> [1, 4, 2, 5, 3, 6];
 
     StreamController<num> sc1 = new StreamController<num>();
     StreamController<num> sc2 = new StreamController<num>();
 
     int count = 0;
     Stream<num> observable = new Observable<num>.merge([sc1.stream, sc2.stream]);
 
     observable.listen(expectAsync1((num result) {
       // test to see if the combined output matches
       expect(result, expectedOutput[count++]);
     }, count: expectedOutput.length));
 
     sc1.add(1);
     sc1.add(2);
     sc1.add(3);
     sc2.add(4);
     sc2.add(5);
     sc2.add(6);
  });

Or
lestathc@c211c2a

If I changed the stream controller to StreamController(sync: true), then the output will be [1, 2, 3, 4, 5, 6].

I think any order sensitive operators may have this kind of potential bug. The reason is that multiple async StreamControllers has its own execution strategy for multi add operations in one event loop, but which is not FIFO.

Rework buffer and window

Currently we have bufferCount/windowCount and bufferTime/windowTime, but we have no way to buffer on arbitrary intervals (for example, buffer on another Stream, or on a Future, ...)

Proposal:
We create a generic buffer/window operator which takes a StreamSampler implementation.
This sampler is a simple interface which handles how the buffer builds and when the buffer should be emitted.

We provide a set of samplers that are commonly used:

• for count, [skip]: OnCount
• for Duration: OnTime
• for sampling on another Stream: OnStream
• for a recurring Future: OnFuture
• for a condition: OnTest

The user is free to create his/her own implementation OnX.

This will provide the following API:

observable
    .buffer(onTime(const Duration(seconds: 1)))
   .listen(...);

observable
    .buffer(onFuture(() => window.animationFrame()))
   .listen(...);

// stream emits: 1, 5, 20, 80, 120, 150, 200
observable
    .buffer(onTest((int event) => event > 100))
   .listen(...); // prints [1, 5, 20, 80, 120], [150, 200]

Which is equivalent to the following shorthand:

observable
    .bufferTime(const Duration(seconds: 1))
   .listen(...);

observable
    .bufferFuture(() => window.animationFrame())
   .listen(...);

// stream emits: 1, 5, 20, 80, 120, 150, 200
observable
    .bufferTest((int event) => event > 100)
   .listen(...); // prints [1, 5, 20, 80, 120], [150, 200]

Does rxdart Observable want to follow the standard ReaxtiveX onError contract?

Hi,

I am very interesting in this lib, this is a big improvement for Dart.

But I have a concern:

1. as described in the contract of observable, the onError will terminate the subscription.

2. but as described in the Stream.listen, the cancelOnError is false (by default), subscription will still exist.

How does this lib want to deal this case?

Thanks

Document Stream Spec vs Observable Spec quirks

This includes:

• Dart Sync vs Async Streams vs / in combination with Rx Schedulers as we introduce that concept
• Return type of first, last, etc returns Future not an Observable
• Error contract in Observable (closes stream) vs Dart (does not by default)

Synchronously get current value of BehaviorSubject

With RxJS I can use .value to synchronously get the current value of the BehaviorSubject. How can I do this with rxdart?

Question : How does one see the test coverage for this code?

Add 'if' operator

Creating complex if / else structures is quite hard when your method chain is growing.
RxJS has a nice solution to it:
https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/if.md

It takes a 'condition' method, a 'thenSource' and an 'elseSource' as parameters.

DematerializeStreamTransformer<#lib2::Observable::T>' can't be assigned to

[+8917 ms] compiler message: file:///Users/zoechi/.pub-cache/hosted/pub.dartlang.org/rxdart-0.16.0+2/lib/src/observable.dart:1162:26: Error: A value of type '#lib1::DematerializeStreamTransformer<#lib2::Observable::T>' can't be assigned to a variable of type 'dart.async::StreamTransformer<#lib2::Observable::T, #lib2::Observable::dematerialize::S>'.
[        ] compiler message: Try changing the type of the left hand side, or casting the right hand side to 'dart.async::StreamTransformer<#lib2::Observable::T, #lib2::Observable::dematerialize::S>'.
[        ] compiler message:     return transform(new DematerializeStreamTransformer<T>());

Flutter 0.2.2-pre.4 • channel dev • https://github.com/flutter/flutter.git
Framework • revision ce75b44760 (3 hours ago) • 2018-03-20 07:48:43 +0100
Engine • revision e61bb9ac3a
Tools • Dart 2.0.0-dev.35.flutter-290c576264

with --preview-dart-2

rxdart 0.16.0+2

Scan/Min/Max transformer may has side effect leaking

Considering I have such code:

var transformer = new ScanStreamTransformer(myPredictFunc);
var scan1 = myStream.transform(transformer);
var scan2 = myAnotherStream.transform(transformer);

I may dirty the intermedia result of scan transformer.

Given the transforms are public, if I only want to reuse the transforms without using the Observable interface, and according to "It is good practice to write transformers that can be used multiple times.". I may want to use transformer like that.

How about either mark transformers private, or document them/add annotations to indicate that these transformers is not sharable.

startWith that takes a single value?

Another proposal, happy to help! I find when using this library + startWith, I'm always passing in a single value rather than a list. Therefore, I always have to wrap this initial value in brackets. Not a biggie, but seems like this method might cater for the 20% rather than the 80%.

Would you be open to having two methods for startWith (since Dart doesn't support overloading)?

What I'm thinking:

startWith(A value)
startWithIterable(Iterable<A> values)

Thanks again for the lib!

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?

Dartdocs doesn't seem to work

The link to Dartdocs doesn't seem to work:
https://www.dartdocs.org/documentation/rxdart

Update tests to better cover streams / transformers / observable classes independently

We have pretty good test coverage, but it's all pretty intertwined -- IE streams and transformers are tested using the Observable wrapper.

Might be good to slightly refactor our tests to:

• Directly test our Stream constructor classes
• Directly test our transformers
• Very lightly test our Observable glue code.

Improve documentation on Observable

The documentation on ReplaySubject, BehaviourSubject and Observable could use a bit of improvement.
Although I think lots of users coming from RxJava or RxJS will understand what it does, documentation should never assume people have used an Rx* before.

ctr/op wishlist

This issue acts as a priority list for implementing new operators and constructors,
see the full list here: http://reactivex.io/documentation/operators.html

To vote, simply leave a comment which lists the op/ctr names and add a hyperlink to the rx implementation.

For example:

• ignoreElements - http://reactivex.io/documentation/operators/ignoreelements.html

BehaviorSubject declared with .startWith() returns the initial value instead of the latest value.

A BehaviorSubject declared with .startWith() will return the initial value instead of the latest value.
But using the seedValueproperty it seems to work correctly.
I would expect .startWith() to act similar as when using the seedValue property.
Let's have a look at following examples:

Example 1

• sets seedValue: 0

rx.BehaviorSubject<int> countController = new rx.BehaviorSubject<int>(seedValue: 0);

new rx.Observable(document.onKeyUp)
    .doOnData((_) => print('KEY UP'))
    .flatMapLatest((_) => countController.stream.take(1))
    .map((int curr) => curr + 1)
    .listen((int curr) => countController.add(curr));

new rx.Observable(countController.stream)
   .listen(print);

This results in:

0 // (—> This one triggers when listening to the BehaviorSubject of course)
KEY UP
1
KEY UP
2
KEY UP
3

Example 2

• has no seedValue
• has a ‘count’ Observable wich sets .startWith(0)

rx.BehaviorSubject<int> countController = new rx.BehaviorSubject<int>();
rx.Observable count = countController.stream.startWith(0);

new rx.Observable(document.onKeyUp)
    .doOnData((_) => print('KEY UP'))
    .flatMapLatest((_) => count.take(1))
    .map((int curr) => curr + 1)
    .listen((int curr) => countController.add(curr));

new rx.Observable(countController.stream)
    .listen(print);

This results in:

KEY UP
1
KEY UP
1
KEY UP
1

Just as in example 1, I would expect example 2 to just increment the counter on each keystroke.

sink getter on BehaviorSubject and ReplaySubject has different behavior

Adding events to the subject's sink will properly add events to the Stream, but the event is not picked up for the respective use cases of the Subjects.

For example:

var subject = new BehaviorSubject<int>();

subject.add(1);

expect(subject.value, 1); /// works

var subject = new BehaviorSubject<int>();

subject.sink.add(1);

expect(subject.value, 1); /// fails!

expected:
Using sink does not break the Subject like shown above

repo description still describes rxdart as a wrapper

easy fix, small nitpick :)

Forward exceptions to StreamSubscription

Problem

All Exceptions that happen within an operator should be forwarded to the StreamSubscription. Currently, Stream-based operators do this, such as map, but our custom Observable operators do not, such as flatMap. Our operators will throw an exception at the CallSite, but will not forward the exception to the Subscription.

Solution

Any operator that takes a function that could throw an exception needs to be wrapped in a try / catch block. If the function runs fine, we keep going. If we encounter an error, we forward that error to the Subscription via the operator's StreamController.

Contributing.md

We should write one.

Looking for a way to have a stream with debounce cancelled when the base stream is closed

    _tickerSubscription = new Observable(onStateChange)
        .doOnDone(() {
          print('done: ${new DateTime.now().second}');
          _tickerSubscription.cancel();
        })
        .debounce(const Duration(seconds: 5))
        .listen((_) {
      print('event: ${new DateTime.now().second}');
        });

prints

done: 56
00:00 �[32m+29�[0m: All tests passed!�[0m
event: 1

This is only an issue when I run it with

#!/usr/bin/env bash
dart --pause-isolates-on-exit --enable-vm-service=8899 test/all.dart &
sleep 1s
pub global run coverage:collect_coverage --uri=http://localhost:8899 -o coverage/coverage.json --resume-isolates
pub global run coverage:format_coverage --packages=.packages -i coverage/coverage.json -l -o coverage/lcov.info

With pub run test when "All tests passed!" is printed, there is no dart process hanging around anymore.

I use this in a test with built_redux where I call store.dispose() in tearDown()

Give us Subject

StreamController is a good replacement for javas PublishSubject. But there is no replacement for the BehaviorSubject in rxdart.

Or I'm to stupid to get the latest emitted item of a StreamController

Update GitHub description and web site - point to pub site

Makes it easy to see what version is published, etc

rxdart 0.5.0 appears to be missing from pub?

A few days ago, I was able to pull this version from pub but now I cannot. 0.4.8 is the latest I can successfully get.

Verify test coverage of closing Streams

After my PR this morning, I realized we don't have many tests cases around closing streams / transformers. Probs good to add to our test coverage.

Improve Subject Documentation

Class 'BehaviorSubject<int>' has no instance method 'listen'

On you documentation you have this:

/// ### Example
///
///     final subject = new BehaviorSubject<int>();
///
///     subject.add(1);
///     subject.add(2);
///     subject.add(3);
///
///     subject.listen(print); // prints 3
///     subject.listen(print); // prints 3
///     subject.listen(print); // prints 3

But when I try to run this same example I get this error: Class 'BehaviorSubject<int>' has no instance method 'listen'

Should we call this RxDart or something else?

Overall, we want to make it easy for peeps coming from other languages to dive right into Dart, with all of the Rx power they're used to in other languages. We want to fit into the Dart ecosystem, and so we've based RxDart's Observable on Dart's Stream class. However, it's important to fundamentally understand Streams before diving into Observables, and there are some fundamental differences between Streams in Dart and Observables from the ReactiveX Contract. When those differences arise, we've decided the best thing to do is favor fitting into the Dart ecosystem over following the ReactiveX spec.

Despite the differences between Dart Streams and Rx Observables, they're conceptually very similar, and we hoped calling it RxDart would make it easy for folks to get going quickly.

What we've seen, however, is a decent amount of confusion, and so we wanted to open a ticket and leave it open for a bit:

Should this library actually be a ReactiveX / RxDart library, or should we rename it to reinforce the fact that we are extending Stream, and that Streams are the most important thing to learn and understand? Has the name been helpful or confusing to you? Thanks for your feedback!

Override Subject exception

I am using RxDart to add functionality over async:dart in my Flutter application.

I am seeing this exception:

I/FlutterActivityDelegate( 5299): onResume setting current activity to this
E/flutter ( 5299): [ERROR:topaz/lib/tonic/logging/dart_error.cc(16)] Unhandled exception:
E/flutter ( 5299): 'dart:async/stream.dart': error: line 942: signature type '(TypeToken) => Observable' of function 'cast' is not a subtype of signature type '() => Stream' of function 'cast'
E/flutter ( 5299): 
E/flutter ( 5299):   Stream < R > cast < R >() {
E/flutter ( 5299):   ^
E/flutter ( 5299): 'package:rxdart/src/observable.dart': error: line 1156: class 'Observable' overrides method 'cast' of super class or interface 'Stream' with incompatible parameters
E/flutter ( 5299):   Observable < S > cast < S >(TypeToken < S > typeToken) =>
E/flutter ( 5299):   ^
E/flutter ( 5299): 
E/flutter ( 5299): #0      new MobileAuthentication (package:lib.frontend.apps.mobile.authentication/src/authentication/authenticator/mobile_authentication.dart:14)
E/flutter ( 5299): #1      MobileContext.setupProduction (file:///Users/marc/BasicWindow/source/monorepo/dart/projects/nightpal/mobile/lib/src/app/context.dart:9)
E/flutter ( 5299): #2      ContextBase.setup (package:nightpal.shared/src/app/context.dart:33)
E/flutter ( 5299): #3      main (file:///Users/marc/BasicWindow/source/monorepo/dart/projects/nightpal/mobile/lib/main.dart:7)
E/flutter ( 5299): #4      _startIsolate.<anonymous closure> (dart:isolate-patch/dart:isolate/isolate_patch.dart:279)
E/flutter ( 5299): #5      _RawReceivePortImpl._handleMessage (dart:isolate-patch/dart:isolate/isolate_patch.dart:165)
I/FlutterActivityDelegate( 5299): onResume setting current activity to this

I am creating a subject like this in a class:

class _State extends AuthenticationStateBase<SignInPage> implements ISignInView {

  final PublishSubject _onSignIn = new PublishSubject();
  @override Observable<SignInEvent> get onSignIn => _onSignIn.observable;

  _State(IController<IView> controller) : super(controller);
  //...
}

Line 14 in the error is where I create the PublishSubject.

In my pubspec.yaml I have the dependency rxdart: ^0.15.0

Migrate to ReactiveX?

Hey, as this project becomes more and more complete (thanks so much for your hard work!), I was wondering if it made sense to potentially move it under the ReactiveX umbrella (https://github.com/ReactiveX)?

That way it could have more "official" weight as a project, and wouldn't feel like a project so tied to a single maintainer?

Just some food for thought, thanks again!

Exception in Flutter 0.1.2 Observable' of function 'cast' is not a subtype of ...

'(TypeToken) => Observable' of function 'cast' is not a subtype of signature type '() => Stream' of function 'cast'
[        ] I/flutter ( 5115): 
[        ] I/flutter ( 5115):   Stream < R > cast < R >() {
[        ] I/flutter ( 5115):   ^
[        ] I/flutter ( 5115): 'package:rxdart/src/observable.dart': error: line 1044: class 'Observable' overrides method 'cast' of super class or interface 'Stream' with incompatible parameters
[        ] I/flutter ( 5115):   Observable < S > cast < S >(TypeToken < S > typeToken) =>
[        ] I/flutter ( 5115):   ^
[        ] I/flutter ( 5115): #0      initAndRunApp (package:myproject/app.dart:33)
[        ] I/flutter ( 5115): <asynchronous suspension>
[        ] I/flutter ( 5115): #1      main.<anonymous closure> 

[✓] Flutter (on Mac OS X 10.13.3 17D47, locale en-AT, channel dev)
    • Flutter version 0.1.2 at /Users/zoechi/flutter/flutter
    • Framework revision 8a65872ef9 (29 hours ago), 2018-02-13 23:32:28 -0800
    • Engine revision 05c5f817eb
    • Tools Dart version 2.0.0-dev.22.0
    • Engine Dart version 2.0.0-edge.3c4dccbd46f152be9e1b6ca95c57357e8e48057c