BypassingMerge.scala

import akka.stream.scaladsl.FlexiMerge.{ReadAny, MergeLogic}
import akka.stream.{Attributes, FanInShape2}
import akka.stream.scaladsl.FlexiMerge

class BypassingMerge[A, B] extends FlexiMerge[A, FanInShape2[A, B, A]](new FanInShape2("BMerge"), Attributes.name("BMerge")) {

  def createMergeLogic(p: PortT): MergeLogic[A] = new MergeLogic[A] {
    override def initialState =
      State[Any](ReadAny(p)) { (ctx, input, element) =>
        if (input eq p.in0)
          ctx.emit(element.asInstanceOf[A])
        SameState
      }
  }
}

object BypassingMerge {
  def apply[A,B]() = new BypassingMerge[A,B]
}

@rkuhn it has generic parameters A and B ... ?