public static < In , FlowIn , Out > Flow < In , Out , ? > bypassWith ( Flow < In , F . Either < FlowIn , Out > , ? > splitter , Graph < UniformFanInShape < Out , Out > , ? > mergeStrategy , Flow < FlowIn , Out , ? > flow ) { return splitter . via ( Flow . fromGraph ( GraphDSL . < FlowShape < F . Either < FlowIn , Out > , Out > > create ( builder -> { // Eager cancel must be true so that if the flow cancels, that will be propagated // upstream. // However, that means the bypasser must block cancel, since when this flow // finishes, the merge // will result in a cancel flowing up through the bypasser, which could lead to // dropped messages. // Using scaladsl here because of https://github.com/akka/akka/issues/18384 UniformFanOutShape < F . Either < FlowIn , Out > , F . Either < FlowIn , Out > > broadcast = builder . add ( Broadcast . create ( 2 , true ) ) ; UniformFanInShape < Out , Out > merge = builder . add ( mergeStrategy ) ; Flow < F . Either < FlowIn , Out > , FlowIn , ? > collectIn = Flow . < F . Either < FlowIn , Out > > create ( ) . collect ( Scala . partialFunction ( x -> { if ( x . left . isPresent ( ) ) { return x . left . get ( ) ; } else { throw Scala . noMatch ( ) ; } } ) ) ; Flow < F . Either < FlowIn , Out > , Out , ? > collectOut = Flow . < F . Either < FlowIn , Out > > create ( ) . collect ( Scala . partialFunction ( x -> { if ( x . right . isPresent ( ) ) { return x . right . get ( ) ; } else { throw Scala . noMatch ( ) ; } } ) ) ; Flow < F . Either < FlowIn , Out > , F . Either < FlowIn , Out > , ? > blockCancel = play . api . libs . streams . AkkaStreams . < F . Either < FlowIn , Out > > ignoreAfterCancellation ( ) . asJava ( ) ; // Normal flow builder . from ( broadcast . out ( 0 ) ) . via ( builder . add ( collectIn ) ) . via ( builder . add ( flow ) ) . toInlet ( merge . in ( 0 ) ) ; // Bypass flow, need to ignore downstream finish builder . from ( broadcast . out ( 1 ) ) . via ( builder . add ( blockCancel ) ) . via ( builder . add ( collectOut ) ) . toInlet ( merge . in ( 1 ) ) ; return new FlowShape <> ( broadcast . in ( ) , merge . out ( ) ) ; } ) ) ) ; }
Using the given splitter flow allow messages to bypass a flow .