Implement streaming with state changes

[nightscape]

This is a draft of a "correct" streaming API where the state changes with the emitted stream elements.
@softinio @mtsokol @mijicd could you have a look if this looks plausible?

I'm not sure what would be the correct behaviour of any element in the stream failing. Should the supervisor be called in that case or would we just continue streaming?

In order to maintain the current Stateful API, one could introduce an additional method receiveS (or any name you prefer) and do something like the following:

def receiveS[A](state: S, msg: F[A], context: Context): ActorResponse[R, S, A] = oneTimeResponse(receive(state, msg, context))
def receive[A](state: S, msg: F[A], context: Context): RIO[R, (S, A)] = ???

That way users could implement/override the new receiveS if they need the additional functionality or receive if they don't.
The downside of this is that IDEs will not know that one of the two methods has to be overriden because both contain an implementation.
I'd be happy for any suggestions 😃

[nightscape]

I'm also just writing a RemoteSpec for the ZStream case and it shows that my approach in #269 was too naive 😉
I'm getting all kinds of ClassCastExceptions now because it tries to cast a single element from the stream into a ZStream.

[nightscape]

@softinio @mtsokol @mijicd I could need some help 😅
I've added a test that should stream 10 messages back to the requester, but after 2 messages it just stops.
I couldn't find out yet what's happening, if there's a channel that's being closed, or my usage of ZStream#repeat or ZStream#takeWhile is incorrect.
Would you have any idea?

[mtsokol]

@nightscape Hi! Thanks for reminder!
I'll have some time later this week to fully look at it. Do you mind waiting?

[nightscape]

Hi @mtsokol, thanks for the offer!
I can do other things in the meantime, so later this week is fine 👍

[nightscape]

Hi @mtsokol, did you already find some time to have a look at this?
I don't need a full analyis or solution, but some hints that point me in the right direction could help me to figure this out 😃

[mtsokol]

@nightscape Sorry for the delay (started new semester this week).
Thanks for a ping, I wanted to do in-depth review of the PR but I ended up fighting with that issue you described (the test that should stream 10 messages but ends after 2). After a while of debugging I think that the issue is with .takeWhile from here.

Here are snippets reproducing it:

• with takeWhile
• with takeUntil
  WDYT?

After changing it to takeUntill with e == StreamEnd condition all 10 elements from stream are send and received. But unfortunately test hangs after receiving StreamEnd on runCollect so there's still something about handling this stream (or in socket communication), will look at it.

I only focused on this issue so I'll deliver rest of the review.

[EDIT]
I believe it's due to the implementation of ZStream#takeWhile, here's modified version that works (will ask about it):
https://scastie.scala-lang.org/RHDosRy0TdGjMvOP881JWQ

[nightscape]

Great find! That actually looks like a bug in takeWhile.
Thanks a lot for digging into this!
I was always searching for some problem w.r.t. closed connections or the like and didn't take the implementation of any ZStream methods into the set of suspects.

[mtsokol]

@nightscape I got this test passing but there's still one issue (with deserialization I guess?). The whole stream is producing Strings. If I augment producer site with: (stream ++ ZStream("StreamEnd")) instead of (stream ++ ZStream(StreamEnd)) it's all correctly consumed. Also on consumer site after .takeUntil I added .filterNot(_ == "StreamEnd") as takeUntil adds that termination message. After that test completes correctly.
I don't know concrete reason of this and how to handle it correctly yet
(maybe add wrapper stream message that is being send as protocol? WDYT?)

sealed trait StreamProtocol
case class StreamMsg(obj: Any) extends StreamProtocol
case object StreamEnd extends StreamProtocol

