Is it worth to hardcode done handling inside library?

[ivan-kleshnin]

I'm very curious about the subj. For me, the history of FRP is a sequence of cutting down assumptions. RxJS handles both ERROR and DONE situations and the result speaks for itself. Tens of kilobytes of code. Tons of quirky situations. Too many assumptions.

Just one example:

import {Observable} from "rx";

let xs = Observable.from([1, 2, 3, 4, 5]);
let ys = Observable.interval(100);

ys
  .withLatestFrom(xs, (ys, xs) => [xs, ys])
  .subscribe(console.log);

In case you don't know, withLatestFrom is a sampling operator.

So what it should output? Hard to predict because RxJS complects every possible case (sync vs async, push vs pull, cold vs hold...) and makes assumptions about everything. The actual answer is [5-0]--[5-1]--[5-2]--... which is... strange.

Another related example. @rpominov points out (and people tend to agree) that removal of sync Stream creation operators in Kefir was a right step to simplier API and reasoning.

Flyd library shifts error handling to client proposing to either try/catch error cases or wrap errors with Maybe. I think it's beneficial because Error handling is not a Channel prerogative. JS-CSP does the same trick.

Elm goes further and bans finite signals. Signal just can't end, end of sentence. If you need to pass end condition (end of file for example) to consumer – you just pass corresponding contract value between consumers.

In JS, channels can't be static because there is no compile time. They are values which can be created and garbage collected. But the question still applies as the following.

Do we really need to hardcode DONE handling in library?
Is it worth the complexity and assumptions?

The problem has already started to reveal itself here through channel.close vs channel.close(true) IMO. I'll be honest, I still don't have strong opinion on this, only suspicions. But let me speculate that we should at least consider DONE handling removal, unless it allows us to do something we can't (painlessly) do in other way.

I'm very interested in such examples. So far, I'm not aware of a single one.

Additional information

Evan Czaplicki presents different notions of FRP and compares their benefits and drawbacks.
https://www.youtube.com/watch?v=Agu6jipKfYw
He mentions a number of situations where every possible behavior of Higher Order channels is quirky / unsuspected for different reasons.

And we discussed a bit of it here:
js-csp/js-csp#40

Done handling is a question tightly coupled with permission or forbiddance of Higher Order channels (channels of channels). Presence of higher order channels require that channels end, otherwise we get memory leaks. For a record: I've found no examples of HO channels usage in CSP implementations so far.

[staltz]

RxJS Observables are not signals, neither event streams. They are "Duals to Iterables", or "push-style generators", and as such must have an error handling mechanism (just like functions use try/catch for handling errors and bubbling them up). Read more here: http://stackoverflow.com/questions/25338930/reactive-programming-rxjs-vs-eventemitter-in-node-js/25340716#25340716

[ivan-kleshnin]

@staltz I'm afraid I disagree.

Emitters are push-style generators as well.
So it's fair to call RxJS Observables "event emitters". (you
The just have composable API. It's a very productive but not a structural difference.
I know you're angry of people who blame RxJS without even trying it but I'm not one of them.

Btw, here is your own quote which I'm agreed with

In a way, this isn't anything new. Event buses or your typical click events are really an asynchronous event stream, on which you can observe and do some side effects. Reactive is that idea on steroids.

---

Since Observables don't use generators under the carpet (because of historical precedence), they surely need to implement catch() to catch async errors. But what about DONE? Could we get rid of it? Can you provide an example where we would end up with a lot of app code in that case (and example would not include Higher Order streams)?

With generators we can catch async errors as well as sync so the same question applies to both ERROR and DONE handling.

[danyx23]

Interesting question. I have never worked with CSPs directly, but my gut feeling would be that they are so low level that including semantics for ending would be already a bit too opinionated.

As you said, in Elm, since Signals can never end, a user has to take care of the semantics of "ending" a stream if required - and I think this is not a bad choice.

There are operations that I could think of where having an ended state encoded in the library might be useful, e.g. if you have a batch operator that always batches 10 messages together - if the library supports closing the channel it's a no-brainer that at this point you emmit the remaining messages. But when you don't have an ended state?

I think it's definitely worth thinking about this and maybe implementing a branch that gets rid of the ended state and see what problems come up. Maybe edge cases like the one I mentioned are reason enough to keep the ended state. But maybe the API surface can get significantly smaller without any major downsides if this is done.

