Improve `parJoin` with the ability to select a join strategy

The current implementation of parJoin with maxOpen does not start the remaining streams until the earlier ones are complete. This is a problem in cases where the inner streams are infinite as in the example below:

Listing A - parJoin with 10 infinite streams, maxOpen = 2

    Stream.range(1, 10)
      .map {
        c =>
          Stream
            .repeatEval(IO(c))
      }
      .parJoin(maxOpen = 2)
      .evalTap {
        c =>
          IO(println(s"I am Stream $c"))
      }
      .take(10)
      .compile
      .drain

Output of Listing A

I am Stream 1
I am Stream 2
I am Stream 2
I am Stream 1
I am Stream 1
I am Stream 2
I am Stream 2
I am Stream 1
I am Stream 1
I am Stream 2

This limitation makes parJoin unusable with libraries like fs2-kafka when processing individual partitions is desirable before joining them. The only alternative in that case is to use parJoinUnbounded which fixes the above problem but has no way of controlling how many inner streams will be evaluated in parallel. This is desirable if the inner streams are memory intensive and you cannot let all of them run in parallel.

I would like to propose a new signature for parJoin as follows:

sealed trait JoinStrategy

object JoinStrategy {
  case object Random extends JoinStrategy
  case object RoundRobin extends JoinStrategy
  case object Sequential extends JoinStrategy
}

def parJoin(maxOpen: Int, joinStrategy: JoinStrategy = Sequential)

Sequential would be the current logic and the default
RoundRobin would evaluate each inner stream in turn
Random would be how the parJoinUnbounded works but with maximum parallel evaluations limited to maxOpen

Surly there can be better names selected for these and there could be other strategies but these are what I can think of now.

Start by locating the parJoin and parJoinUnbounded entry points and compare their current scheduling behavior. Define how Sequential, RoundRobin, and Random should behave under maxOpen, including the default, then verify the selected strategy works for finite and infinite inner streams like the examples.】【。