### Question:
InMemoryQueueIO { public static PTransform<PCollection<Record>, PCollection<Void>> to(final LoopState state) { return new QueuedOutputTransform(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 output() { final Collection<Record> objects = new ArrayList<>(); try (final LoopState state = LoopState.newTracker(null)) { pipeline .apply(Create.of(IntStream.range(0, 5).mapToObj(RowStruct::new).collect(toList()))) .setCoder(SerializableCoder.of(RowStruct.class)) .apply(ParDo.of(new DoFn<RowStruct, Record>() { @ProcessElement public void onElement(final ProcessContext context) { final Record record = ComponentManager .instance() .getRecordBuilderFactoryProvider() .apply(null) .newRecordBuilder() .withInt("id", context.element().id) .build(); context.output(record); } })) .setCoder(SchemaRegistryCoder.of()) .apply(InMemoryQueueIO.to(state)); pipeline.run().waitUntilFinish(); Record next; do { next = state.next(); if (next != null) { objects.add(next); } } while (next != null); } assertEquals(5, objects.size()); assertEquals(IntStream.range(0, 5).boxed().collect(toSet()), objects.stream().mapToInt(o -> o.getInt("id")).boxed().collect(toSet())); }