FS2 Running streams in sequence

Question

I have a fairly simple use case. I have two web service calls one fetches products and another fetches relationship. I want to run fetchProducts() first extract a field from the set of products and then pass the output to fetchRelationships(ids: Seq[String]) so that I can then set the relationship back on the product. Here's the code:

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])

I am constrained by the external Api to get the results in batches. So I'm not sure how to express this using fs2 or if I should be using it at all.

Answer 1

Unfortunately you code in the question doesn't match your text description and misses quite a few important bits (such as whole Relationship class). Also it is not clear what

I am constrained by the external Api to get the results in batches

really means. Also it is not clear why all fields in Product including productId are Option.

The following code compiles and might or might not be what you need:

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
}