com.google.cloud.dataflow.sdk.io

## Class FileBasedSource.FileBasedReader<T>

- java.lang.Object
- - com.google.cloud.dataflow.sdk.io.Source.Reader<T>
  - - com.google.cloud.dataflow.sdk.io.BoundedSource.BoundedReader<T>
    - - com.google.cloud.dataflow.sdk.io.OffsetBasedSource.OffsetBasedReader<T>
      - - com.google.cloud.dataflow.sdk.io.FileBasedSource.FileBasedReader<T>

All Implemented Interfaces:
  AutoCloseable

Direct Known Subclasses:
  BlockBasedSource.BlockBasedReader, CompressedSource.CompressedReader

Enclosing class:
  FileBasedSource<T>

    public abstract static class FileBasedSource.FileBasedReader<T>
    extends OffsetBasedSource.OffsetBasedReader<T>

A `reader` that implements code common to readers of `FileBasedSource`s.

## Seekability

This reader uses a `ReadableByteChannel` created for the file represented by the corresponding source to efficiently move to the correct starting position defined in the source. Subclasses of this reader should implement `startReading(java.nio.channels.ReadableByteChannel)` to get access to this channel. If the source corresponding to the reader is for a subrange of a file the `ReadableByteChannel` provided is guaranteed to be an instance of the type `SeekableByteChannel`, which may be used by subclass to traverse back in the channel to determine the correct starting position.

## Reading Records

Sequential reading is implemented using `readNextRecord()`.

Then `FileBasedReader` implements "reading a range [A, B)" in the following way.

1. `OffsetBasedSource.OffsetBasedReader.start()` opens the file
2. `OffsetBasedSource.OffsetBasedReader.start()` seeks the `SeekableByteChannel` to A (reading offset ranges for non-seekable files is not supported) and calls `startReading()`
3. `OffsetBasedSource.OffsetBasedReader.start()` calls `OffsetBasedSource.OffsetBasedReader.advance()` once, which, via `readNextRecord()`, locates the first record which is at a split point AND its offset is at or after A. If this record is at or after B, `OffsetBasedSource.OffsetBasedReader.advance()` returns false and reading is finished.
4. if the previous advance call returned `true` sequential reading starts and `advance()` will be called repeatedly

`advance()` calls `readNextRecord()` on the subclass, and stops (returns false) if the new record is at a split point AND the offset of the new record is at or after B.

## Thread Safety

Since this class implements `Source.Reader` it guarantees thread safety. Abstract methods defined here will not be accessed by more than one thread concurrently.

### Field Summary

### Fields inherited from class com.google.cloud.dataflow.sdk.io.BoundedSource.BoundedReader

`SPLIT_POINTS_UNKNOWN`

### Constructor Summary

Constructors
Constructor and Description
`FileBasedReader(FileBasedSource<T> source)` Subclasses should not perform IO operations at the constructor.

### Method Summary

All Methods Instance Methods Abstract Methods Concrete Methods
`protected boolean` `advanceImpl()` Advances to the next record and returns `true`, or returns false if there is no next record.
`void` `close()` Closes any `ReadableByteChannel` created for the current reader.
`FileBasedSource<T>` `getCurrentSource()` Returns a `Source` describing the same input that this `Reader` currently reads (including items already read).
`protected abstract boolean` `readNextRecord()` Reads the next record from the channel provided by `startReading(java.nio.channels.ReadableByteChannel)`.
`protected boolean` `startImpl()` Initializes the `OffsetBasedSource.OffsetBasedReader` and advances to the first record, returning `true` if there is a record available to be read.
`protected abstract void` `startReading(ReadableByteChannel channel)` Performs any initialization of the subclass of `FileBasedReader` that involves IO operations.

### Methods inherited from class com.google.cloud.dataflow.sdk.io.OffsetBasedSource.OffsetBasedReader

`advance, getCurrentOffset, getFractionConsumed, getSplitPointsConsumed, getSplitPointsRemaining, isAtSplitPoint, isDone, isStarted, splitAtFraction, start`

### Methods inherited from class com.google.cloud.dataflow.sdk.io.BoundedSource.BoundedReader

`getCurrentTimestamp`

### Methods inherited from class com.google.cloud.dataflow.sdk.io.Source.Reader

`getCurrent`

### Methods inherited from class java.lang.Object

`clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait`

### Constructor Detail

#### FileBasedReader

    public FileBasedReader(FileBasedSource<T> source)

Subclasses should not perform IO operations at the constructor. All IO operations should be delayed until the `startReading(java.nio.channels.ReadableByteChannel)` method is invoked.

### Method Detail

#### getCurrentSource

    public FileBasedSource<T> getCurrentSource()

Description copied from class: `BoundedSource.BoundedReader`
Returns a `Source` describing the same input that this `Reader` currently reads (including items already read).

### Usage

Reader subclasses can use this method for convenience to access unchanging properties of the source being read. Alternatively, they can cache these properties in the constructor.

The framework will call this method in the course of dynamic work rebalancing, e.g. after a successful `BoundedSource.BoundedReader.splitAtFraction(double)` call.

### Mutability and thread safety

