Merge pull request #2813 from SystemFw/pauseWhen/delays

Do not withhold emission in `pauseWhen`

core/shared/src/main/scala/fs2/Stream.scala

@@ -2196,12 +2196,16 @@ final class Stream[+F[_], +O] private[fs2] (private[fs2] val underlying: Pull[F,
     def waitToResume =
       pauseWhenTrue.discrete.dropWhile(_ == true).take(1).compile.drain
 
+    // The logic can be expressed entirely with `waitToResume`, but
+    // `Signal.get` is lighter than `Signal.discrete`, so the preliminary
+    // check with `get` in `pauseIfNeeded` acts as an optimisation, since
+    // we expect a stream to generally not be in a paused state.
     def pauseIfNeeded = Stream.exec {
       pauseWhenTrue.get.flatMap(paused => waitToResume.whenA(paused))
     }
 
     pauseIfNeeded ++ chunks.flatMap { chunk =>
-      pauseIfNeeded ++ Stream.chunk(chunk)
+      Stream.chunk(chunk) ++ pauseIfNeeded
     }
   }