class FileBasedSource extends OffsetBasedSource<T> {

  @Override
  public final BoundedReader<T> createReader(PipelineOptions options) throws IOException {
    // Validate the current source prior to creating a reader for it.
    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);

}