Inner Flow for Vitaly

innerFlow.sc

#!/usr/bin/env amm
import $plugin.$ivy.`org.typelevel:::kind-projector:0.11.0`

import $ivy.{
  `co.fs2::fs2-core:2.0.1`,
  `com.github.julien-truffaut::monocle-generic:2.0.0`
}
import cats.Applicative
import fs2._
import monocle._
import monocle.std.all._
import monocle.syntax.all._

def innerFlow1[F[_], Person, Name, CTX](s: Stream[F, (Option[Person], CTX)])(get: Person => Name, set: Person => Name => Person)(innerFlow: Pipe[F, Name, Name]): Stream[F, (Option[Person], CTX)] = {
  s.flatMap {
    case (Some(p), ctx) => Stream.emit(get(p)).covary[F].through(innerFlow).map(set(p)).map(p => (Some(p), ctx))
    case x@(None, _)    => Stream.emit(x)
  }
}

//with Lenses (naive)
def innerFlow2[F[_], Person, Name, CTX](s: Stream[F, (Option[Person], CTX)])(lens: Lens[Person, Name])(innerFlow: Pipe[F, Name, Name]): Stream[F, (Option[Person], CTX)] = {
  def mod(p: Person, ctx: CTX): Name => (Option[Person], CTX) =
    name => (Some(lens.set(name)(p)), ctx)
  s.flatMap {
    case (Some(p), ctx) => Stream.emit(lens.get(p)).covary[F].through(innerFlow).map(mod(p, ctx))
    case x@(None, _)    => Stream.emit(x)
  }
}

//with Lenses (leveraging Optional#modifyF)
def innerFlow3[F[_], A, B, C](lens: Lens[A, B])(innerFlow: Pipe[F, B, B]): Pipe[F, (Option[A], C), (Option[A], C)] =
  _.flatMap(
    _1[(Option[A], C), Option[A]]
      .composePrism(some)
      .composeLens(lens)
      .modifyF(Stream.emit(_).through(innerFlow)))