Add HTTP/websocket event sources

[ajaychandran]

I am currently using these as HTTP event sources.
Would it make sense to add these to the library?

[lolgab]

You can wrap Ajax requests this way:

import org.scalajs.dom.ext.Ajax
EventStream.fromFuture(Ajax.get(url))

[ajaychandran]

@lolgab
Yes, but the Future is evaluated eagerly.
In contrast, AjaxEventStream submits the request in onStart, thus delaying effects (network request) until observers are registered.

[lolgab]

Then maybe there should be something to integrate things lazily but on a more general level.
Integrating Ajax directly in Airstream sounds too specific to me.

[raquo]

I think having a reference implementation of doing basic stuff like Ajax would be useful. There are three potential places for it:

• In Airstream itself
• In a third party library that depends on Airstream
• As a live example on laminar.dev if publishing a library is too much overhead

Stuff that we add to Airstream should be pretty unopinionated. For example, I've been wanting to move DomEventStream from Laminar to Airstream for a while now. Given that the implementation of AjaxEventStream is pretty bare bones, I think it would be nice to include it in Airstream itself.

A few notes though:

• it should go into a new directory, airstream/web, since it's web-specific
• You should use AjaxException class from org.scalajs.dom.ext.Ajax , Airstream already depends on scala-js-dom
• Inside onreadystatechange you should only do stuff if (isStarted). That way it will behave more predictably if you start then stop then start your stream again before the first response comes back.
• I'd add the methods for put / patch / delete too for the sake of completeness.

---

there should be something to integrate things lazily but on a more general level

Currently one can achieve this with something like:

def fromLazyFuture[A](future: => Future[A], emitOnce: Boolean): EventStream[A] = {
  EventStream
    .fromValue((), emitOnce =emitOnce)
    .flatMap(_ => future)
}

I guess I could add this EventStream and a similar one to Signal.

[raquo]

Oh and, I didn't have the time to evaluate the websockets helper yet. I think websockets might require a more opinionated API, but I'll check it out a bit later.

[ajaychandran]

Thanks for your suggestions. I will incorporate them and submit a PR.

Regarding websockets,

• I would like to have single implementation that is bi-directional (and can be used for the uni-directional case). I will work on this in a separate PR.
• It would be nice to have a retry/restart-on-error mechanism based on some strategy, similar to ZIO Schedule. Have you considered this before?

[raquo]

Ok wrt websockets! Try to keep it unopinionated so that users can build their custom code based on it, like you did with ajax.

regarding "restart on error" – I'm guessing recoverToTry followed by flatMap can achieve this, but it could be combined into a single operator like recoverTo(err => observable) or recoverTo(observable). The latter one can be made to behave better wrt glitches. Not 100% sure about the other one, that might require a transaction boundary similar to the flatMap solution.

regarding retrys – what kind of operators / signatures are you thinking about, specifically? Kinda hard to translate ZIO stuff directly as it has many more concepts than Airstream. We have EventStream.periodic which is flexible enough to be made exponential or fibonacci, and that can probably be combined into a retry pattern again with flatMap. If there is a canonical use case for retrys in Airstream, we could include it as a standalone operator.

[ajaychandran]

The usecase I had in mind is to restart a websocket stream when the underlying connection stops, for instance, when the user loses connectivity. When connectivity is resumed, it may be desirable to establish a new websocket connection automatically.

One way to do this is to use a control parameter like maxReconnectAttempts in the WebSocketEventStream constructor.
I was wondering if this could be generalized to a stream combinator.

[raquo]

Yes, sorry, I worded it poorly, I understand the use cases for retrys, but I was also wondering the same thing, whether you had a specific method signature and behaviour in mind that would be general enough to include as an operator for all streams. I'm thinking it could be a certain variation of recoverTo.

[ajaychandran]

How about restartOn(control: EventStream[Any])?
This is flexible enough to accommodate a wide variety of usecases (restart max n times, restart after d duration, restart max n times + after d duration).
Can this be implemented by calling onStop followed by onStart?

[raquo]

But, this would only work for streams that perform a side effect (that we want to re-run) when they start. Most streams don't do that.

Even if you're listening to a stream that depends on one of those streams that perform a side effect on start, and you stop & start that child stream, that is no guarantee that the parent stream that actually performs the side effect will be restarted (that will only happen if the child stream was its only listener).

I think the problem is, streams aren't effects. You can retry an effect like IO or Task or even Future, but retrying a stream doesn't seem to make sense in general. Or, to the extent that it does, the retry logic needs to be contained within the stream that actually performs the side effect. So if there is any reusability to be gained there, it does not seem like it will be in the form of a general purpose operator.