Can I feed items into ZSink manually?

[pvgoran]

I have a task combinator that repeatedly executes the task, retries if it fails, and reports the execution status to the provided ZSink. The simplified code (without retrying, for brevity):

  def repeatAndFeedStatus(
    initTask : Task[Unit], repeatTask : Task[Unit], repeatDelay : Duration, retrySchedule : Schedule[Any, Throwable, Unit],
    statusSink : Sink[Throwable, RunStatus, Nothing, Unit]) : Task[Nothing] =
  {
    val repeatSchedule : Schedule[Any, Unit, ?] = Schedule.fixed(repeatDelay)

    for {
      statusQueue : Queue[RunStatus] <- Queue.bounded(1)

      runLoop : Task[Nothing] =
        initTask
        *> (repeatTask.tap { _ => statusQueue.offer(RunStatus.Running) })
          .repeat(repeatSchedule)
        *> ZIO.fail(IllegalStateException("Repeat isn't expected to complete"))

      result <-
        (statusQueue.offer(RunStatus.Starting) *> runLoop)
        <& ZStream.fromQueue(statusQueue).run(statusSink)
    } yield result
  }

I have some troubles with the "report" part. As you can see, I use a Queue to emit RunStatus values, and this Queue is then streamed to statusSink in parallel with the main computation. The caller defines statusSink like this:

  val statusSink : Sink[Throwable, RunStatus, Nothing, Unit] =
    ZSink.foreach(status => zio.Console.printLine("Status: " + status))

This works, but the printLine computation above executes out-of-sequence with the repeatTask computation (the latter can begin before the former completes). This is expected due to parallelism, but I'd like sequential execution, so that the RunStatus value is first consumed by statusSink, and then repeatTask is executed.

I would (probably) be able to achieve this if I could feed RunStatus values into statusSink directly, without running the auxuliary Stream. However, I don't see a way to do this; is it possible?

Also, it's entirely possible that I'm trying to use inappropriate abstractions for the thing I'm trying to accomplish.

For example, I might have used a statusTask : Task[RunStatus] instead of statusSink, and just run this computation on each state change; however, I feel that ZSink better captures the intent: it's a "receiver of stream of events", which is exactly what I need, and a Task is just a generic "thing to execute", which doesn't capture the intent of repetition and sequentiality.

Also, I could try to return a ZStream of RunStatus values instead of feeding a provided ZSink, but (1) it looks less flexible, and (2) I don't see a natural way to convert the composite task created in repeatAndFeedStatus to a stream.

(This was also posted on StackOverflow. Let me know if such cross-posting is inappropriate.)

[adamgfraser]

If you want to run the sink before repeating you could do something like statusQueue.offer(RunStatus.Starting) *> ZStream.fromQueue(statusQueue).run(statusSink) *> runLoop. I'm not sure how much this example is simplified from your actual use case but the sink doesn't seem to be doing a lot of work here and I wonder if you want to use it all versus a simple ZIO workflow or function returning a ZIO workflow.

[pvgoran]

Signaling from the sink implementation is an unnecessary complication that it (the sink) shouldn't have to deal with. For now, I managed to serialize running of the sink and running of repeatTask inside repeatAndFeedStatus with the help of an MVar and a Semaphore; it's not elegant (and possibly not performant), but it gets the job done in an encapsulated way (this implementation detail is nicely hidden inside repeatAndFeedStatus).

