Rxdart: Add pipe similar to RxJS

[rullyalves]

Is there any claim to be added to the pipe feature seen in RXJs? if so, I wish I could help.

[lukepighetti]

Typically we string methods together to achieve this functionality

[rullyalves]

The need for this comes from creating operators that are fully decoupled from the default library.

[hoc081098]

Use transform method + a StreamTransformer :blush:

[rullyalves]

Can this probably become the standard in the future? new / previous operators being implemented as StreamTransformers?

[lukepighetti]

I know it's not the way rxjs etc do it, but that gets a downvote from me. rxdart is built on Dart streams so historically it has maintained a similar API to the Dart streams api. There is a package that was recently released that has an API very close to rxjs. Can't remember the name right now. That has new classes from the ground up instead of classes that extend StreamController iirc. If you find it and check out the issues you'll see some discussion about how .pipe doesn't work well with Dart's static type checker.

[brianegan]

Unfortunately, I don't think pipe will fit well with Dart because it relies heavily on JavaScripts dynamic nature. For pipe to work in Dart, you'd need to provide type information to the Pipe method, like so:

myStream.pipe3<int, String, String>(
  firstStreamTransformer,
  secondStreamTransformer,
  thirdStreamTransformer,
)

If you wanted to add / remove / change a transformer in a list, you'd need to update the pipe:

myStream.pipe2<int, int>(
  firstStreamTransformer,
  newSecondStreamTransformer,
)

That seems like a really awkward API, especially when you start having 3 or more transformations.

Therefore, I'd recommend sticking with transform + StreamTransformers as @hoc081098 suggested. And in fact -- if you look at the RxDart source code -- all of our transformation methods are implemented as StreamTransformers! You can even import and use them without needing the Observable class and all Transformers have docs on how they can be used independent of the Observable class:

import 'package:rxdart/transformers.dart';

myStream
  .transform(ScanStreamTransformer(...))
  .transform(FlatMapStreamTransformer(...));

• The README contains a list of all transformation methods and their corresponding StreamTransformers: https://github.com/ReactiveX/rxdart#transformations
• Usage of the transformer in the Observable class: https://github.com/ReactiveX/rxdart/blob/master/lib/src/observables/observable.dart#L2245
• One documentation sample for those transformers: https://pub.dev/documentation/rxdart/latest/rx/ScanStreamTransformer-class.html

Back to the core issue: I really don't think pipe and friends are the right way to go. I think it will be an awkward API to work with and you can already plug in your own StreamTransformers without any problem using transform. Furthermore, the Dart team is working on adding extension methods to the language spec. This would mean we would be able to remove the Observable class entirely and all of these transformation methods could be implemented as extensions directly on the Stream class instead. You will be able to do the same for any transformations you use regularly!

[rullyalves]

@brianegan Thank you very much for the detailed and clear explanation.

[lukepighetti]

That's exactly how we handled pipe in the aforementioned more rxjs'y package, pipe2,pipe3,pipe4 etc and the typings were a mess.

[brianegan]

Thanks for all the feedback and opening the issue. I'll close it out for now, but if ya find a compelling reason to open it back up please feel free to open a new issue!