Issue with defining custom partition stage (Cannot pull in port twice)

Question

So I have this little Custom Stage for partitioning in Akka Streams.

object CustomPartitioner {

    /**
     * Creates a Partition stage that, given a type A, makes a decision to whether to partition to subtype B or subtype C
     *
     * @param partitionF applies function, if true, route to B, otherwise route to C.
     *
     * @tparam A type of input
     * @tparam B type of output on the first outlet.
     * @tparam C type of output on the second outlet.
     *
     * @return A partition stage
     */
    def apply[A, B, C](partitionF: A => Either[B, C]) =
      new GraphStage[FanOutShape2[A, B, C]] {
        private val in: Inlet[A] = Inlet[A]("in")
        private val outB = Outlet[B]("outB")
        private val outC = Outlet[C]("outC")
        private val pendingB = MutableQueue.empty[B]
        private val pendingC = MutableQueue.empty[C]

        override def shape: FanOutShape2[A, B, C] = new FanOutShape2(in, outB, outC)
        override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
      new GraphStageLogic(shape) with InHandler with OutHandler {

        setHandler(in, this)
        setHandler(outB, this)
        setHandler(outC, this)

        override def onPush(): Unit = {
          val elem = grab(in)
          partitionF(elem) match {
            case Left(b) =>
              pendingB.enqueue(b)
              tryPush(outB, pendingB, b)
            case Right(c) =>
              pendingC.enqueue(c)
              tryPush(outC, pendingC, c)
          }
        }

        override def onPull(): Unit = pull(in)

        private def tryPush[T](out: Outlet[T], pending: MutableQueue[T]): Unit =
          if (isAvailable(out) && pending.nonEmpty) push(out, pending.dequeue())
      }
  }

I have hooked this as a partitioner into a flow and then merged it back into a sink.

When I try to push a message through the stream using a component test

java.lang.IllegalArgumentException: Cannot pull port (in(256390569)) twice

and then the test fails with

java.lang.AssertionError: assertion failed: expected: expecting request() signal but got unexpected message CancelSubscription(PublisherProbeSubscription(akka.stream.impl.fusing.ActorGraphInterpreter$BatchingActorInputBoundary$$anon$1@53c99b09,akka.testkit.TestProbe@2539cd1c))

I am pretty certain I am messing up the setHandler calls, since there are two of them to handle both outB and outC. However I do not know how to fix it, to make this entire system only call onPush and onPull exactly once.

Answer 1

I managed to get it to work by

override def onPull(): Unit =
   if (!hasBeenPulled(in))
       pull(in)