### Question:
InMemoryQueueIO { public static PTransform<PBegin, PCollection<Record>> from(final LoopState state) { return Read.from(new UnboundedQueuedInput(state.id)); }  static PTransform<PBegin, PCollection<Record>> from(final LoopState state); static PTransform<PCollection<Record>, PCollection<Void>> to(final LoopState state);  }### Answer:
@Test(timeout = 60000) public void input() { INPUT_OUTPUTS.clear(); final PipelineResult result; try (final LoopState state = LoopState.newTracker(null)) { IntStream.range(0, 2).forEach(i -> state.push(new RowStruct(i))); pipeline.apply(InMemoryQueueIO.from(state)).apply(ParDo.of(new DoFn<Record, Void>() { @ProcessElement public void onElement(@Element final Record record) { INPUT_OUTPUTS.add(record); } })); result = pipeline.run(); IntStream.range(2, 5).forEach(i -> state.push(new RowStruct(i))); state.end(); final long end = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(2); long lastLog = System.currentTimeMillis(); while (INPUT_OUTPUTS.size() < 5 && end - System.currentTimeMillis() >= 0) { try { if (lastLog - System.currentTimeMillis() > TimeUnit.SECONDS.toMillis(10)) { log.info("Not yet 5 records: {}, waiting", INPUT_OUTPUTS.size()); } sleep(150); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); } } } result.waitUntilFinish(); assertEquals(5, INPUT_OUTPUTS.size()); assertEquals(IntStream.range(0, 5).boxed().collect(toSet()), INPUT_OUTPUTS.stream().mapToInt(o -> o.getInt("id")).boxed().collect(toSet())); }