A piped Stream should retain its Observer interface

[dariomannu]

Describe the solution

Hey, I just discovered Eventkit, I love it and would like to explore it better, as an RxJS alternative to power stream-oriented apps with Rimmel.js

Now, here's a little issue that I think is very important: in monadic law, given a monad m, the operation m flatMap n should return another monad of the same type of m.

This means, if Stream is both Observable and Observer, calling the pipe method on it should still preserve both interfaces, but currently it doesn't appear to.

const s = Stream(0).pipe(
  map(x=>x+1)
);

s.push(123) // should still work
s.subscribe() // just like this

There are some cases when it's desirable to have a read-only derived stream given a source, but other cases where having both is best.

Take this click-counter example. It doesn't work yet, but that's what I'm trying to achieve with this request.

In the code, total is a stream: a pipeline that takes events in and emits their count. It's conceptually one thing, doesn't make sense to split it, so we want to keep it as one, just like in the code, so we can connect both ends to an HTML template to make it talk to the real world (so we can also implement advanced FP concepts such as Extensible Effects in the most ergonomic way).

[hntrl]

Hey @dariomannu! Glad to hear it, and thanks for all the feedback!

This means, if Stream is both Observable and Observer, calling the pipe method on it should still preserve both interfaces

One quick correction here -- eventkit intentionally doesn't take on the same inheritance that comes from the observable contract. Observables here work functionally the same as those in the WICG proposal/rxjs/zen/your dogs favorite observable library/etc, but the primitives take on slightly different semantics because it's architecturally very different (see Motivations).

--
To be frank I'm not too familiar with monadic law, but a google search + claude thread leads me to believe that piping a stream means a flatMap operation -- if we're treating the values of a Stream as m, piping through map to get n, that means the output value is n, correct? It more has to do with what the output is instead of what interface is returned?

Take this click-counter example

Maybe just to help me better understand the rationale, how do you compare your example to something like this?

  const countStream = new Stream<number>();
  const total = stream.pipe(scan(x => x + 1);
  // additional example with multiple subcribers
  const squared = stream.pipe(scan(x => x * x));

  return rml`
    <button onclick="${countStream}">
      click me
    </button>

    Count: <span>${total}</span>
    Squared: <span>${squared}</span>
  `;

There are some cases when it's desirable to have a read-only derived stream given a source, but other cases where having both is best.

what are the cases for + against that come to mind?

so we can also implement advanced FP concepts such as Extensible Effects in the most ergonomic way

do you have an example of this? Am not familiar with what this is

[dariomannu]

Primitives:

The most important distinction between Observables and (sync/async) Iterables/Generators is that the former are push streams and the latter are pull streams. So, with Observables they decide whether and when to send data, in the latter case the consumer chooses when to pull and how much.

Both types can manage pressure when data is too much. Observables can buffer() or sample() data with these same operators and Generators can just decide to wait before yielding or yield*ing anything.

So, both have a similar [data in] => [processing] => [data out] anatomy, except the forces moving the data sit on the opposite ways of these streams.

If we start building with their similarities in mind, we can create a hybrid construct that's both a push and a pull stream. Given an Eventkit Stream, which is a pull stream, its very subscribe is the new moving force that internally pulls whatever is in the stream's internal Iterable and re-pushes it down to its subscribers.

The way we feeding these streams has also some similarities: for an Observer you do obs.next( newData ), for a Generator you do output = await gen.next( newData ). You can push into a Subject and expect something to come out. With a Generator you can only push data when you're ready to consume the previous chunk, then wait for something to be yielded later (this often creates some confusion).

Now, unifying the two primitives is the great opportunity here. "Observable", "Observer", "Iterable", "AsyncIterable" are just interfaces, contracts, but what happens internally it's just an implementation detail. This is why you've been able to create a library that's so similar to Observables extrnally, yet so much different internally (as you're using Generators).

Monads:

the "flatMap", aka "bind" operation may cause confusion because it has a different meaning in maths than in JS where you have Function.bind() and Array.flatMap(). The former is just a naming coincidence, so ignore that; the latter is a real flatMap operation that's also called flatMap (but it doesn't have to!).

• anArray.flatMap( fn ) is a flatMap operation, as the name happens to suggest.
• anArray.filter( fn ) is also a flatMap operation.
• anArray<number>.map( toString ) is also a flatMap. This shows how you turn an array of numbers into an array of strings.
• aPromise.then( doSomething ) is a flatMap operation, even if it's not called flatMap.
• anObservable.pipe( filter(something) ) is also a flatMap operation
• anObservable.filter( something ) is a flatMap, too, even if it uses an instance method vs a piped function. The difference is just an language-specific implementation detail. The true "flatMap" is a conceptual, mathematical operation we're representing here.
• Similarly, when a UI component's reactive template embeds other components is a flatMap operation, too:

