Fs2: Remove observe / observerAsync

[pchlupacek]

I would suggest to remove these combinators from 1.0.

The reasons fro this are :

1- Its hard to signal when chunk was / was not processed and semantics leads to strange errors where people write sinks that may /may not consume some elements and whole orchestration deadlocks
2 - the performance is low, overhead high
3 - in our codebase of like 500K fs2 lines we found only one usage, and that was in test and was possible to replace it with alternative A => F[Unit] // Stream.evalTap

[SystemFw]

I'm pretty surprised by this, I would assume observe is used all the time in user code.
If anything, I'd like to change the behaviour that makes the observed stream terminate if the Sink terminates _successfully_ , I find that counterintuitive.

[pchlupacek]

@SystemFw I think users are not really aware of how fragile observe is. I am really scared by this. Again if ppl are happy with it fine. Note though it is usually very simple to convert any sink to Resource[A => F[Unit]] or Stream[F, A => F[Unit]]. where still that function may depend on signal changing its state etc.

[SystemFw]

I do hear your concerns. I hope there can be something to do to improve it though instead of eliminating it, because I think there are patterns that aren't easily expressed otherwise. This applies to the current version of observe as well, because of that termination characteristic.

For example, I'd like to be able to print the first to elements of a Stream with:

myStream.observe(_.take(2).to(Sink.showLinesStdOut).through(restOfProgram)

[pchlupacek]

yeah this is exactly the example that kills me. the sink there terminates early, and did it confirmed what the 2 elements or the chunk or what? If it terminates, it is ok so the other elements are ignored? And why the hell everything blocks when that sink did not terminate at all ?

I am ok to keep it if we will find a precise and clean semantics for it. TO me it still sounds more like hack nowadays, really.

[SystemFw]

Another example, sending batches to a DB and then operating element by element

myStreamOfInts
   .observe(_.chunkN(3).to(DBSink))
   .map(_ + 1)

How do you do this easily with evalTap? You'd need to add an extraneous flatMap(Stream.chunk), losing modularity compared to observe, and so on for most other sinks which aren't simple A => F[Unit] or can be easily transformed to that.

[pchlupacek]

just to be clear is ok to see whats going on when you really see full code on line, but when you don't see that sink code, then you may have very heard time to reason about.

[SystemFw]

I am ok to keep it if we will find a precise and clean semantics for it.

Yes, I agree that the current semantics are a bit wonky, absolutely. Maybe we can design a different combinator to fill the role of observe in cases where evalTap is not enough, and encourage users towards evalTap by default? (I agree that like 90% of the time evalTap is enough, I'm just wary of making things inexpressible by users that aren't capable of writing the equivalent of observe themselves)

[pchlupacek]

I would like to see, if we really need sink as the way to write a program.

I remember we had a ton of Channel and Sinks in 0.8 code, but that is all gone with 0.9/0.10/1.0. Most of them was :

1. acquire resource
2. do (A => F[Unit]).repeat or (A => F[Unit]).mapIn(...).repeat or similar
3. clean resource

That could be very nicely expressed now with Resource.

[djspiewak]

@pchlupacek I agree that most sinks take on the form you're describing. However, a large number of them do not. Sometimes it is very useful to represent a series of linked states as a Sink. (I had a good example of this years ago that I can't remember right now) This is possible using Pull, which is quite nice, but it does require that people use the appropriate mechanism to apply that sink.

[pchlupacek]

@djspiewak I would be really interested to see that example. I remeber with 0.8 we did stuff quite often like

Signal.continuous.map { s => (a: A) => doWithState(s)(a):F[Unit] } : Sink[F, A]

but it turns out you can nicely do the same with StateT[F, S, ?] or similar abstraction.

[djspiewak]

My original issue was here: https://github.com/functional-streams-for-scala/fs2/issues/351

StateT doesn't conform to Concurrent or similar. And ultimately, it's a much less elegant way of representing a state machine than the mutual recursion you can do with Pull. A trivial example of what I mean is, imagine implementing FizzBuzz as a Sink[F, Int]. You need Pull to do it if you don't want to force F to implement MonadState or Sync (which you would then use to wrap a hidden var). Of those three solutions, Pull is far and away the most elegant.

[mpilquist]

Agreed on keeping observe. Can w shift focus to semantics we’d like to improve and implementation details we could change for better performance?

[pchlupacek]

Ah ok. I didn't meant that StateT alone is the solution, I am not strictly against defining this on Stream level.

In fact we really loved sinks, because they gave you so much flexibility in pre 0.9 time.We had sinks that worked on A, B, C then we were able to combine them for Coproducts and it was it. However all that is gone with 0.9+ and now is just pure function that does not obey anything and can do everything. Thats why I am scared to really promote it and give to users the combinators, that could do more harm.

[pchlupacek]

Ok folks I just wanted to bring it on. If it useful for you thats ok. As for the semantics, @SystemFw you had some ideas?

[SystemFw]

So, I need to have a closer look, but the one thing that springs to mind immediately is this:

If sink halts the evaluation will halt too.

I don't know if this is necessary or not, but it means that

myStream
  .observe(_.take(2).to(Sink.showLinesStdOut))
  .through(restOfProgram)

doesn't work in what I think is a counter intuitive way.

Note that observe will only output full chunks of O that are known to be successfully processed
* by sink. So if Sink terminates/fail in middle of chunk processing, the chunk will not be available
* in resulting stream.

This one is another subtle point, and I actually I agree with you that it's pretty wonky, but I don't have a clear idea yet on how it could be improved, without destroying chunkiness.

I think my more general point is that termination and failure of the Sink shouldn't be treated in the same way: I want to get notified of failure, but Sinks that terminate early are perfectly valid imho and shouldn't stop processing

[pchlupacek]

@SystemFw cool. So I'm gonna close this, hence this is to kill observe. And please feel free to open another issue.