batch.scala

  def batch[F[_]: Concurrent, T, U](f: T => U, bound: Int)(implicit U: Monoid[U]): Pipe[F, T, U] =
    input =>
      Stream.eval(Queue.bounded[F, U](1)).flatMap { queue =>
        val combiner = input.evalMapAccumulate((U.empty, 0)) {
          case ((acc, currBound), t) =>
            val u = f(t)

            queue.offer1(u).flatMap { stored =>
              if (stored) ((U.empty, 0), ()).pure[F]
              else {
                if (currBound < bound) ((U.combine(acc, u), currBound + 1), ()).pure[F]
                else queue.enqueue1(acc) >> ((u, 1), ()).pure[F]
              }
            }
        }

        combiner.drain.concurrently(queue.dequeue)
    }