class BidirectionalTracerouteAnswerer extends Answerer {

  @VisibleForTesting
  static List<BidirectionalTrace> computeBidirectionalTraces(
      Set<Flow> flows, TracerouteEngine tracerouteEngine, boolean ignoreFilters) {
    SortedMap<Flow, List<TraceAndReverseFlow>> forwardTraces =
        tracerouteEngine.computeTracesAndReverseFlows(flows, ignoreFilters);

    Set<FlowAndSessions> reverseFlowsAndSessions =
        forwardTraces.values().stream()
            .flatMap(List::stream)
            .filter(tarf -> tarf.getReverseFlow() != null)
            .map(tarf -> new FlowAndSessions(tarf.getReverseFlow(), tarf.getNewFirewallSessions()))
            .collect(ImmutableSet.toImmutableSet());

    Map<FlowAndSessions, List<Trace>> reverseTraces =
        computeReverseTraces(tracerouteEngine, reverseFlowsAndSessions, ignoreFilters);

    List<BidirectionalTrace> result = new ArrayList<>();
    forwardTraces.forEach(
        (forwardFlow, forwardTraceAndReverseFlows) ->
            forwardTraceAndReverseFlows.forEach(
                forwardTraceAndReverseFlow -> {
                  Trace forwardTrace = forwardTraceAndReverseFlow.getTrace();
                  Flow reverseFlow = forwardTraceAndReverseFlow.getReverseFlow();
                  Set<FirewallSessionTraceInfo> newSessions =
                      forwardTraceAndReverseFlow.getNewFirewallSessions();
                  if (reverseFlow == null) {
                    result.add(
                        new BidirectionalTrace(forwardFlow, forwardTrace, newSessions, null, null));
                  } else {
                    FlowAndSessions fas = new FlowAndSessions(reverseFlow, newSessions);
                    reverseTraces.get(fas).stream()
                        .map(
                            reverseTrace ->
                                new BidirectionalTrace(
                                    forwardFlow,
                                    forwardTrace,
                                    newSessions,
                                    reverseFlow,
                                    reverseTrace))
                        .forEach(result::add);
                  }
                }));
    return result;
  }

  public  BidirectionalTracerouteAnswerer(Question question, IBatfish batfish);

}