基于Shapeless与FS2将Coproduct流转换为HList流

我需要实现一个FS2 Pipe[F, C, H]，将Stream[F, C]（其中C <: Coproduct）转换为Stream[F, H]（其中H <: HList，包含C的所有成员）。Pipe的逻辑是：等待收集到C的每个成员至少一个后，将它们组合成HList输出。

使用示例如下：

type MyCoprod = A :+: B :+: C :+: CNil
type MyHList = A :: B :: C :: HNil

val stream: Stream[F, MyHList] = Stream
  .emits(List(A, B, C)) // 输入的Coproduct流
  .through(pullAll) // 等待收集齐A、B、C后输出对应的HList
  .map { hlist => ... }

作为Shapeless新手，我写出了部分代码框架但遇到瓶颈：

trait WaitFor[F[_], C <: Coproduct] {
  type Out <: HList

  def apply: Pipe[F, C, Out]
}

object WaitFor {
  type Aux[F[_], C <: Coproduct, Out0 <: HList] =
    WaitFor[F, C] { type Out = Out0 }

  implicit def make[F[_], C <: Coproduct, L <: HList](implicit
    toHList: ToHList.Aux[C, L]
  ): Aux[F, C, L] = new WaitFor.Aux[F, C, L] {
    override type Out = L

    override def apply: Pipe[F, C, Out] = {
      def go(s2: Stream[F, C], currHList: L): Pull[F, L, Unit] = {
        s2.pull.uncons1.flatMap {
          case Some((coproduct, s3)) => {
            // 将Coproduct成员添加或更新到currHList
            // 如果currHList包含所有成员，则输出并清空状态
            // 否则继续迭代
            go(s3, ???)
          }

          case None => Pull.done
        }
      }
      go(s1, ???).stream
    }
  }

  def pullAll[F[_], C <: Coproduct](
    stream: Stream[F, C]
  )(implicit ev: WaitFor[F, C]): Stream[F, ev.Out] = {
    stream.through(ev.apply)
  }
}

我的瓶颈在于实现override def apply: Pipe[F, C, Out]，思路是用(Option[C1], Option[C2], ...)这样的元组跟踪所有Coproduct成员，待所有元素均为Some时转换为HList输出，但在值层面无法处理元组的动态长度和构造，需要解决指引。

核心思路是用Shapeless HList作为状态容器（每个元素对应原HList成员的Option类型），通过类型类递归处理Coproduct的更新和状态完整性检查，结合FS2 Pull实现状态流转。

1. 定义状态相关的类型类 #

首先需要两个类型类：一个用于更新状态HList，另一个用于检查状态是否完整并转换为目标HList。

import shapeless._
import shapeless.ops.hlist._
import fs2._

// 类型类：将Coproduct的成员更新到状态HList（每个元素是Option）
trait UpdateState[C <: Coproduct, S <: HList] {
  def update(c: C, state: S): S
}

object UpdateState {
  // 处理CNil的边界情况（不会触发，因为Coproduct非空）
  implicit val cnilUpdate: UpdateState[CNil, HNil] = (_: CNil, _: HNil) => HNil

  // 递归处理Coproduct的每个成员
  implicit def coproductUpdate[Head, Tail <: Coproduct, SHead, STail <: HList](
    implicit
    tailUpdate: UpdateState[Tail, STail]
  ): UpdateState[Head :+: Tail, Option[Head] :: STail] = {
    case (Inl(head), Some(_) :: tail) => Some(head) :: tail // 覆盖已有值
    case (Inl(head), None :: tail) => Some(head) :: tail
    case (Inr(tailCoprod), headOpt :: tail) => headOpt :: tailUpdate.update(tailCoprod, tail)
  }
}

// 类型类：检查状态HList是否所有元素都是Some，转换为目标HList
trait CompleteState[S <: HList, H <: HList] {
  def complete(state: S): Option[H]
}

object CompleteState {
  // 处理HNil的边界情况
  implicit val hnilComplete: CompleteState[HNil, HNil] = (_: HNil) => Some(HNil)

  // 递归检查每个元素
  implicit def hlistComplete[Head, STail <: HList, HTail <: HList](
    implicit
    tailComplete: CompleteState[STail, HTail]
  ): CompleteState[Option[Head] :: STail, Head :: HTail] = {
    case (Some(head) :: tail) => tailComplete.complete(tail).map(head :: _)
    case (_ :: _) => None // 存在None元素，状态不完整
  }
}

2. 完善WaitFor类型类的实现 #

现在可以基于上述类型类实现状态跟踪和FS2 Pull逻辑：

trait WaitFor[F[_], C <: Coproduct] {
  type Out <: HList
  type State <: HList // 状态类型：每个元素是Option[Out的成员]

  def apply: Pipe[F, C, Out]
}

object WaitFor {
  type Aux[F[_], C <: Coproduct, Out0 <: HList, State0 <: HList] =
    WaitFor[F, C] { type Out = Out0; type State = State0 }

  // 为每个Coproduct生成对应的WaitFor实例
  implicit def make[F[_], C <: Coproduct, Out <: HList, State <: HList](
    implicit
    // 将Coproduct转换为对应的HList（获取输出类型）
    toHList: ToHList.Aux[C, Out],
    // 生成初始状态：所有元素为None的HList
    initialState: Fill[Out, None.type] { type Out = State },
    // 状态更新逻辑
    updateState: UpdateState[C, State],
    // 状态完整性检查与转换
    completeState: CompleteState[State, Out]
  ): Aux[F, C, Out, State] = new WaitFor.Aux[F, C, Out, State] {
    override type Out = Out
    override type State = State

    private val emptyState: State = initialState.apply(None)

    override def apply: Pipe[F, C, Out] = { stream =>
      def loop(s: Stream[F, C], currentState: State): Pull[F, Out, Unit] =
        s.pull.uncons1.flatMap {
          case Some((c, rest)) =>
            val newState = updateState.update(c, currentState)
            completeState.complete(newState) match {
              case Some(completedHList) =>
                // 输出完整的HList，重置状态继续收集
                Pull.output1(completedHList) >> loop(rest, emptyState)
              case None =>
                // 状态未完整，继续拉取
                loop(rest, newState)
            }
          case None =>
            Pull.done // 流结束，终止
        }

      loop(stream, emptyState).stream
    }
  }

  // 简化调用的pullAll方法
  def pullAll[F[_], C <: Coproduct](stream: Stream[F, C])(
    implicit ev: WaitFor[F, C]
  ): Stream[F, ev.Out] = stream.through(ev.apply)
}

3. 使用示例 #

假设我们定义了简单的case类：

case class A()
case class B()
case class C()

type MyCoprod = A :+: B :+: C :+: CNil
type MyHList = A :: B :: C :: HNil

import cats.effect.IO

val testStream: Stream[IO, MyHList] = Stream
  .emits[MyCoprod](List(Inl(A()), Inr(Inl(B())), Inr(Inr(Inl(C())))))
  .through(WaitFor.pullAll)

这个流会在收集到A、B、C各一个后输出对应的A()::B()::C()::HNil，如果后续还有元素，会继续收集下一组。

关键说明 #

• 状态类型对齐：通过Shapeless的类型推导，确保状态HList的结构和输出HList完全对应，每个元素是Option类型，解决了元组动态长度的问题。
• 递归类型类：UpdateState和CompleteState通过递归处理Coproduct和HList的每个成员，实现类型安全的状态更新和检查。
• FS2 Pull状态流转：用loop函数递归维护状态，每次拉取元素后更新状态，检查是否完成，完成则输出并重置状态，否则继续拉取。

内容的提问来源于stack exchange，提问作者eitaporra