const template1= rml`<button...>`;
const template2 = rml`<div>${template2}</div>`;
// we're doing flatMap by using ${}

Now, flatMap is an operation that must return a monad of the original (the container's) type: Array.flatMap => Array, Promise.then => Promise, Observable.pipe( something ) => Observable, and most importantly Stream.pipe( something ) => Stream.

It's irrelevant whether the returned type is an Array, a Promise, an Observable. It's the base type that matters. In fact, the whole point of flatMap is to map (to a collection of different values or different types), and flatten (which also means "lifting" it back to the original container type). You don't want to return an array of arrays, a promise of a promise, etc.

There may still be some confusion around Streams. What is a stream? Is it just the instance of the Stream class we're creating? How about the operators you attach to it?

The answer is actually... both:

Stream.op1().op2().op3() can be seen as 4 separate Streams, as Stream1.op1() returns a new Stream, then .op2() on it returns yet another Stream, etc.

At the same time, you can also treat the whole Stream.op1().op2().op3() as one monolyth. This is the beauty of composition.

Treating a Stream with its operators as a single entity is very convenient when writing applications.
A "counter" is a very convenient type of stream that takes something in and counts. We don't care about its internals, we don't want to decompose it.
This is especially true for more complex streams. Think of one that makes network requests, handles retries, shows a popup if network is down and asks the user to retry or cancel. You could split all this in 3, 4, 5 streams that pipe from each-other, but then you're left with a bunch of details nobody cares about.

A piped Stream is just as much of a useful grouping as it is a function as a group of instructions/other functions.

Hope this explains why a piped Stream should remain a Stream and not just a basic read-only Observable.

Example code:

Your example code is perfectly fine. You have a base stream and two derived ones.

How is it different from a single stream as in the basic click-counter?
According to what I wrote above, a basic click counter should be a single "entity", so it should be assigned exactly to one const counter.
In your example you have a base entity (the stream) and two derived streams. It makes perfect sense to split them the way you did because that's the exact architecture of your application.

Read-Only derived Streams

You may want to have read-only streams if you have security requirements. Say you're creating an app and you want to extend it with third-party plugins, you may want to only expose read-only Observables to make sure nobody can push data where they shouldn't...
Otherwise, if it's all your code in a secure context, read-write streams offer a lot of convenience,

Extensible Effects

The idea behind Extensible Effects, in an oversimplified explanation, is to make your code only declare effects rather than performing them. Somewhere else you'll have something that can interpret these declarations and apply the effects to the real world.

Now, given that we like architecting UI using components, which already return a declaration of effects (an HTML template is exactly that!), we can preserve the same component architecture to declare even ambient effects, such as window.onload, document.onmousedown, localStorage writes, etc.

There can be several ways to implement Extensible Effects and this is WIP in Rimmel, but one possible way could be using custom elements:

N.B.: What you'll see below looks very different from the original Extensible Effects you do in Haskell, the concept is the same and I see this as an ergonomic improvement because in the web platform we already have HTML, so we can choose to use it to declare any effects.

const component = () => {
  const stream1 = new Stream;
  const stream2 = new Stream<string>.pipe(
    map(str => str.toUpperCase())
  );

  const template = rml`
    <local-storage key1="${stream1}" key2="${stream2}"><local-storage>

    Key1: <input onchange="${ emit.value.into(stream1) }"><br>
    Key2: <input onchange="${ emit.value.into(stream2) }"><br>
  `;
};
```;

So, somewhere we'll define a custom element called `local-storage` that can be used to set or get values. Whether you're setting or getting values will be obvious from the context.

This is not released yet, it's just one path we're exploring at the moment, but Eventkit streams could very nicely fit in this scenario.

From your comments I understand you may also want to use Eventkit for native or server applications. This same pattern could be used there, as well. Suppose we had some hypothetical SML "Server Markup Language", we could implement a sort of Extensible Effects this way (roughly, types mostly ignored and oversimplified, for now):

```js
const Endpoint = () => {

  const handler = new Stream<Request, string>().pipe(
    whatever()
  );

  const errors = handler.pipe(
    catchError(...)
  );

  return sml`
    <endpoint path="/get-data" onrequest="${handler}">
      ${handler}
    </endpoint>

    <stderr>
      ${errors}
    </stderr>
  `
}

Same thing you could do for a CLI application. Suppose we had some hypothetical CML "Command-Line Markup Language", it could look something like this:

const Feature = () => {

  const handler = new Stream<stdin>().pipe(
    whatever()
  );

  const filedata = handler.pipe(
    serialize()
  );

  // .derive() would be like .pipe(), but making sure it creates a new, separate stream, pushable on its own:
  const errors = handler.derive(
    catchError(...)
  );

  return cml`
    <command onrequest="${handler}" />

    <stdout>
      ${handler}
    </endpoint>

    <stderr>
      ${errors}
    </stderr>

    <file name="output.json" onerror="${errors}">
      ${filedata}
    </file>
  `
}

Ah... I missed: TL;DR

[hntrl]

...we can create a hybrid construct that's both a push and a pull stream. Given an Eventkit Stream, which is a pull stream

I'm not totally with you on this. It's true that eventkit streams/observables do employ the "pulling" mechanism from generators, but I treat that purely as an implementation detail. If approaching streams from first principles when working with the API thats exposed by eventkit, I think it's very intentional that these are treated as push systems. (see the docs)

Also aren't push vs. pull mutually exclusive? The way I approach that conceptually is the push vs. pull classifier belongs purely to how a single subject interacts with the object, not as an aggregate of two parts. E.g. you can say that Streams are a push system, and a stream's subscriber is a pull system (though I don't think thats true, this is me just trying to wrap my head around what you mean by a "hybrid system"). But both of those are two separate actors.

Stream.op1().op2().op3() can be seen as 4 separate Streams, as Stream1.op1() returns a new Stream, then .op2() on it returns yet another Stream, etc.

I think the issue that I have with this is that it's unclear what the behavior of Stream's created in operations should behave. For instance:

const stream = new Stream<number>();

const streamA = stream.pipe();
const subA = streamA.subscribe(x => console.log("a:", x));

const streamB = streamA.pipe();
const subB = streamB.subscribe(x => console.log("b:", x));

streamB.push(1);
// if streamB is a monad of streamA and stream, does that mean this should log:
// "b: 1", "a: 1"

What about when types from piped operations are different? Are we expecting that piped Stream's retain the input value type if values published to the chain are emitted everywhere in the stream?

const streamA = stream.pipe(map(x => x.toString());
// do I push a number or a string to streamA?

If what you're saying is that piped streams are a union of 'piped stream' + a separate stream instance, then a piped stream is no longer a canonical representation of a predicate stream + some amount of transformations applied -- it's that plus whatever gets emitted by this new stream instance that I can call .push on!

There may still be some confusion around Streams. What is a stream? Is it just the instance of the Stream class we're creating?

If we're talking about Stream as an eventkit primitive, I've been treating them a little bit differently then what you're describing I think. From the docs:

"A Stream is a special type of AsyncObservable that allows you to push values into it at any time."

In other words, a Stream is an implementation of AsyncObservable which lets you call .push as the mechanism for how values get emitted.

Now, flatMap is an operation that must return a monad of the original (the container's) type:

With the above in mind, and since the foundational type of a Stream is AsyncObservable, Stream.pipe should return AsyncObservable then. Maybe it's an arbitrary line in the sand to decide which of those primitives we consider "the monad", but if we're treating the observable as the foundational unit (which I believe to be true since Stream extends AsyncObservable) then we're not breaking convention here.

This isn't reason enough to justify why Stream doesn't need to change in this respect, but it is common convention that piped observable implementations return observables instead of the original object (i.e. BehaviorSubject.pipe returns an Observable)

[dariomannu]

...we can create a hybrid construct that's both a push and a pull stream. Given an Eventkit Stream, which is a pull stream

I'm not totally with you on this. It's true that eventkit streams/observables do employ the "pulling" mechanism from generators, but I treat that purely as an implementation detail. If approaching streams from first principles when working with the API thats exposed by eventkit, I think it's very intentional that these are treated as push systems. (see the docs)

Also aren't push vs. pull mutually exclusive? The way I approach that conceptually is the push vs. pull classifier belongs purely to how a single subject interacts with the object, not as an aggregate of two parts. E.g. you can say that Streams are a push system, and a stream's subscriber is a pull system (though I don't think thats true, this is me just trying to wrap my head around what you mean by a "hybrid system"). But both of those are two separate actors.

I would say Observables are push streams until you apply any buffering/throttling/debouncing operator.
Once those run, they take control over the flow of data and act just like any pulling/polling mechanism would do. There's kind of zero difference between debounceTime(1000) and polling every second for the last value.
The output is another push stream, though, as these operators also decide when to push that data downstream.

Something similar happens in reverse, too. If you're consuming an async generator through a for..of loop, you're pulling data that's a sequence of promises, so if the generator wants to add any additional delays, it will effectivel act as a pushing step in a pull stream.

Anyway, I don't think we're disagreeing on any of this, it's just the push and pull words that have some "flexible" meaning.

Stream.op1().op2().op3() can be seen as 4 separate Streams, as Stream1.op1() returns a new Stream, then .op2() on it returns yet another Stream, etc.

I think the issue that I have with this is that it's unclear what the behavior of Stream's created in operations should behave. For instance:

const stream = new Stream();

const streamA = stream.pipe();
const subA = streamA.subscribe(x => console.log("a:", x));

const streamB = streamA.pipe();
const subB = streamB.subscribe(x => console.log("b:", x));

streamB.push(1);
// if streamB is a monad of streamA and stream, does that mean this should log:
// "b: 1", "a: 1"

Excellent point, you spotted it very quick!
So, what I would do is making the stream creation an immutable process in one of the following ways:

1. .pipe() binds a list of operators to a stream irreversibly, so you can't call it twice. If you want to create a derived stream, we add a .derive() method, to make it explicit that now we are creating a new stream. It would be an interesting one, because you would be able to feed it from two different sources: its original source stream and now directly, via the reference just created:

const derived = source.derive(
  map(something)
);

source.push(...)
derived.push(...)

2. change the behaviour of .pipe() such that the first time it's called on a stream, it adds operators to it, otherwise it creates a new derived stream like in the previous option:

const source = new Stream().pipe(
);

const derived = source.pipe(
);

source.push(...) // pushes both to `source` and through `source` into `derived`
derived.push(...) // only pushes into `derived`

Weird? Definitely. A simple use case?

const numbers = Stream<number>;
const strings = numbers.pipe(
  map(toString)
);

numbers.push(7);
strings.push("7");

If what you're saying is that piped streams are a union of 'piped stream' + a separate stream instance, then a piped stream is no longer a canonical representation of a predicate stream + some amount of transformations applied -- it's that plus whatever gets emitted by this new stream instance that I can call .push on!

Ok,wrt typing,I think an Observable is fine to have a single type, but a Stream should have an input type and and optionally distinct output type. Internally we are already doing this:

type Stream<I, O=I> = Observer<I> & Observable<O>;

What about when types from piped operations are different? Are we expecting that piped Stream's retain the input value type if values published to the chain are emitted everywhere in the stream?

const streamA = stream.pipe(map(x => x.toString());
// do I push a number or a string to streamA?

streamA should really be a Stream<number, string>, so it's right to push a number.
streamB, unless it maps to different types in its pipeline, would remain a Stream<string>, so you would push a string into it.

If we're talking about Stream as an eventkit primitive, I've been treating them a little bit differently then what you're describing I think. From the docs:

"A Stream is a special type of AsyncObservable that allows you to push values into it at any time."

In other words, a Stream is an implementation of AsyncObservable which lets you call .push as the mechanism for how values get emitted.

That makes perfect sense

Now, flatMap is an operation that must return a monad of the original (the container's) type:

With the above in mind, and since the foundational type of a Stream is AsyncObservable, Stream.pipe should return AsyncObservable then. Maybe it's an arbitrary line in the sand to decide which of those primitives we consider "the monad", but if we're treating the observable as the foundational unit (which I believe to be true since Stream extends AsyncObservable) then we're not breaking convention here.

This isn't reason enough to justify why Stream doesn't need to change in this respect, but it is common convention that piped observable implementations return observables instead of the original object (i.e. BehaviorSubject.pipe returns an Observable)

Well, that's what the documentation and the types say, but it's actually not the case.
BehaviorSubject.pipe(anything), with some darker magic happening behind the scenes, returns effectively the same Behaviorsubject, piped. Technically it's still a new Observable, but one that exposes the original BehaviorSubject's Observer interface, so you can .next into it and that's exactly how stuff like the click-counter works using just just a single, piped BehaviorSubject.

Anyway, this seems to be changing with RxJS8+ and I've been complaining, but still need to make them understand and agree on the value of these piped streams as first-class citizens. RxJS8 is kind of abandoned and will be replaced by 9 with native observable support, so we'll see what's going to happen.

One thing I quite like in Rx8, though, is the new rx(...pipeline) function, which makes it easier/shorter, I'd say more ergonomical to create Observables.

I believe the main reason they have against returning another BehaviorSubject is a lot of abuse, especially in Angular, where devs have been creating Subjects/BehaviorSubjects and imperatively pushing them, which is as big of an anti-pattern as it is mutability. If done from other reactive constructs like Promises or Signals, or other Observables, it's even worse and can introduce serious bugs and broken functionality.

However, when doing stream oriented programming, these read-write streams are an essential primitive, a first-class citizen we use to keep code cleaner than ever (we have a growing number of examples showing exactly how)

RxJS, on the other hand, is an Observable library, to make it possible to use Observables inside a variety of applications, so the Stream-Oriented paradigm may not be their top priority at this point.