FS2 - How to route an element to a specific nested stream/pipe?

Question

I want to run N nested streams/pipes in parallel and send each element to only one of the nested streams. Balance allows me to do this but I want to route elements with the same "key" to the same nested stream or pipe.

I can't see any functions to do this so I wrote a basic POC which broadcasts each element to every stream. The stream/pipe then filters only the elements it should handle (see below). This seems quite inefficient, is there a better way to route elements to specific nested streams?

package io.xxx.streams

import cats.effect.{ExitCode, IO, IOApp}
import fs2.{Pipe, Stream}

object StreamsApp extends IOApp {

  import cats.syntax.functor._
  import scala.concurrent.duration._

  case class StreamMessage(routingKey: Int, value: String)

  // filter elements which belong to the given bin
  def filterAndLog(bin: Int, numBins: Int): IO[Pipe[IO, StreamMessage, Unit]] = IO {
    val predicate = (m: StreamMessage) => m.routingKey % numBins == bin

    in: Stream[IO, StreamMessage] => {
      in.filter(predicate).evalMap(m => IO {
        println(s"bin $bin - ${m.value}")
      })
    }
  }

  override def run(args: List[String]): IO[ExitCode] = {
    val effectsStream = for {
      pipeOne <- Stream.eval(filterAndLog(0, 2))
      pipeTwo <- Stream.eval(filterAndLog(1, 2))
      s <- Stream
        .fixedDelay[IO](100.millis)
        .zipRight(Stream.range(0, 50))
        .map(i => StreamMessage(i, s"message $i"))
        .broadcastThrough(pipeOne, pipeTwo)
    } yield s

    effectsStream.compile.drain.as(ExitCode(0))
  }

}

Messages with the same routing key should be handled by the same stream/pipe