FS2 按顺序运行流

Question

我有一个相当简单的用例。我有两个 Web 服务调用，一个获取产品，另一个获取关系。我想运行 fetchProducts(( 首先从产品集中提取一个字段，然后将输出传递给 fetchRelations(ids： Seq[String](，以便我可以在产品上重新设置关系。代码如下：

def fetchProducts(): Stream[IO, Seq[Product]]= {
 //webservice call
}
def fetchRelationship(ids: Seq[Product]): Stream[IO, Seq[Relationship]] = {
 //webservice call
}
//Pseudocode. How can I do this with fs2 Streams?
def process = {
      val prods = fetchProducts() //run this first
      val prodIds = prods.flatMap(i => i.productId)
      val rels = fetchRelationships(prodIds) //run after all all products are fetched 
      prods.forEach(p => p.setRelation(rels.get(p.id))
    }
}
 case class Product(productId: Option[String],
                        name: Option[String],
                        description: Option[String],
                        brandName: Option[String])

我受到外部 API 的限制，无法批量获取结果。所以我不确定如何使用 fs2 表达这一点，或者我是否应该使用它。

Answer 1

不幸的是，您在问题中的代码与您的文本描述不匹配，并且错过了很多重要的部分(例如整个Relationship类(。也不清楚是什么

我受到外部 API 的限制，无法批量获取结果

真的是这个意思。也不清楚为什么Product中的所有字段(包括productId(都是Option。

以下代码将编译，可能是您所需要的，也可能不是您需要的：

case class Product(productId: Option[String],
                   name: Option[String],
                   description: Option[String],
                   brandName: Option[String],
                   relationships: mutable.ListBuffer[Relationship]) {
}
case class Relationship(productId: String, someInfo: String)
def fetchProducts(): Stream[IO, Seq[Product]] = {
  //webservice call
  ???
}
//    def fetchRelationships(ids: Seq[Product]): Stream[IO, Seq[Relationship]] = {
def fetchRelationships(ids: Seq[String]): Stream[IO, Seq[Relationship]] = {
  //webservice call
  ???
}
def process():  = {
  val prods = fetchProducts() //run this first
  val prodsAndRels: Stream[IO, (Seq[Product], Seq[Relationship])] = prods.flatMap(ps => fetchRelationships(ps.map(p => p.productId.get)).map(rs => (ps, rs)))
  val prodsWithFilledRels: Stream[IO, immutable.Seq[Product]] = prodsAndRels.map({ case (ps, rs) => {
    val productsMap = ps.map(p => (p.productId.get, p)).toMap
    rs.foreach(rel => productsMap(rel.productId).relationships += rel)
    ps.toList
  }
  })
  prodsWithFilledRels
}