@Override
  public RecordStreamWithMetadata<DI, SI> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState) throws StreamProcessingException {
    init(workUnitState);

    setInputGlobalMetadata(inputStream.getGlobalMetadata(), workUnitState);

    Flowable<StreamEntity<DI>> outputStream =
        inputStream.getRecordStream()
            .flatMap(in -> {
              if (in instanceof ControlMessage) {
                if (in instanceof MetadataUpdateControlMessage) {
                  setInputGlobalMetadata(((MetadataUpdateControlMessage) in).getGlobalMetadata(), workUnitState);
                }

                getMessageHandler().handleMessage((ControlMessage) in);
                return Flowable.just(in);
              } else if (in instanceof RecordEnvelope) {
                RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in;
                Iterable<ControlMessage<DI>> injectedBeforeIterable =
                    injectControlMessagesBefore(recordEnvelope, workUnitState);
                Iterable<ControlMessage<DI>> injectedAfterIterable =
                    injectControlMessagesAfter(recordEnvelope, workUnitState);

                if (injectedBeforeIterable == null && injectedAfterIterable == null) {
                  // nothing injected so return the record envelope
                  return Flowable.just(recordEnvelope);
                } else {
                  Flowable<StreamEntity<DI>> flowable;

                  if (injectedBeforeIterable != null) {
                    flowable = Flowable.<StreamEntity<DI>>fromIterable(injectedBeforeIterable)
                        .concatWith(Flowable.just(recordEnvelope));
                  } else {
                    flowable = Flowable.just(recordEnvelope);
                  }

                  if (injectedAfterIterable != null) {
                    flowable.concatWith(Flowable.fromIterable(injectedAfterIterable));
                  }
                  return flowable;
                }
              } else {
                throw new UnsupportedOperationException();
              }
            }, 1);
    outputStream = outputStream.doOnComplete(this::close);
    return inputStream.withRecordStream(outputStream, inputStream.getGlobalMetadata());
  }
@ Override public RecordStreamWithMetadata < DI , SI > processStream ( RecordStreamWithMetadata < DI , SI > inputStream , WorkUnitState workUnitState ) throws StreamProcessingException { init ( workUnitState ) ; setInputGlobalMetadata ( inputStream . getGlobalMetadata ( ) , workUnitState ) ; Flowable < StreamEntity < DI > > outputStream = inputStream . getRecordStream ( ) . flatMap ( in -> { if ( in instanceof ControlMessage ) { if ( in instanceof MetadataUpdateControlMessage ) { setInputGlobalMetadata ( ( ( MetadataUpdateControlMessage ) in ) . getGlobalMetadata ( ) , workUnitState ) ; } getMessageHandler ( ) . handleMessage ( ( ControlMessage ) in ) ; return Flowable . just ( in ) ; } else if ( in instanceof RecordEnvelope ) { RecordEnvelope < DI > recordEnvelope = ( RecordEnvelope < DI > ) in ; Iterable < ControlMessage < DI > > injectedBeforeIterable = injectControlMessagesBefore ( recordEnvelope , workUnitState ) ; Iterable < ControlMessage < DI > > injectedAfterIterable = injectControlMessagesAfter ( recordEnvelope , workUnitState ) ; if ( injectedBeforeIterable == null && injectedAfterIterable == null ) { return Flowable . just ( recordEnvelope ) ; } else { Flowable < StreamEntity < DI > > flowable ; if ( injectedBeforeIterable != null ) { flowable = Flowable . < StreamEntity < DI > > fromIterable ( injectedBeforeIterable ) . concatWith ( Flowable . just ( recordEnvelope ) ) ; } else { flowable = Flowable . just ( recordEnvelope ) ; } if ( injectedAfterIterable != null ) { flowable . concatWith ( Flowable . fromIterable ( injectedAfterIterable ) ) ; } return flowable ; } } else { throw new UnsupportedOperationException ( ) ; } } , 1 ) ; outputStream = outputStream . doOnComplete ( this :: close ) ; return inputStream . withRecordStream ( outputStream , inputStream . getGlobalMetadata ( ) ) ; }
Apply injections to the input {@link RecordStreamWithMetadata}.
{@link ControlMessage}s may be injected before, after, or around the input record.
A {@link MetadataUpdateControlMessage} will update the current input {@link GlobalMetadata} and pass the
updated input {@link GlobalMetadata} to the next processor to propagate the metadata update down the pipeline.