class OperationProcessor extends MonitoredProcess {

  public Stream<InternalEvent> perform(Stream<InternalEvent> input) {
    Stream<InternalEvent> output = null;

    if (this.op instanceof EventOperation) {
      output = input.map(ievent -> {
        this.getRuntimeStat().start();
        try {
          InternalEvent i = ((EventOperation) op).perform(ievent);
          this.getSuccessCountStat().increment();
          return i;
        } catch (OperationException e) {
          this.getErrorCountStat().increment();
          logger.warn(e);
          return null;
        } finally {
          this.getRuntimeStat().stop();
        }
      });
    } else if (this.op instanceof MultiplexOperation) {
      /*
       * MultiplexOperations require the use of flatmap which allows a single stream item to produce
       * multiple results.
       */
      output = input.flatMap(ievent -> {
        this.getRuntimeStat().start();
        try {
          Stream<InternalEvent> s = ((MultiplexOperation) op).perform(ievent).stream();
          this.getSuccessCountStat().increment();
          return s;
        } catch (OperationException e) {
          this.getErrorCountStat().increment();
          logger.warn(e);
          return Stream.empty();
        } finally {
          this.getRuntimeStat().stop();
        }
      });
    } else if (this.op instanceof StreamOperation) {
      StreamOperation forkOp = (StreamOperation) this.op;
      output = forkOp.getOutputStream(input);
    } else if (this.op instanceof FilterOperation) {
      output = input.filter(ievent -> {
        try {
          return ((FilterOperation) this.op).test(ievent);
        } catch (OperationException e) {
          this.getErrorCountStat().increment();
          logger.warn(e);
          return false;
        } finally {
          this.getRuntimeStat().stop();
        }
      });
    } else {
      throw new OperationException("Invalid type of operation");
    }

    /*
     * Filter out events if an operation did something that resulted in a null event or payload.
     * This protects future operations from running on invalid data.
     */
    return output.filter(ievent -> {
      if (ievent == null) {
        logger.warn(op.getClass().getName() + " produced a null InternalEvent");
        return false;
      }
      if (ievent.getEventObj() == null) {
        logger.warn(op.getClass().getName() + " produced a null DeserializedEvent");
        return false;
      }
      if (ievent.getEventObj().getPayload() == null) {
        logger.warn(op.getClass().getName() + " produced a null DeserializedEvent payload");
        return false;
      }

      return true;
    });
  }

  public  OperationProcessor(OperationFactory operationFactory);

}