FileBasedSource extends OffsetBasedSource<T> { @Override public final BoundedReader<T> createReader(PipelineOptions options) throws IOException { this.validate(); String fileOrPattern = fileOrPatternSpec.get(); if (mode == Mode.FILEPATTERN) { long startTime = System.currentTimeMillis(); List<Metadata> fileMetadata = FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata(); LOG.info("Matched {} files for pattern {}", fileMetadata.size(), fileOrPattern); List<FileBasedReader<T>> fileReaders = new ArrayList<>(); for (Metadata metadata : fileMetadata) { long endOffset = metadata.sizeBytes(); fileReaders.add( createForSubrangeOfFile(metadata, 0, endOffset).createSingleFileReader(options)); } LOG.debug( "Creating a reader for file pattern {} took {} ms", fileOrPattern, System.currentTimeMillis() - startTime); if (fileReaders.size() == 1) { return fileReaders.get(0); } return new FilePatternReader(this, fileReaders); } else { return createSingleFileReader(options); } } protected  FileBasedSource(
      ValueProvider<String> fileOrPatternSpec,
      EmptyMatchTreatment emptyMatchTreatment,
      long minBundleSize); protected  FileBasedSource(ValueProvider<String> fileOrPatternSpec, long minBundleSize); protected  FileBasedSource(
      Metadata fileMetadata, long minBundleSize, long startOffset, long endOffset); final MatchResult.Metadata getSingleFileMetadata(); final String getFileOrPatternSpec(); final ValueProvider<String> getFileOrPatternSpecProvider(); final EmptyMatchTreatment getEmptyMatchTreatment(); final Mode getMode(); @Override final FileBasedSource<T> createSourceForSubrange(long start, long end); @Override final long getEstimatedSizeBytes(PipelineOptions options); @Override void populateDisplayData(DisplayData.Builder builder); @Override final List<? extends FileBasedSource<T>> split(
      long desiredBundleSizeBytes, PipelineOptions options); @Override final BoundedReader<T> createReader(PipelineOptions options); @Override String toString(); @Override void validate(); @Override final long getMaxEndOffset(PipelineOptions options);  }
@Test public void testCloseUnstartedFilePatternReader() throws IOException { PipelineOptions options = PipelineOptionsFactory.create(); List<String> data1 = createStringDataset(3, 50); File file1 = createFileWithData("file1", data1); List<String> data2 = createStringDataset(3, 50); createFileWithData("file2", data2); List<String> data3 = createStringDataset(3, 50); createFileWithData("file3", data3); List<String> data4 = createStringDataset(3, 50); createFileWithData("otherfile", data4); TestFileBasedSource source = new TestFileBasedSource(new File(file1.getParent(), "file*").getPath(), 64, null); Reader<String> reader = source.createReader(options); try { reader.close(); } catch (Exception e) { throw new AssertionError( "Closing an unstarted FilePatternReader should not throw an exception", e); } }
@Test public void testFractionConsumedWhenReadingFilepattern() throws IOException { List<String> data1 = createStringDataset(3, 1000); File file1 = createFileWithData("file1", data1); List<String> data2 = createStringDataset(3, 1000); createFileWithData("file2", data2); List<String> data3 = createStringDataset(3, 1000); createFileWithData("file3", data3); TestFileBasedSource source = new TestFileBasedSource(file1.getParent() + "/" + "file*", 1024, null); try (BoundedSource.BoundedReader<String> reader = source.createReader(null)) { double lastFractionConsumed = 0.0; assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertTrue(reader.start()); assertTrue(reader.advance()); assertTrue(reader.advance()); assertTrue(reader.getFractionConsumed() > 0.0); assertTrue(reader.getFractionConsumed() < 1.0 / 3.0); while (reader.advance()) { double fractionConsumed = reader.getFractionConsumed(); assertTrue(fractionConsumed > lastFractionConsumed); lastFractionConsumed = fractionConsumed; } assertEquals(1.0, reader.getFractionConsumed(), 1e-6); } }