Add Align for Stream

core/src/main/scala-2.12/cats/instances/stream.scala

@@ -1,16 +1,16 @@
 package cats
 package instances
 
-import cats.data.ZipStream
+import cats.data.{Ior, ZipStream}
 import cats.syntax.show._
 
 import scala.annotation.tailrec
 
 trait StreamInstances extends cats.kernel.instances.StreamInstances {
 
   implicit val catsStdInstancesForStream
-    : Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] =
-    new Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] {
+    : Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] with Align[Stream] =
+    new Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] with Align[Stream] {
 
       def empty[A]: Stream[A] = Stream.Empty
 
@@ -151,6 +151,14 @@ trait StreamInstances extends cats.kernel.instances.StreamInstances {
 
       override def collectFirstSome[A, B](fa: Stream[A])(f: A => Option[B]): Option[B] =
         fa.collectFirst(Function.unlift(f))
+
+      def functor: Functor[Stream] = this
+
+      def align[A, B](fa: Stream[A], fb: Stream[B]): Stream[Ior[A, B]] =
+        alignWith(fa, fb)(identity)
+
+      override def alignWith[A, B, C](fa: Stream[A], fb: Stream[B])(f: Ior[A, B] => C): Stream[C] =
+        Align.alignWithIterator[A, B, C](fa, fb)(f).toStream
     }
 
   implicit def catsStdShowForStream[A: Show]: Show[Stream[A]] =

core/src/main/scala-2.13+/cats/instances/lazyList.scala

@@ -135,22 +135,8 @@ trait LazyListInstances extends cats.kernel.instances.LazyListInstances {
       def align[A, B](fa: LazyList[A], fb: LazyList[B]): LazyList[Ior[A, B]] =
         alignWith(fa, fb)(identity)
 
-      override def alignWith[A, B, C](fa: LazyList[A], fb: LazyList[B])(f: Ior[A, B] => C): LazyList[C] = {
-
-        val alignIterator = new Iterator[C] {
-          val iterA = fa.iterator
-          val iterB = fb.iterator
-          def hasNext: Boolean = iterA.hasNext || iterB.hasNext
-          def next(): C =
-            f(
-              if (iterA.hasNext && iterB.hasNext) Ior.both(iterA.next(), iterB.next())
-              else if (iterA.hasNext) Ior.left(iterA.next())
-              else Ior.right(iterB.next())
-            )
-        }
-
-        LazyList.from(alignIterator)
-      }
+      override def alignWith[A, B, C](fa: LazyList[A], fb: LazyList[B])(f: Ior[A, B] => C): LazyList[C] =
+        LazyList.from(Align.alignWithIterator[A, B, C](fa, fb)(f))
     }
 
   implicit def catsStdShowForLazyList[A: Show]: Show[LazyList[A]] =

core/src/main/scala-2.13+/cats/instances/stream.scala

@@ -1,7 +1,7 @@
 package cats
 package instances
 
-import cats.data.ZipStream
+import cats.data.{Ior, ZipStream}
 import cats.syntax.show._
 
 import scala.annotation.tailrec
@@ -10,8 +10,8 @@ trait StreamInstances extends cats.kernel.instances.StreamInstances {
 
   @deprecated("Use cats.instances.lazyList", "2.0.0-RC2")
   implicit val catsStdInstancesForStream
-    : Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] =
-    new Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] {
+    : Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] with Align[Stream] =
+    new Traverse[Stream] with Alternative[Stream] with Monad[Stream] with CoflatMap[Stream] with Align[Stream] {
 
       def empty[A]: Stream[A] = Stream.Empty
 
@@ -152,6 +152,14 @@ trait StreamInstances extends cats.kernel.instances.StreamInstances {
 
       override def collectFirstSome[A, B](fa: Stream[A])(f: A => Option[B]): Option[B] =
         fa.collectFirst(Function.unlift(f))
+
+      def functor: Functor[Stream] = this
+
+      def align[A, B](fa: Stream[A], fb: Stream[B]): Stream[Ior[A, B]] =
+        alignWith(fa, fb)(identity)
+
+      override def alignWith[A, B, C](fa: Stream[A], fb: Stream[B])(f: Ior[A, B] => C): Stream[C] =
+        Stream.from(Align.alignWithIterator[A, B, C](fa, fb)(f))
     }
 
   @deprecated("Use cats.instances.lazyList", "2.0.0-RC2")

core/src/main/scala/cats/Align.scala

@@ -87,4 +87,17 @@ object Align {
   def semigroup[F[_], A](implicit F: Align[F], A: Semigroup[A]): Semigroup[F[A]] = new Semigroup[F[A]] {
     def combine(x: F[A], y: F[A]): F[A] = Align[F].alignCombine(x, y)
   }
+
+  private[cats] def alignWithIterator[A, B, C](fa: Iterable[A], fb: Iterable[B])(f: Ior[A, B] => C): Iterator[C] =
+    new Iterator[C] {
+      private[this] val iterA = fa.iterator
+      private[this] val iterB = fb.iterator
+      def hasNext: Boolean = iterA.hasNext || iterB.hasNext
+      def next(): C =
+        f(
+          if (iterA.hasNext && iterB.hasNext) Ior.both(iterA.next(), iterB.next())
+          else if (iterA.hasNext) Ior.left(iterA.next())
+          else Ior.right(iterB.next())
+        )
+    }
 }

tests/src/test/scala/cats/tests/StreamSuite.scala

@@ -2,6 +2,7 @@ package cats
 package tests
 
 import cats.laws.discipline.{
+  AlignTests,
   AlternativeTests,
   CoflatMapTests,
   CommutativeApplyTests,
@@ -34,6 +35,9 @@ class StreamSuite extends CatsSuite {
   checkAll("Stream[Int]", TraverseFilterTests[Stream].traverseFilter[Int, Int, Int])
   checkAll("TraverseFilter[Stream]", SerializableTests.serializable(TraverseFilter[Stream]))
 
+  checkAll("Stream[Int]", AlignTests[Stream].align[Int, Int, Int, Int])
+  checkAll("Align[Stream]", SerializableTests.serializable(Align[Stream]))
+
   // Can't test applicative laws as they don't terminate
   checkAll("ZipStream[Int]", CommutativeApplyTests[ZipStream].apply[Int, Int, Int])