Any plans to have RxDart be more close to canonical implementation of Rx about rxdart HOT 9 CLOSED

I'm not trying to pick on your work guys but the whole point of using rx is that it's the same on every platform/language.

from rxdart.

"I see. Any way to have it more "inline"/concise like RxJS or RxSwift examples I added to this post? Ideally I'd like to wrap my observable creation in a single method in my, let's say, view model class."

Yep! To create streams, there are a couple ways. First, using StreamControllers (most visually similar to what you were looking to do):

Stream<String> createObservable() {
  final controller = new StreamController<String>();

  controller.add("First event will be emitted!");
  controller.addError(new Exception("Oh Noes! This error will be emitted"));
  controller.close();

  return new Observable(controller.stream);
}

You can then listen to these like so:

createObservable().listen((x) => print("Next: $x"),
        onError: (e, s) => print("Error: $e"),
        onDone: () => print("Completed"));

Alternatively, the exact same thing can be achieved using async generator functions! These might be a bit advanced / look funny at first, and only return raw Streams (would have to be converted to an observable outside if you need da Rx Powers), but are really cool and quite expressive. Any values you yield from the function will be emitted as items in your stream! If you throw any errors, those will be propagated as well to the listener! This lets you write "synchronous looking" code that is in fact async :)

Stream<String> createStreamWithGenerator() async* { // Notice that little async* 
  yield "First event will be emitted!";

  // Run a fake expensive operation for 2 seconds 
  await new Future.delayed(new Duration(seconds: 2)); 

  // This exception will be sent to the `onError` handlers
  throw new Exception("Oh Noes! Then this error will be emitted");

  // Then the stream closes implicitly
}

You can then listen to these like so:

createStreamWithGenerator().listen((x) => print("Next: $x"),
        onError: (dynamic error) => print("Error: $e"),
        onDone: () => print("Completed"));

Or, if you needed to beef that async function up and flatmap it,

new Observable(createStreamWithGenerator()).flatMap((x) => new Observable.just("Flatmapped"))
        .listen((x) => print("Next: $x"),
        onError: (e, s) => print("Error: $e"),
        onDone: () => print("Completed"));

If you wanted to test these streams, you could do so using the Test package + Stream matchers:

test('Creates a stream and throws an error', () async {
    await expect(createObservable(), emitsInOrder(["First event will be emitted!", emitsError(isException), emitsDone]));
  });

"That makes sense. Although I'd love to have more of my fellow RxSwift peeps swayed towards trying Dart through RxDart. And my main argument usually is that it's exactly the same thing you already know...."

Ya, it's something we've thought a lot about, but it's a difficult tradeoff. We're certainly trying to accommodate new users as much as we can, but feel it would be a disservice in many ways to try to Rx-ify everything over the Dart standards. What if one Stream in DartLand cancelled when an error occurred, then you switch to a non-Rx Stream and it doesn't? Or one stream can be listened to multiple times whereas every other stream can't be? Those details could become maddening -- "Wait, am I dealing with Rx where it will cancel or regular Streams here?! Oh, I can't listen twice to a stream I had in the flatMap chain!?"

A bit harder up front for adoption, but will make all our lives much easier when peeps start writing a bit more Dart, and let us be friends with the wider Dart community!

from rxdart.

Hi Alex,

No need to apologize, we appreciate all forms of feedback!

You are right to say rxdart is different, we chose to build upon the existing Dart Stream (which is a part of the Dart SDK) to benefit from the language support out of the box, rather than reinventing it to make a more compatible Rx implementation.

This facilitates the step for a Dart developer (who has a good understanding of Dart streams) to try out rxdart, but does indeed make it more difficult for devs coming from other languages to just jump in and be productive right away.
Have you already worked with Dart streams before? There's some good reads here and there, the official Dart docs provide this starting point: https://www.dartlang.org/tutorials/language/streams

Looking at your code example, I assume this is Swift?

The following Dart code should help you get on your way:

import 'dart:async';

import 'package:rxdart/rxdart.dart';

void main() {
  // reference to the subscription
  var subscription;

  // onNext handler
  void onNext(event) => print(event);

  // completed handler
  void onCompleted() {
    // do something...
    print('all done!');
    // cancel the subscription
    subscription.cancel();
  }

  // subscribing to any of the below Streams...
  subscription = new Observable(_createStreamFromList())
      // some rx operator here...
      .debounce(const Duration(seconds: 1))
      // listen aka subscribe
      .listen(onNext, onDone: onCompleted);
}

// new Stream from an Array
Stream _createStreamFromList() => new Stream.fromIterable([1, 2, 3]);

// nex Stream using a Subject
Stream _createStreamWithController() {
  var controller = new StreamController();

  controller.add(1);
  controller.add(2);
  controller.add(3);

  return controller.stream;
}

// new Stream using a generator
int index = 0;

Stream _createStream() async* {
  yield ++index;
}

If you need any help or have questions, feel free to also visit us in the rxdart Gitter channel!

from rxdart.

this is great! thank you @brianegan!

from rxdart.

For example right now I after 2 hours of research I was unable to create a simple Observable of my own just like I'd do with RxSwift or RxJS. An example of what I wanted to do would be this:

let signal = Observable.create { observer in
        observer.on(.next("some value produced after long computation"))
        observer.on(.completed)
        return Disposables.create()
}

signal.subscribe(onNext: { n in
      print(n)
})

That is something that should very easy to do with Rx but unfortunately, as far as I understand, this is not as straight forward with current implementation of RxDart.

