groupedWithin emits empty groups

[Sebruck]

Hi

I use fs2 version 1.0.2 and the groupedWithin function seems not to behave as documented.

/**
    * Divide this streams into groups of elements received within a time window,
    * or limited by the number of the elements, whichever happens first.
    * Empty groups, which can occur if no elements can be pulled from upstream
    * in a given time window, **will not be emitted.**
    *
    * Note: a time window starts each time downstream pulls.
    */
  def groupWithin[F2[x] >: F[x]](n: Int, d: FiniteDuration)(

The documentation says that empty groups will not be emitted. But when numberOfElements % n == 0 then the last group is still empty.

Reproducer:

import cats.effect.{ContextShift, IO, Timer}
import fs2.Stream

import scala.concurrent.duration._

object Fs2Test extends App {

  implicit val cs: ContextShift[IO] = IO.contextShift(concurrent.ExecutionContext.global)
  implicit val timer: Timer[IO]     = IO.timer(concurrent.ExecutionContext.global)

  val numberStrings = (1 to 30).map(_.toString).toList
  Stream
    .emits(numberStrings)
    .covary[IO]
    .groupWithin(30, 1.milli)
    .map(chunk => println(chunk.size))
    .compile
    .drain
    .unsafeRunSync()

  // Output is:
  // 30
  // 0
}

Is this a bug or is the documentation miss-leading?

[Sebruck]

I am currently looking into this.

[SystemFw]

Nice, this specific issue should be in the algorithmic (non concurrent) part of the code so hopefully not too bad

[yannick-cw]

One thing I was wondering, when the groupedWithin always emits nonempty batches, why not make it something like a NonEmptyList?

[SystemFw]

Mostly because Chunk is optimised and very often you want to keep working with it. NonEmptyList forces a very specific access pattern. So, it's a tradeoff :) (we don't have NonEmptyChunk anymore)