Remember that `Source` objects must always be immutable. However, the return value of this function may be affected by dynamic work rebalancing, happening asynchronously via `BoundedSource.BoundedReader.splitAtFraction(double)`, meaning it can return a different `Source` object. However, the returned object itself will still itself be immutable. Callers must take care not to rely on properties of the returned source that may be asynchronously changed as a result of this process (e.g. do not cache an end offset when reading a file).

### Implementation

For convenience, subclasses should usually return the most concrete subclass of `Source` possible. In practice, the implementation of this method should nearly always be one of the following:

- Source that inherits from a base class that already implements `BoundedSource.BoundedReader.getCurrentSource()`: delegate to base class. In this case, it is almost always an error for the subclass to maintain its own copy of the source.

         public FooReader(FooSource<T> source) {
           super(source);
         }

         public FooSource<T> getCurrentSource() {
           return (FooSource<T>)super.getCurrentSource();
         }

- Source that does not support dynamic work rebalancing: return a private final variable.

         private final FooSource<T> source;

         public FooReader(FooSource<T> source) {
           this.source = source;
         }

         public FooSource<T> getCurrentSource() {
           return source;
         }

- `BoundedSource.BoundedReader` that explicitly supports dynamic work rebalancing: maintain a variable pointing to an immutable source object, and protect it with synchronization.

         private FooSource<T> source;

         public FooReader(FooSource<T> source) {
           this.source = source;
         }

         public synchronized FooSource<T> getCurrentSource() {
           return source;
         }

         public synchronized FooSource<T> splitAtFraction(double fraction) {
           ...
           FooSource<T> primary = ...;
           FooSource<T> residual = ...;
           this.source = primary;
           return residual;
         }

Overrides:
  `getCurrentSource` in class `OffsetBasedSource.OffsetBasedReader<T>`

#### startImpl

    protected final boolean startImpl()
                               throws IOException

Description copied from class: `OffsetBasedSource.OffsetBasedReader`
Initializes the `OffsetBasedSource.OffsetBasedReader` and advances to the first record, returning `true` if there is a record available to be read. This method will be invoked exactly once and may perform expensive setup operations that are needed to initialize the reader.

This function is the `OffsetBasedReader` implementation of `BoundedReader#start`. The key difference is that the implementor can ignore the possibility that it should no longer produce the first record, either because it has exceeded the original `endOffset` assigned to the reader, or because a concurrent call to `OffsetBasedSource.OffsetBasedReader.splitAtFraction(double)` has changed the source to shrink the offset range being read.

Specified by:
  `startImpl` in class `OffsetBasedSource.OffsetBasedReader<T>`
Throws:
  `IOException`
See Also:
  `BoundedReader#start`

#### advanceImpl

    protected final boolean advanceImpl()
                                 throws IOException

Description copied from class: `OffsetBasedSource.OffsetBasedReader`
Advances to the next record and returns `true`, or returns false if there is no next record.

This function is the `OffsetBasedReader` implementation of `BoundedReader#advance`. The key difference is that the implementor can ignore the possibility that it should no longer produce the next record, either because it has exceeded the original `endOffset` assigned to the reader, or because a concurrent call to `OffsetBasedSource.OffsetBasedReader.splitAtFraction(double)` has changed the source to shrink the offset range being read.

Specified by:
  `advanceImpl` in class `OffsetBasedSource.OffsetBasedReader<T>`
Throws:
  `IOException`
See Also:
  `BoundedReader#advance`

#### close

    public void close()
               throws IOException

Closes any `ReadableByteChannel` created for the current reader. This implementation is idempotent. Any `close()` method introduced by a subclass must be idempotent and must call the `close()` method in the `FileBasedReader`.

Specified by:
  `close` in interface `AutoCloseable`
Specified by:
  `close` in class `Source.Reader<T>`
Throws:
  `IOException`

#### startReading

    protected abstract void startReading(ReadableByteChannel channel)
                                  throws IOException

Performs any initialization of the subclass of `FileBasedReader` that involves IO operations. Will only be invoked once and before that invocation the base class will seek the channel to the source's starting offset.

Provided `ReadableByteChannel` is for the file represented by the source of this reader. Subclass may use the `channel` to build a higher level IO abstraction, e.g., a BufferedReader or an XML parser.

If the corresponding source is for a subrange of a file, `channel` is guaranteed to be an instance of the type `SeekableByteChannel`.

After this method is invoked the base class will not be reading data from the channel or adjusting the position of the channel. But the base class is responsible for properly closing the channel.

Parameters:
  `channel` - a byte channel representing the file backing the reader.
Throws:
  `IOException`

#### readNextRecord

    protected abstract boolean readNextRecord()
                                       throws IOException

Reads the next record from the channel provided by `startReading(java.nio.channels.ReadableByteChannel)`. Methods `Source.Reader.getCurrent()`, `OffsetBasedSource.OffsetBasedReader.getCurrentOffset()`, and `OffsetBasedSource.OffsetBasedReader.isAtSplitPoint()` should return the corresponding information about the record read by the last invocation of this method.

Note that this method will be called the same way for reading the first record in the source (file or offset range in the file) and for reading subsequent records. It is up to the subclass to do anything special for locating and reading the first record, if necessary.

Returns:
  `true` if a record was successfully read, `false` if the end of the channel was reached before successfully reading a new record.
Throws:
  `IOException`