ordered mapAsync for fs2

Fs2MapAsync.scala

import cats.implicits._
import cats.effect.Effect
import fs2.{Pipe, Stream}
import fs2.async.Promise
import fs2.async.mutable.Queue

import scala.concurrent.ExecutionContext

def mapAsync[F[_]: Effect, A, B](parallelism: Int)(f: A => F[B])(implicit executionContext: ExecutionContext): Pipe[F, A, B] = { stream =>
  Stream.eval(Queue.bounded[F, Option[F[Either[Throwable, B]]]](parallelism)).flatMap { q =>
    q.dequeueAvailable
      .unNoneTerminate
      .evalMap(identity)
      .rethrow
      .concurrently {
        stream.evalMap { a =>
          Promise.empty[F, Either[Throwable, B]].flatMap { promise =>
            val sb = Stream.eval(f(a).attempt).evalMap(promise.complete)
            q.enqueue1(Some(promise.get)).as(sb)
          }
        }.join(parallelism).drain ++ Stream.eval(q.enqueue1(None))
      }
  }
}