@Test public void testMixedStream() throws Exception { MyConverter converter = new MyConverter(); BasicAckableForTesting ackable = new BasicAckableForTesting(); RecordStreamWithMetadata<Integer, String> stream = new RecordStreamWithMetadata<>(Flowable.just(new RecordEnvelope<>(1), new MyControlMessage<>()), GlobalMetadata.<String>builder().schema("schema").build()).mapRecords(r -> { r.addCallBack(ackable); return r; }); List<StreamEntity<Integer>> outputRecords = Lists.newArrayList(); converter.processStream(stream, new WorkUnitState()).getRecordStream().subscribe(outputRecords::add); Assert.assertEquals(outputRecords.size(), 2); Assert.assertEquals(((RecordEnvelope<Integer>) outputRecords.get(0)).getRecord(), new Integer(0)); Assert.assertTrue(outputRecords.get(1) instanceof MyControlMessage); }
@Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { init(workUnitState); this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build(); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMap(in -> { if (in instanceof ControlMessage) { ControlMessage out = (ControlMessage) in; getMessageHandler().handleMessage((ControlMessage) in); if (in instanceof MetadataUpdateControlMessage) { this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput( ((MetadataUpdateControlMessage) in).getGlobalMetadata(), Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata() .getSchema(), workUnitState))).build(); out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata); } return Flowable.just(((ControlMessage<DO>) out)); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState); } else { throw new UnsupportedOperationException(); } }, 1); outputStream = outputStream.doOnComplete(this::close); return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata); }
Converter implements Closeable, FinalState, RecordStreamProcessor<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { init(workUnitState); this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build(); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMap(in -> { if (in instanceof ControlMessage) { ControlMessage out = (ControlMessage) in; getMessageHandler().handleMessage((ControlMessage) in); if (in instanceof MetadataUpdateControlMessage) { this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput( ((MetadataUpdateControlMessage) in).getGlobalMetadata(), Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata() .getSchema(), workUnitState))).build(); out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata); } return Flowable.just(((ControlMessage<DO>) out)); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState); } else { throw new UnsupportedOperationException(); } }, 1); outputStream = outputStream.doOnComplete(this::close); return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata); } }
Converter implements Closeable, FinalState, RecordStreamProcessor<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { init(workUnitState); this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build(); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMap(in -> { if (in instanceof ControlMessage) { ControlMessage out = (ControlMessage) in; getMessageHandler().handleMessage((ControlMessage) in); if (in instanceof MetadataUpdateControlMessage) { this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput( ((MetadataUpdateControlMessage) in).getGlobalMetadata(), Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata() .getSchema(), workUnitState))).build(); out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata); } return Flowable.just(((ControlMessage<DO>) out)); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState); } else { throw new UnsupportedOperationException(); } }, 1); outputStream = outputStream.doOnComplete(this::close); return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata); }  }
Converter implements Closeable, FinalState, RecordStreamProcessor<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { init(workUnitState); this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build(); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMap(in -> { if (in instanceof ControlMessage) { ControlMessage out = (ControlMessage) in; getMessageHandler().handleMessage((ControlMessage) in); if (in instanceof MetadataUpdateControlMessage) { this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput( ((MetadataUpdateControlMessage) in).getGlobalMetadata(), Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata() .getSchema(), workUnitState))).build(); out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata); } return Flowable.just(((ControlMessage<DO>) out)); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState); } else { throw new UnsupportedOperationException(); } }, 1); outputStream = outputStream.doOnComplete(this::close); return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata); }  Converter<SI, SO, DI, DO> init(WorkUnitState workUnit); @Override void close(); abstract SO convertSchema(SI inputSchema, WorkUnitState workUnit); abstract Iterable<DO> convertRecord(SO outputSchema, DI inputRecord, WorkUnitState workUnit); @Override State getFinalState(); @Override RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState); ControlMessageHandler getMessageHandler(); ConverterInitializer getInitializer(State state, WorkUnitStream workUnits, int branches, int branchId); }
Converter implements Closeable, FinalState, RecordStreamProcessor<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { init(workUnitState); this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState))).build(); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMap(in -> { if (in instanceof ControlMessage) { ControlMessage out = (ControlMessage) in; getMessageHandler().handleMessage((ControlMessage) in); if (in instanceof MetadataUpdateControlMessage) { this.outputGlobalMetadata = GlobalMetadata.<SI, SO>builderWithInput( ((MetadataUpdateControlMessage) in).getGlobalMetadata(), Optional.fromNullable(convertSchema((SI)((MetadataUpdateControlMessage) in).getGlobalMetadata() .getSchema(), workUnitState))).build(); out = new MetadataUpdateControlMessage<SO, DO>(this.outputGlobalMetadata); } return Flowable.just(((ControlMessage<DO>) out)); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return convertRecordEnvelope(this.outputGlobalMetadata.getSchema(), recordEnvelope, workUnitState); } else { throw new UnsupportedOperationException(); } }, 1); outputStream = outputStream.doOnComplete(this::close); return inputStream.withRecordStream(outputStream, this.outputGlobalMetadata); }  Converter<SI, SO, DI, DO> init(WorkUnitState workUnit); @Override void close(); abstract SO convertSchema(SI inputSchema, WorkUnitState workUnit); abstract Iterable<DO> convertRecord(SO outputSchema, DI inputRecord, WorkUnitState workUnit); @Override State getFinalState(); @Override RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState); ControlMessageHandler getMessageHandler(); ConverterInitializer getInitializer(State state, WorkUnitStream workUnits, int branches, int branchId);  }