[staltz]

@ivan-kleshnin

Emitters are push-style generators as well.
So it's fair to call RxJS Observables "event emitters".

You really did not read the stackoverflow thread.

But what about DONE?

Without done, we can't accomplish .concat(), .last(), etc which are very real-world useful tools.

[ivan-kleshnin]

You really did not read the stackoverflow thread.

What makes you think so? Streams are event emitters. There is a common agreement about it. Node Streams are instances of EventEmitter proof. RxJS authors just chose alternative API. It's not productive to deny it.

Without done, we can't accomplish .concat(), .last(), etc which are very real-world useful tools.

See, they are tools to work with finite streams. So there is some logical loop behind it. Do we need DONE because we need last() or do we need last() because we already have DONE case and therefore we need to handle it somehow?

So the question is not "How to work with finite streams without corresponding operators?"
Obviously it's impossible.

The questions is "Which real world cases is trivial to model with finite streams and hard or impossible without them?"

We've come up with a single example of producing "waves of values" in one of the above links.
It seems impossible to do next without higher-order streams:

let s$ = Observable.interval(1000).flatMap(
  Observable.for([0, 1, 2, 3, 4], (v) => Observable.of(v).delay(v * 50))
); // 0-1-2-3-4---0-1-2-3-4---

But bidirectional nature of CSP channels makes a difference. This version of throttle is much more powerful than one in RxJS. Consumer balances the producer.

let ch0 = interval(x => x, 100); // 0-1-2-3-4-5-6-7-8-9-10-...
let ch1 = map(x => x % 6, ch0);  // 0-1-2-3-4-5-0-1-2-3-4-...
let ch2 = throttle(x => {        
  return x == 5 ? 500 : 0;
}, ch1); // 0-1-2-3-4-5---0-1-2-3-4-5---

[rpominov]

In FRP built-in completion is needed for some operations like reduce or flatMapFirst, without completion that operations can be implemented only in user land. I'm exploring this in basic-streams, here is how reduce can be implemented in user land https://jsfiddle.net/sp6mj2ng/

Channels are different though, maybe completion not so important for them indeed. But I don't have much experience with channels, and can't really say...

[staltz]

@ivan-kleshnin

What makes you think so? Streams are event emitters. There is a common agreement about it. Node Streams are instances of EventEmitter proof. RxJS authors just chose alternative API. It's not productive to deny it.

You still didn't read it. Observables are not streams.

[ivan-kleshnin]

You still didn't read it. Observables are not streams.

Lol, ok if you insist. Though they actually are.

So, in practice, if you look at the concepts, and if you use the option { objectMode: true }, you can match Observable with the Readable stream and Observer with the Writable stream. You can even create some simple adapters between the two models.

And both
http://reactivex.io/intro.html

and
https://github.com/Reactive-Extensions/RxJS

use this term.

[dvlsg]

Seems like this might be starting to get off topic. To get this back on topic, I can say that, personally, I have found many uses for calling #close() on a channel, then awaiting #done() to signify that all values have been handled from a flowing channel / pipe. I would be the first to admit this functionality may not be typical to other CSP libraries, though. I have definitely taken influence / direction from standard nodejs streams.

Both the file-streams and char-counter examples in the repo make use of #done(). For the file-streams example, it is simply to determine the runtime from start to finish. For the char-counter example, it is to reduce all values to an object through the use of #consume(), and then sort the result's Object.entries(). Without #done(), both of these operations would be quite difficult to handle inside user-land, in my opinion. I would be happy to be proven wrong, but I can't see it at the moment.

That being said, you definitely shouldn't need to treat channels as finite. You should be able to keep them open forever, and nothing should force you to use #done(), or #close() -- I intend to keep those optional. Is this not the case for something you're working with? Or is this just a concern for code complexity / library maintainability?

[ivan-kleshnin]

@dvlsg thanks! Very informative and straight to the point.

I'm trying to figure out how CSP concept should fit JavaScript (both Go and Clojure have different design, async primitives, goals, etc...). I was sure JS deserves a better API than direct port (JS-CSP) so since I discovered this lib I'm very excited.

I wanted to get a feedback about the subj. from a person with real-world experience. You answer totally satisfies it and gives me a direction to look for. So now I'm going to recheck your examples with appopriate attention to details.