@Test public void test1to1() throws Exception { MyAsyncConverter1to1 converter = new MyAsyncConverter1to1(); List<Throwable> errors = Lists.newArrayList(); AtomicBoolean done = new AtomicBoolean(false); WorkUnitState workUnitState = new WorkUnitState(); workUnitState.setProp(AsyncConverter1to1.MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, 3); RecordStreamWithMetadata<String, String> stream = new RecordStreamWithMetadata<>(Flowable.range(0, 5).map(i -> i.toString()).map(RecordEnvelope::new), GlobalMetadata.<String>builder().schema("schema").build()); Set<String> outputRecords = Sets.newConcurrentHashSet(); converter.processStream(stream, workUnitState).getRecordStream().subscribeOn(Schedulers.newThread()) .subscribe(r -> outputRecords.add(((RecordEnvelope<String>)r).getRecord()), errors::add, () -> done.set(true)); Assert.assertTrue( ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> converter.completeFutureIfPresent("0")).await()); Assert.assertTrue(ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> outputRecords.contains("0")).await()); Assert.assertEquals(outputRecords.size(), 1); Assert.assertFalse( ExponentialBackoff.awaitCondition().maxWait(200L).callable(() -> converter.completeFutureIfPresent("4")).await()); Assert.assertTrue( ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> converter.completeFutureIfPresent("3")).await()); Assert.assertTrue(ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> outputRecords.contains("3")).await()); Assert.assertEquals(outputRecords.size(), 2); Assert.assertTrue( ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> converter.completeFutureIfPresent("4")).await()); Assert.assertTrue(ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> outputRecords.contains("4")).await()); Assert.assertEquals(outputRecords.size(), 3); Assert.assertTrue( ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> converter.completeFutureIfPresent("1")).await()); Assert.assertTrue( ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> converter.completeFutureIfPresent("2")).await()); Assert.assertTrue(ExponentialBackoff.awaitCondition().maxWait(100L).callable(() -> outputRecords.size() == 5).await()); Assert.assertEquals(outputRecords, Sets.newHashSet("0", "1", "2", "3", "4")); Assert.assertTrue(errors.isEmpty()); Assert.assertTrue(done.get()); }
@Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS); SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMapSingle(in -> { if (in instanceof ControlMessage) { getMessageHandler().handleMessage((ControlMessage) in); return Single.just((ControlMessage<DO>) in); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState)); } else { throw new IllegalStateException("Expected ControlMessage or RecordEnvelope."); } }, false, maxConcurrentAsyncConversions); return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(outputSchema)).build()); }
AsyncConverter1to1 extends Converter<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS); SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMapSingle(in -> { if (in instanceof ControlMessage) { getMessageHandler().handleMessage((ControlMessage) in); return Single.just((ControlMessage<DO>) in); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState)); } else { throw new IllegalStateException("Expected ControlMessage or RecordEnvelope."); } }, false, maxConcurrentAsyncConversions); return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(outputSchema)).build()); } }
AsyncConverter1to1 extends Converter<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS); SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMapSingle(in -> { if (in instanceof ControlMessage) { getMessageHandler().handleMessage((ControlMessage) in); return Single.just((ControlMessage<DO>) in); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState)); } else { throw new IllegalStateException("Expected ControlMessage or RecordEnvelope."); } }, false, maxConcurrentAsyncConversions); return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(outputSchema)).build()); }  }
AsyncConverter1to1 extends Converter<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS); SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMapSingle(in -> { if (in instanceof ControlMessage) { getMessageHandler().handleMessage((ControlMessage) in); return Single.just((ControlMessage<DO>) in); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState)); } else { throw new IllegalStateException("Expected ControlMessage or RecordEnvelope."); } }, false, maxConcurrentAsyncConversions); return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(outputSchema)).build()); }  @Override abstract SO convertSchema(SI inputSchema, WorkUnitState workUnit); @Override final Iterable<DO> convertRecord(SO outputSchema, DI inputRecord, WorkUnitState workUnit); @Override RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState); }
AsyncConverter1to1 extends Converter<SI, SO, DI, DO> { @Override public RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream, WorkUnitState workUnitState) throws SchemaConversionException { int maxConcurrentAsyncConversions = workUnitState.getPropAsInt(MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY, DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS); SO outputSchema = convertSchema(inputStream.getGlobalMetadata().getSchema(), workUnitState); Flowable<StreamEntity<DO>> outputStream = inputStream.getRecordStream() .flatMapSingle(in -> { if (in instanceof ControlMessage) { getMessageHandler().handleMessage((ControlMessage) in); return Single.just((ControlMessage<DO>) in); } else if (in instanceof RecordEnvelope) { RecordEnvelope<DI> recordEnvelope = (RecordEnvelope<DI>) in; return new SingleAsync(recordEnvelope, convertRecordAsync(outputSchema, recordEnvelope.getRecord(), workUnitState)); } else { throw new IllegalStateException("Expected ControlMessage or RecordEnvelope."); } }, false, maxConcurrentAsyncConversions); return inputStream.withRecordStream(outputStream, GlobalMetadata.<SI, SO>builderWithInput(inputStream.getGlobalMetadata(), Optional.fromNullable(outputSchema)).build()); }  @Override abstract SO convertSchema(SI inputSchema, WorkUnitState workUnit); @Override final Iterable<DO> convertRecord(SO outputSchema, DI inputRecord, WorkUnitState workUnit); @Override RecordStreamWithMetadata<DO, SO> processStream(RecordStreamWithMetadata<DI, SI> inputStream,
      WorkUnitState workUnitState); static final String MAX_CONCURRENT_ASYNC_CONVERSIONS_KEY; static final int DEFAULT_MAX_CONCURRENT_ASYNC_CONVERSIONS; }