I already considered using a ZIO instead of a sink (it's in the original question), but as I stated, if I do this I lose the semantics of repetition and sequentiality that streams/sinks have. So it's not an optimal solution in terms of expressiveness.

So, back to the original question: is it not possible to feed items into a ZSink manually?

[adamgfraser]

You can feed items to a sink manually by offering them to a queue. ZStream.fromQueue(queue).run(sink).fork and then queue.offer.

[pvgoran]

You can feed items to a sink manually by offering them to a queue. ZStream.fromQueue(queue).run(sink).fork and then queue.offer.

Well, that's what I'm already doing, but it doesn't look manual to me because it's the stream that does the actual feeding, in its own fiber. (I suppose I should have written "directly" instead of "manually".) And it doesn't have the properties that I was hoping for. :(

Anyway, thanks for your input.

[adamgfraser]

Yeah, I think you are looking for this further property that writing to the sink doesn't return until the sink has "completed" processing that element and the most that could be guaranteed is that the sink has "started" processing the element. I feel like this problem could be more easily solved with ZIO or possibly with Schedule.

[pvgoran]

Yeah, I think you are looking for this further property that writing to the sink doesn't return until the sink has "completed" processing that element and the most that could be guaranteed is that the sink has "started" processing the element.

Yes, this may be the case. I did further testing of the solution that I posted in my answer, and when I add a delay to the sink implementation, it does not currently work like I want it to. Maybe your other answer will lead to a better result.

[adamgfraser]

Isn't this just something like (working off your original example)?

import zio._
import zio.stream._

object Example {

  sealed trait RunStatus
  object RunStatus {
    case object Running  extends RunStatus
    case object Starting extends RunStatus
  }

  def repeatAndFeedStatus(
    initTask: Task[Unit],
    repeatTask: Task[Unit],
    repeatDelay: Duration,
    retrySchedule: Schedule[Any, Throwable, Unit],
    statusSink: Sink[Throwable, RunStatus, Nothing, Unit]
  ): Task[Unit] = {
    val repeatSchedule: Schedule[Any, Any, Any] =
      Schedule.fixed(repeatDelay)
    val stream: ZStream[Any, Throwable, RunStatus] =
      ZStream.succeed(RunStatus.Starting) ++
        ZStream.repeatZIOWithSchedule(initTask *> repeatTask *> ZIO.succeed(RunStatus.Running), repeatSchedule) ++
        ZStream.fail(new IllegalStateException("Repeat isn't expected to complete"))
    stream.run(statusSink)
  }
}

I think the fundamental problem is you are using a streaming data type here with a sink and that is going to work best if you use it with other streaming data types or if you don't want to do that just use ZIO.

[pvgoran]

So it's using a Stream all the way... I wasn't sure this was possible (and I'm still not sure that I can make a version that also does error handling/retrying in addition to normal repetition) and I'm not familiar with the way streams are composed, but it may be worth a try.

[pvgoran]

I managed to implement the full functionality of the function via streams, as suggested:

def repeatAndFeedStatus2(
  initTask : Task[Unit], repeatTask : Task[Unit], repeatDelay : Duration, retrySchedule : Schedule[Any, Throwable, Unit],
  statusSink : Sink[Throwable, RunStatus, Nothing, Nothing]) : Task[Nothing] =
{
  val repeatSchedule : Schedule[Any, Any, ?] = Schedule.fixed(repeatDelay)

  Ref.make(false).flatMap { (failedRef : Ref[Boolean]) =>
    val loopStream : ZStream[Any, Throwable, RunStatus] =
      ZStream.whenZIO(failedRef.get) { ZStream.succeed(RunStatus.Recovering) }
      ++ ZStream.fromZIO(initTask).drain
      ++ ZStream.repeatZIOWithSchedule(repeatTask *> ZIO.succeed(RunStatus.Running), repeatSchedule)
      ++ ZStream.fail(IllegalStateException("Repeat isn't expected to complete"))

    val loopWithRetryStream : ZStream[Any, Throwable, RunStatus] =
      loopStream
        .either
        .flatMap[Any, Throwable, RunStatus] {
          case Right(runStatus) => ZStream.succeed(runStatus)
          case Left(ex) =>
            ZStream.fromZIO(failedRef.set(true)).drain
            ++ ZStream.succeed(RunStatus.Failing(ex, Instant.now()))
            ++ ZStream.fail(ex)
        }
        .retry(retrySchedule)

    (ZStream.succeed(RunStatus.Starting) ++ loopWithRetryStream).run(statusSink)
  }
}

This seems to be a satisfactory solution to the specific problem that I faced. It doesn't use additional state or parallelism, which is definitely a win. Error handling is unwieldy, but not too much.

[pvgoran]

So apparently the answer to the question "Can I feed items into ZSink manually?" is "No".

However, thanks to @adamgfraser's suggestion, I was able to implement the repeatAndFeedStatus function via streams, see here.

After some more thought, it occured to me that I actually wanted some kind of coroutine mechanics, using a sink as "something to give intermediate results to". I'll write about this separately.

[pvgoran]

Here are my thoughts on coroutines: #7199