issue-2428 (typelevel#2429)

* issue-2428 * issue-2428 - simplify stepNotSmallerThanSize * issue-2428 - stepSmallerThanSize * issue-2428 - simplify stepSmallerThanSize * issue-2428 - tests * issue-2428 - test
doytsujin · Jun 4, 2021 · c365f27 · c365f27
1 parent ce8cda9
commit c365f27
Show file tree

Hide file tree

Showing 2 changed files with 43 additions and 13 deletions.
diff --git a/core/shared/src/main/scala/fs2/Stream.scala b/core/shared/src/main/scala/fs2/Stream.scala
@@ -2396,14 +2396,13 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
             else Pull.output1(prev.take(size))
           case Some((hd, tl)) =>
             val buffer = ArrayBuffer.empty[Chunk[O]]
-            var (heads, tails) = (prev ++ hd).splitAt(step)
-            while (tails.nonEmpty) {
-              buffer += heads.take(size)
-              val (nHeads, nTails) = tails.splitAt(step)
-              heads = nHeads
-              tails = nTails
+            var current = prev ++ hd
+            while (current.size >= step) {
+              val (nHeads, nTails) = current.splitAt(step)
+              buffer += nHeads.take(size)
+              current = nTails
             }
-            Pull.output(Chunk.buffer(buffer)) >> stepNotSmallerThanSize(tl, heads)
+            Pull.output(Chunk.buffer(buffer)) >> stepNotSmallerThanSize(tl, current)
         }
 
     def stepSmallerThanSize(
@@ -2419,16 +2418,16 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
           case Some((hd, tl)) =>
             val buffer = ArrayBuffer.empty[Chunk[O]]
             var w = window
-            var (heads, tails) = (prev ++ hd).splitAt(step)
-            while (tails.nonEmpty) {
-              val wind = w ++ heads.take(step)
+            var current = prev ++ hd
+            while (current.size >= step) {
+              val (head, tail) = current.splitAt(step)
+              val wind = w ++ head
               buffer += wind
               w = wind.drop(step)
-              heads = tails.take(step)
-              tails = tails.drop(step)
+              current = tail
             }
 
-            Pull.output(Chunk.buffer(buffer)) >> stepSmallerThanSize(tl, w, heads)
+            Pull.output(Chunk.buffer(buffer)) >> stepSmallerThanSize(tl, w, current)
         }
 
     val resultPull =

diff --git a/core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala b/core/shared/src/test/scala/fs2/StreamCombinatorsSuite.scala
@@ -25,13 +25,15 @@ import scala.concurrent.duration._
 import scala.concurrent.TimeoutException
 import cats.effect.{IO, SyncIO}
 import cats.effect.kernel.Ref
+import cats.effect.std.Queue
 import cats.effect.std.Semaphore
 import cats.syntax.all._
 import org.scalacheck.Gen
 import org.scalacheck.Prop.forAll
 import org.scalacheck.effect.PropF.forAllF
 
 import fs2.concurrent.SignallingRef
+import cats.effect.kernel.Deferred
 
 class StreamCombinatorsSuite extends Fs2Suite {
 
@@ -1305,6 +1307,35 @@ class StreamCombinatorsSuite extends Fs2Suite {
     }
   }
 
+  test("sliding shouldn't swallow last issue-2428") {
+    forAllF { (n0: Int, n1: Int, n2: Int) =>
+      val streamSize = (n0 % 1000).abs + 1
+      val size = (n1 % 20).abs + 1
+      val step = (n2 % 20).abs + 1
+
+      val action =
+        Vector.fill(streamSize)(Deferred[IO, Unit]).sequence.map { seenArr =>
+          def peek(ind: Int)(f: Option[Unit] => Boolean) =
+            seenArr.get(ind).fold(true.pure[IO])(_.tryGet.map(f))
+
+          Stream
+            .emits(0 until streamSize)
+            .unchunk
+            .evalTap(seenArr(_).complete(()))
+            .sliding(size, step)
+            .evalMap { chunk =>
+              val next = chunk.head.get + size + step - 1
+              val viewed = chunk.map(i => peek(i)(_.nonEmpty))
+              val notViewed = Chunk.singleton(peek(next)(_.isEmpty))
+              (notViewed ++ viewed).sequence
+            }
+            .flatMap(Stream.chunk)
+        }
+
+      Stream.force(action).compile.fold(true)(_ && _).assert
+    }
+  }
+
   group("split") {
     property("1") {
       forAll { (s: Stream[Pure, Int], n0: Int) =>