And .map this Stream in sendEnvelope after .filterNot to get inner messages? (I guess there's better solution)
Let me know if you want me to send this snippet or if have more questions.

[nightscape]

@mtsokol is this what you had in mind? 97ed0eb

[nightscape]

And I guess this should fix the takeWhile problem: https://github.com/zio/zio/pull/4273/files#diff-f8e5bb3d67d9b3c721d3cac422fb50f6R2502
Unfortunately, we can't rely on users using ZIO 1.0.2, so the current implementation is safer.

[nightscape]

From my side everything except supervision of streams is looking good.

[nightscape]

@mtsokol @softinio @mijicd any ideas how Supervision could/should work in the streaming case?

[mtsokol]

I think that ResponseType here should be ZStream[R, Throwable, (S, I)].
As actors are location transparent and interaction can involve remote communication our interaction patterns ! and ? result in Task[...] (so a ZIO that can fail with Throwable as it might involve write/readFromWire) although in actor's behavior there could be no failures that can occur.

Notice that here https://github.com/zio/zio-actors/pull/270/files#diff-c1a93d1001aa018540bc59b83dd7a09dR134 stream elements are created from Task[Any] so resulting stream that is returned will be Stream[Any, Throwable, Any].

That is - a stream returned in our RemoteSpec will perform readFromWire to get elements (until receives StreamEnd) so every time it might fail with communication error but here it test the signature is ZStream[Any, Nothing, String] which seems that pulling elements out of stream can't fail.

I propose to remove E from implicit def stream[R, S, E, I] and fix returned ZStream signature to ZStream[R, Throwable, I].

This might seem unclear but if we notice where communication occurs it should be clear - when we do ? and the response it oneTime the call is synchronous and we wait for the response.
When the response is a stream on the caller site we only create a recipe with ? -> the communication occurs when we do .runCollect and this is a counterpart of getting response out of ? with oneTime.

Please correct me if I got this wrong.

---

[EDIT]

Problem

I think that support for streaming responses will require overhaul for remoting (due to the fact that we decide when we pull responses out of stream, so perform remote communication) as it breaks current design.

I introduced simple change to our test case:

_              <- actorRef ? Str("ZIO-Actor response... ")
resultStream   <- actorRef ? Strs(messages)
result         <- actorRef ? Str("ZIO-Actor response... ")
resultStream2  <- actorRef ? Strs(messages2)

And it fails - first we do the first ask -> remote actor system writes to the wire - we read it.
Then we ask for a stream response and get the stream - the remote actor system writes all responses to the wire but we do not read them (!) but only define recipe for reading them. Then we ask for another oneTime result but we read a bit of those stream responses instead and it fails. This should definitely be covered in a test case like this (intertwine oneTime and stream interactions)

Solution?

Off the top of my head - we can perform runCollect in sendEnvelope but it misses the point of #268 what you described as your goal.

You could just make it fully akka-like and pass ActorRef in a message you send as a part of your protocol (so you know where send messages back) - and perform all communication with ! (tell) and drop stream responses.

Otherwise it would require rewriting internals (which are really basic now) to support it that way.

WDYT?

Sorry for writing about this issue now - only now I had some time to think about it deeper.

[nightscape]

Regarding ZStream being a pull-based stream, but the responses being pushed over the wire:
Would it help to introduce a Queue at the receiver side instead of doing a runCollect?
That way the remote actor can push messages at any time, but from the client side we pull from the Queue.

I had also thought about the Akka way of passing an ActorRef and sending responses via !, but the resulting API would be quite a bit less elegant...

[mtsokol]

Right now I think the issue on response producer site is:

(stream.map(StreamMsg) ++ ZStream(StreamEnd)).foreach(e =>
  writeToWire(worker, e)
)

If we immediately write stream elements to receiver like this then they need to be read right away (e.g. with runCollect), otherwise remoting can be easily broken (but it makes no sense as we are blocked until receive the whole stream).

I don't know a solution for this right now (maybe another abstraction layer for remote communication for atomic and stream responses where stream response elements will be stored on response producer site and only send over the wire when requested (but it would introduce other issues I think), or (also with that additional abstraction layer) response producer will send it like this but on the other side we need to manage reading ? responses and those out-of-order stream responses, or maybe something different)

Yeah, including sender ActorRef in protocol would make it less elegant but making stream responses that comply with current remoting feels to require major design change.

[nightscape]

Would it be possible to have a daemon on the receiver side that permanently tries to readFromWire and pushes into a Queue from which both ZStreams and answers from ? can be read?
StreamMsg would then need some kind of "conversation identifier" so that when multiple streams are running at the same time their messages can be distinguished.

If we can come up with a working solution, would you see the overall goal as worthwhile or do you want to have a simple, no-bells-and-whistles library?

[mtsokol]

Thanks, sounds great but I lack in knowledge and experience to elaborate on such solution (what are the pros and cons). Do you know a project where such solution is used?
As this is a design discussion also asking @mijicd @softinio. What do you think about it?