Rx-style

[acthp]

Is an Rx style really possible with transducers.js? Are there any examples?

I've seen Rich's talk, where he tosses off a few lines about transducers being able to replace Rx, and the docs for this lib and js-csp both specifically mention their use as an Rx work-alike. But at core you need to be able to do a flatmap (monadic bind) of Observables, which is problematic in the synchronous reducer protocol.

Specifically, the 'step' method in the protocol requires a synchronous return. In the transducers.js implementation of Cat, it can only concatenate synchronous sub-streams, and I suspect it can't be extended to async sub-streams without a different protocol. You might be able to work around it in a server-side language like clojure, by blocking on the async source. In the client, you can't block.

Is there some other solution? To be clear, being able to map or filter isn't interesting. You have to be able to flatmap, returning an inner async source.

[ubolonton]

Being able to replace Rx doesn't mean following the "Rx style". It's usually better to approach a specific problem directly with tools provided by CSP + transducers, rather than first reimplementing observable abstractions, which are somewhat higher level but less composable.

Transducers only concern with transformations. The "flatten" part in flatMap would be for a CSP lib to handle. Using mix, it can be something like this:

var mix = mix(outputChannel);

go(function*() {
  while(true) {
    var newChannel = yield take(channelSource);
    if (newChannel === CLOSED) {
      break;
    }
    mix.add(newChannel);
  }
});

[acthp]

I'll take that as a "no". Re: rx style, I literally cut & pasted that term from the docs. Perhaps the docs should be updated.

[js-choi]

For what it’s worth, there has been prior discussion of asynchronous reduction at jlongster/transducers.js#22. The potential for transducers to isolate transformations from both synchronous and asynchronous plural things is significant.