Implement zip

Author: danslapman

src/main/scala/asyncstreams/AsyncStream.scala

@@ -75,6 +75,13 @@ class AsyncStream[F[+_]: Monad, +A](private[asyncstreams] val data: F[Step[A, As
   def foldMap[B](f: A => B)(implicit impl: ASImpl[F], mb: Monoid[B]): F[B] = {
     impl.collectLeft(this)(mb.empty)((b, a) => mb.combine(b, f(a)))
   }
+
+  def zip[B](sb: AsyncStream[F, B]): AsyncStream[F, (A, B)] = AsyncStream {
+    for {
+      stepA <- data
+      stepB <- sb.data
+    } yield Step((stepA.value, stepB.value), stepA.rest zip stepB.rest)
+  }
 }
 
 object AsyncStream {

src/test/scala/asyncstreams/stdFuture/AsyncStreamOperations.scala

@@ -122,4 +122,13 @@ class AsyncStreamOperations extends FunSuite with Matchers {
 
     await(res) shouldBe 465
   }
+
+  test("zip") {
+    val s1 = 1 ~:: 2 ~:: 3 ~:: AsyncStream.asyncNil[Future, Int]
+    val s2 = 4 ~:: 5 ~:: AsyncStream.asyncNil[Future, Int]
+
+    val res = s1 zip s2
+
+    await(res.to[List]) shouldBe (1, 4) :: (2, 5) :: Nil
+  }
 }