Designing an environment which consumes and re-exposes a stream

[ScalaSci]

I'd like to design an app pattern which can use the start of a stream to instantiate an environment, and then provide the rest of the stream to an app, which is modeled as a sink... I am trying to develop ZLayer-style composition patterns to allow creating complex environments based on stream activity.

Possible uses include using a stream of timestamped input events to update the test clock, or simply tracking and composing stats for the incoming stream in the environment.

Following is a snippet of one of my attempts at implementing this:

case class CausalEnvironment[R0, EIn, EOut>:EIn, R, Elem, ElemOut](
    env: ZChannel[R0,
                  EIn,
                  Chunk[Elem],
                  Any,
                  EOut,
                  Nothing,
                  (ZEnvironment[R], ZChannel[Any, EIn, Chunk[Elem], Any, EOut, Chunk[ElemOut], Any])]) {
  def >+>[R1, E1Out>:EOut, ElemOut1](other: CausalEnvironment[R, EOut, E1Out, R1, ElemOut, ElemOut1])
    : CausalEnvironment[R0, EIn, E1Out, R with R1, Elem, ElemOut1] = {

    val chan  =
      env.flatMap{case (r0,downstream0)=>
        //unsure about re-using downstream0 here
        downstream0>>>other.env.provideEnvironment(r0).map{ case (r1, downstream1) =>
          //...and here     V
          r1.unionAll(r0)->(downstream0>>>downstream1)
      }

    }

    CausalEnvironment(chan)
  }

  def buildCausalApp(s: ZSink[R, EIn, ElemOut, Any, Any]) = (s0: ZStream[Any, EIn, Elem]) => {
    s0.pipeThroughChannel(
        env
          .flatMap {
            case (r, c) =>
              c.orDieWith(e => new Exception(e.toString)).>>>(s.channel.provideEnvironment(r))
          })
      .runDrain
  }
}

I am concerned about how I re-use the channel yielded by env, downstream0. If these were streams, this would almost definitely be wrong, though channels seem like they may be simple values in this context, thus safe for re-use? In a few of my tests--calculating running max and min on the incoming stream and yielding them as queryable environment services, I appear to be getting the desired behavior. Is this pattern going to cause problems down the line? Is there a better way to achieve this design aim?

[pvgoran]

So your CausalEnvorinment object provides two kinds of channels: the "startup" channel (env) and the "running" channel (the second component of the end result of env); the "startup" channel processes the first elements from the stream, produces the environment and the "running" channel, which then processes the rest of the stream.

If several CausalEnvorinments are chained, each of them will "eat" a segment of the input with its "startup" channel; so with CausalEnvorinmentA >+> CausalEnvorinmentB >+> CausalEnvorinmentC, CausalEnvorinmentA will eat N first input items, CausalEnvorinmentB will eat M next input items (first processed by the "running" channel of CausalEnvorinmentA), CausalEnvorinmentC will eat K next input items (first processed by the "running" channels of CausalEnvorinmentA and CausalEnvorinmentB), and the remainder will be processed by three "running" channels and then fed to the sink. So the input will be "split" into three "startup" segments and the last "remainder" segment.

Is this a correct assessment of the intent?

If so, then re-using the "running" channels produced by the "startup" channels should be fine, as long as these "running" channels are designed to be run multiple times (one time with each subsequent CausalEnvorinment down the line while processing it "startup" segment, and one final time with the sink while processing the "remainder" segment) and are not expected to reliably maintain internal state (because the internal state, if any, will be reset with the beginning of each of these segments).

(But take my opinion with a grain of salt: my experience with ZIO, and with high-level monadic/functional programming in general, is limited to just about one month.)

[ScalaSci]

Yep you've got the intent right. I think it might be an issue, though, if the channels are somehow stateful. What would apparently happen, is the state of downstream0 would be reset after consuming elements to produce other.env. This is probably not what I want. I'm beginning to think it might be better to model these as sinks and pipelines to explicitly disallow the downstream channels being stateful.

[adamgfraser]

A channel is just a description, so it is fine to "run" a channel multiple times though to the extent running the channel has changed external state (e.g. taking an element from a queue) then that external state may be different the next time the channel is run and may cause running the channel a second time to produce a different result.