Any tips on how I can achieve the above with RxDart?

from rxdart.

Hey @alexvbush -- thanks for writing! We're trying to be as close to the Observable spec as possible within the confines of Dart Streams, but we will always favor Dart Streams over observables, and I don't see that as a decision we plan on revisiting. If we went away from Dart streams, we'd

• lose language-level syntax support for creating streams
• RxDart would be incompatible with all of the libs out there that rely on Dart Streams, most notably Angular2 Dart

That said, I'd be interested in hearing a bit more from ya!

"in my opinion the best thing about is lost in the process - parity of features/api with other Rx implementations."

Other than lacking a create method (I'll add docs around this! -- update: err, turns out I had already added a link to how to create Dart streams in the Docs you found :( Sounds like we need more concrete examples.) -- do you feel there are more missing pieces? Or parts where you lost time or the implementations were so different that they did something unexpected? I'm trying to really narrow down at which points you experienced pain and how we can help! Create is a great example, and something we can help with by improving docs.

We've bent over backwards to carefully comply with the Observable spec as best we can within the constraints of Dart Streams, so I'd be interested in more examples where you ran into trouble or features that were missing so we can chat about it! Thanks!

from rxdart.

Also, in particular, this article might help with what you're interested in:

https://www.dartlang.org/articles/libraries/creating-streams

As Frank put in there, you can easily create a function that returns a Stream from a StreamController that does exactly what you're hoping to do.

from rxdart.

Hi @frankpepermans @brianegan, thank you for your prompt responses!

No need to apologize, we appreciate all forms of feedback!

thank you! again, I am not trying to attack anyone or be negative in any way. Just trying to figure out what was the intent and if my assumptions and expectations are correct or not.

This facilitates the step for a Dart developer (who has a good understanding of Dart streams) to try out rxdart, but does indeed make it more difficult for devs coming from other languages to just jump in and be productive right away.

Indeed. I'll have a few examples of what is in my experience extensively used with both RxSwift and RxJS for you below.

Have you already worked with Dart streams before?

very briefly a few years ago.

Looking at your code example, I assume this is Swift?
The following Dart code should help you get on your way:

Thank you for this dart code sample! It makes sense. Although it seems like a very excessive way to express something that is very straightforward in RxJS and RxSwift:

RxJS Example:

var source = Rx.Observable.create(function (observer) {
  
    // some expensive computation here

    observer.onError(Error("oops"))
        
    return function () {
        console.log('disposed');
    };
});

var subscription = source.subscribe(
    function (x) {
        console.log('Next: ' + x);
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

RxSwift Example:

let signal: Observable<String> = Observable.create { (subscriber) -> Disposable in    
    print("this is some expensive operation")    

    subscriber.onError(NSError.init(domain: "", code: 1, userInfo: nil))
    
    return Disposables.create {            
        print("this is a cleanup when subscription is disposed")
    }
}

signal.subscribe(onNext: { (value) in
    print("next \(value)")
}, onError: { (error) in
    print("error: \(error)")
}, onCompleted: {
    print("completed callback")
})

Hey @alexvbush -- thanks for writing! We're trying to be as close to the Observable spec as possible within the confines of Dart Streams, but we will always favor Dart Streams over observables, and I don't see that as a decision we plan on revisiting. If we went away from Dart streams, we'd
lose language-level syntax support for creating streams
RxDart would be incompatible with all of the libs out there that rely on Dart Streams, most notably Angular2 Dart

That makes sense. Although I'd love to have more of my fellow RxSwift peeps swayed towards trying Dart through RxDart. And my main argument usually is that it's exactly the same thing you already know....

Other than lacking a create method (I'll add docs around this! -- update: err, turns out I had already added a link to how to create Dart streams in the Docs you found :( Sounds like we need more concrete examples.) -- do you feel there are more missing pieces?

Yes, definitely more examples would help immensely! (would love to contribute if I have time)

Or parts where you lost time or the implementations were so different that they did something unexpected? I'm trying to really narrow down at which points you experienced pain and how we can help! Create is a great example, and something we can help with by improving docs.

Yes. For example have a look at two code samples above in both rxjs and rxswift. In both cases the created observable errors out and my gut expectation is that I'll get an error callback in my subscription and right after that by dispose subscription block will be executed because after erroring out subscriptions should cancel and release their resources. As far as I understand this is not the case for Dart stream based singals. This is unexpected. I think there's a flag I can pass but I need to know about it.

Also another unexpected thing is that after subscribing and executing a signal I can't subscribe to it again, because dart streams again (unless it's a broadcast stream).

We've bent over backwards to carefully comply with the Observable spec as best we can within the constraints of Dart Streams, so I'd be interested in more examples where you ran into trouble or features that were missing so we can chat about it! Thanks!

Appreciate your work!!! It's amazing to see Rx implementation in Dart! It's one of those last missing pieces for me to get completely over to Dart and away from JS.

As Frank put in there, you can easily create a function that returns a Stream from a StreamController that does exactly what you're hoping to do.

I see. Any way to have it more "inline"/concise like RxJS or RxSwift examples I added to this post? Ideally I'd like to wrap my observable creation in a single method in my, let's say, view model class.

from rxdart.

Sure thing @alexvbush! I'll go ahead and create an issue to add a more comprehensive "Adoption Guide" or "Coming from other Rx implementations guide". Please feel free to file more issues if you run into trouble so we can add or examine the functionality / improve docs!

from rxdart.