ConcatReaderFactory implements ReaderFactory { @Override public NativeReader<?> create( CloudObject spec, @Nullable Coder<?> coder, @Nullable PipelineOptions options, @Nullable DataflowExecutionContext executionContext, DataflowOperationContext operationContext) throws Exception { @SuppressWarnings("unchecked") Coder<Object> typedCoder = (Coder<Object>) coder; return createTyped(spec, typedCoder, options, executionContext, operationContext); } private  ConcatReaderFactory(ReaderRegistry registry); static ConcatReaderFactory withDefaultRegistry(); static ConcatReaderFactory withRegistry(ReaderRegistry registry); @Override NativeReader<?> create(
      CloudObject spec,
      @Nullable Coder<?> coder,
      @Nullable PipelineOptions options,
      @Nullable DataflowExecutionContext executionContext,
      DataflowOperationContext operationContext); NativeReader<T> createTyped(
      CloudObject spec,
      @Nullable Coder<T> coder,
      @Nullable PipelineOptions options,
      @Nullable DataflowExecutionContext executionContext,
      DataflowOperationContext operationContext); static Source createSourceFromDictionary(Map<String, Object> dictionary);  }
@Test public void testCreateConcatReaderWithOneSubSource() throws Exception { List<List<String>> allData = createInMemorySourceData(1, 10); Source source = createSourcesWithInMemorySources(allData); @SuppressWarnings("unchecked") NativeReader<String> reader = (NativeReader<String>) ReaderRegistry.defaultRegistry().create(source, null, null, null); assertNotNull(reader); List<String> expected = new ArrayList<>(); for (List<String> data : allData) { expected.addAll(data); } assertThat(readAllFromReader(reader), containsInAnyOrder(expected.toArray())); }
@Test public void testCreateConcatReaderWithManySubSources() throws Exception { List<List<String>> allData = createInMemorySourceData(15, 10); Source source = createSourcesWithInMemorySources(allData); @SuppressWarnings("unchecked") NativeReader<String> reader = (NativeReader<String>) ReaderRegistry.defaultRegistry().create(source, null, null, null); assertNotNull(reader); List<String> expected = new ArrayList<>(); for (List<String> data : allData) { expected.addAll(data); } assertThat(readAllFromReader(reader), containsInAnyOrder(expected.toArray())); }