class ConditionalOperation implements StreamOperation {

  public Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input) {
    /*
     * outputStreams keeps track of the output Stream of each Condition.
     */
    List<Stream<InternalEvent>> outputStreams =
        new ArrayList<Stream<InternalEvent>>(this.conditionsAndProcs.size());

    /*
     * From a list of operation configurations in each condition construct queues and streams.
     */
    this.filtersAndQueues =
        new ArrayList<Pair<FilterOperation, Queue<InternalEvent>>>(this.conditionsAndProcs.size());
    for (Pair<FilterOperation, List<OperationProcessor>> filterAndProcs : this.conditionsAndProcs) {

      FilterOperation filter = filterAndProcs.getLeft();
      List<OperationProcessor> procs = filterAndProcs.getRight();

      /*
       * Construct a Queue for each conditional. This is the input to each Condition.
       */
      Queue<InternalEvent> queue =
          new Queue<InternalEvent>(new LinkedBlockingQueue<InternalEvent>(procs.size()));

      this.filtersAndQueues
          .add(new ImmutablePair<FilterOperation, Queue<InternalEvent>>(filter, queue));

      /*
       * Connect the condition's input Queue with operations. Each operation returns a stream with its
       * operation concatenated on.
       */
      Stream<InternalEvent> conditionInput = queue.jdkStream();
      for (OperationProcessor proc : procs) {
        conditionInput = proc.perform(conditionInput);
      }

      /*
       * Last input is the output.
       */
      outputStreams.add(conditionInput);
    }

    /*
     * Condition Consumer Threads
     * 
     * Combine each condition's output stream and write to the output Queue. When all data is consumed
     * the last condition closes the output Queue.
     */
    Queue<InternalEvent> outputQueue = new Queue<InternalEvent>(
        new LinkedBlockingQueue<InternalEvent>(this.conditionsAndProcs.size()));
    AtomicInteger lock = new AtomicInteger(outputStreams.size());

    outputStreams.forEach(stream -> {
      this.es.execute(new StreamToQueue(stream, outputQueue, lock));
    });

    /*
     * Consume input Stream in a thread and publish to each condition's Queue.
     */
    new Thread(new Runnable() {
      @Override
      public void run() {
        input.forEach(ievent -> {
          boolean matches = false;

          for (Pair<FilterOperation, Queue<InternalEvent>> filterAndQueue : filtersAndQueues) {
            FilterOperation filter = filterAndQueue.getLeft();

            /*
             * If event passes the filter offer event to queue.
             */
            if (filter.test(ievent)) {
              filterAndQueue.getRight().offer(ievent);
              matches = true;
              break;
            }
          }

          /*
           * Send to output queue if no case matches
           */
          if (!matches && !filterNonMatch) {
            outputQueue.offer(ievent);
          }
        });

        /*
         * Close queues when source queue is consumed.
         */
        for (Pair<FilterOperation, Queue<InternalEvent>> filterAndQueue : filtersAndQueues) {
          filterAndQueue.getRight().close();
        }
      }
    }).start();

    return outputQueue.jdkStream();
  }

  public  ConditionalOperation(
      List<Pair<FilterOperation, List<OperationProcessor>>> conditionsAndProcs,
      boolean filterNonMatch);

}