class ForkOperation implements StreamOperation {

  public Stream<InternalEvent> getOutputStream(Stream<InternalEvent> input) {
    /*
     * forkOutputStreams keeps track of the output Stream of each fork.
     */
    List<Stream<InternalEvent>> forkOutputStreams =
        new ArrayList<Stream<InternalEvent>>(opProcsInForks.size());

    /*
     * From a list of operation configurations in each fork construct queues and streams.
     */
    this.queues = new ArrayList<Queue<InternalEvent>>(opProcsInForks.size());
    for (List<OperationProcessor> opProcsInFork : opProcsInForks) {
      /*
       * Construct a Queue for each fork. This is the input to each Fork.
       */
      Queue<InternalEvent> queue =
          new Queue<InternalEvent>(new LinkedBlockingQueue<InternalEvent>(opProcsInFork.size()));
      this.queues.add(queue);

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

      /*
       * Last input is the output.
       */
      forkOutputStreams.add(forkInput);
    }

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

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

    /*
     * Consume input Stream in a thread and publish to each fork's Queue.
     */
    new Thread(new Runnable() {
      @Override
      public void run() {
        input.forEach(ievent -> {
          queues.forEach(queue -> {
            /*
             * The original event is NOT sent to each fork. Rather a copy of the event is sent to
             * each fork. This ensures that there is no contention between the operations performed
             * on each event. Caveat is that when the forks join there will be two events produced.
             */
            queue.offer(ievent.copy());
          });
        });

        for (Queue<InternalEvent> queue : queues) {
          queue.close();
        }
      }
    }).start();

    return outputQueue.jdkStream();
  }

  public  ForkOperation(List<List<OperationProcessor>> opProcsInForks);

}