core/play/src/main/java/play/libs/streams/AkkaStreams.java
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());
                })));
  }
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());
                })));
  }
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.

<p>If the splitter flow produces Left, they will be fed into the flow. If it produces Right,
they will bypass the flow.

@param <In> the In type parameter for Flow
@param <FlowIn> the FlowIn type parameter for the left branch in Either.
@param <Out> the Out type parameter for Flow.
@param flow the original flow.
@param splitter the splitter function.
@param mergeStrategy the merge strategy (onlyFirstCanFinishMerge, ignoreAfterFinish,
ignoreAfterCancellation)
@return the flow with a bypass.
Using the given splitter flow allow messages to bypass a flow .