DataSource: Tighten contract to throw if opened beyond end-of-input

Includes fixes for the HTTP implementations, which previously
broke this contract specifically in the case when a server
responds to a range request with a HTTP 200 response. To fix
this case, skipping to the requested position is moved from
read() to open(). As a side effect, this nicely simplifies
CronetDataSource!

PiperOrigin-RevId: 359737301

extensions/cronet/src/main/java/com/google/android/exoplayer2/ext/cronet/CronetDataSource.java

@@ -321,7 +321,6 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
 
   // Accessed by the calling thread only.
   private boolean opened;
-  private long bytesToSkip;
   private long bytesRemaining;
 
   // Written from the calling thread only. currentUrlRequest.start() calls ensure writes are visible
@@ -577,7 +576,7 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
       byte[] responseBody;
       try {
         responseBody = readResponseBody();
-      } catch (HttpDataSourceException e) {
+      } catch (IOException e) {
         responseBody = Util.EMPTY_BYTE_ARRAY;
       }
 
@@ -607,7 +606,7 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
     // If we requested a range starting from a non-zero position and received a 200 rather than a
     // 206, then the server does not support partial requests. We'll need to manually skip to the
     // requested position.
-    bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
+    long bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
 
     // Calculate the content length.
     if (!isCompressed(responseInfo)) {
@@ -627,6 +626,14 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
     opened = true;
     transferStarted(dataSpec);
 
+    try {
+      if (!skipFully(bytesToSkip)) {
+        throw new DataSourceException(DataSourceException.POSITION_OUT_OF_RANGE);
+      }
+    } catch (IOException e) {
+      throw new OpenException(e, dataSpec, Status.READING_RESPONSE);
+    }
+
     return bytesRemaining;
   }
 
@@ -641,25 +648,25 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
     }
 
     ByteBuffer readBuffer = getOrCreateReadBuffer();
-    while (!readBuffer.hasRemaining()) {
+    if (!readBuffer.hasRemaining()) {
       // Fill readBuffer with more data from Cronet.
       operation.close();
       readBuffer.clear();
-      readInternal(readBuffer);
+      try {
+        readInternal(readBuffer);
+      } catch (IOException e) {
+        throw new HttpDataSourceException(
+            e, castNonNull(currentDataSpec), HttpDataSourceException.TYPE_READ);
+      }
 
       if (finished) {
         bytesRemaining = 0;
         return C.RESULT_END_OF_INPUT;
-      } else {
-        // The operation didn't time out, fail or finish, and therefore data must have been read.
-        readBuffer.flip();
-        Assertions.checkState(readBuffer.hasRemaining());
-        if (bytesToSkip > 0) {
-          int bytesSkipped = (int) Math.min(readBuffer.remaining(), bytesToSkip);
-          readBuffer.position(readBuffer.position() + bytesSkipped);
-          bytesToSkip -= bytesSkipped;
-        }
       }
+
+      // The operation didn't time out, fail or finish, and therefore data must have been read.
+      readBuffer.flip();
+      Assertions.checkState(readBuffer.hasRemaining());
     }
 
     // Ensure we read up to bytesRemaining, in case this was a Range request with finite end, but
@@ -718,17 +725,6 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
     int readLength = buffer.remaining();
 
     if (readBuffer != null) {
-      // Skip all the bytes we can from readBuffer if there are still bytes to skip.
-      if (bytesToSkip != 0) {
-        if (bytesToSkip >= readBuffer.remaining()) {
-          bytesToSkip -= readBuffer.remaining();
-          readBuffer.position(readBuffer.limit());
-        } else {
-          readBuffer.position(readBuffer.position() + (int) bytesToSkip);
-          bytesToSkip = 0;
-        }
-      }
-
       // If there is existing data in the readBuffer, read as much as possible. Return if any read.
       int copyBytes = copyByteBuffer(/* src= */ readBuffer, /* dst= */ buffer);
       if (copyBytes != 0) {
@@ -740,44 +736,23 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
       }
     }
 
-    boolean readMore = true;
+    // Fill buffer with more data from Cronet.
-    while (readMore) {
+    operation.close();
-      // If bytesToSkip > 0, read into intermediate buffer that we can discard instead of caller's
+    try {
-      // buffer. If we do not need to skip bytes, we may write to buffer directly.
+      readInternal(buffer);
-      final boolean useCallerBuffer = bytesToSkip == 0;
+    } catch (IOException e) {
-
+      throw new HttpDataSourceException(
-      operation.close();
+          e, castNonNull(currentDataSpec), HttpDataSourceException.TYPE_READ);
-
-      if (!useCallerBuffer) {
-        ByteBuffer readBuffer = getOrCreateReadBuffer();
-        readBuffer.clear();
-        if (bytesToSkip < READ_BUFFER_SIZE_BYTES) {
-          readBuffer.limit((int) bytesToSkip);
-        }
-      }
-
-      // Fill buffer with more data from Cronet.
-      readInternal(useCallerBuffer ? buffer : castNonNull(readBuffer));
-
-      if (finished) {
-        bytesRemaining = 0;
-        return C.RESULT_END_OF_INPUT;
-      } else {
-        // The operation didn't time out, fail or finish, and therefore data must have been read.
-        Assertions.checkState(
-            useCallerBuffer
-                ? readLength > buffer.remaining()
-                : castNonNull(readBuffer).position() > 0);
-        // If we meant to skip bytes, subtract what was left and repeat, otherwise, continue.
-        if (useCallerBuffer) {
-          readMore = false;
-        } else {
-          bytesToSkip -= castNonNull(readBuffer).position();
-        }
-      }
     }
 
-    final int bytesRead = readLength - buffer.remaining();
+    if (finished) {
+      bytesRemaining = 0;
+      return C.RESULT_END_OF_INPUT;
+    }
+
+    // The operation didn't time out, fail or finish, and therefore data must have been read.
+    Assertions.checkState(readLength > buffer.remaining());
+    int bytesRead = readLength - buffer.remaining();
     if (bytesRemaining != C.LENGTH_UNSET) {
       bytesRemaining -= bytesRead;
     }
@@ -885,13 +860,49 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
     currentConnectTimeoutMs = clock.elapsedRealtime() + connectTimeoutMs;
   }
 
+  /**
+   * Attempts to skip the specified number of bytes in full.
+   *
+   * @param bytesToSkip The number of bytes to skip.
+   * @throws InterruptedIOException If the thread is interrupted during the operation.
+   * @throws IOException If an error occurs reading from the source.
+   * @return Whether the bytes were skipped in full. If {@code false} then the data ended before the
+   *     specified number of bytes were skipped. Always {@code true} if {@code bytesToSkip == 0}.
+   */
+  private boolean skipFully(long bytesToSkip) throws IOException {
+    if (bytesToSkip == 0) {
+      return true;
+    }
+    ByteBuffer readBuffer = getOrCreateReadBuffer();
+    while (bytesToSkip > 0) {
+      // Fill readBuffer with more data from Cronet.
+      operation.close();
+      readBuffer.clear();
+      readInternal(readBuffer);
+      if (Thread.currentThread().isInterrupted()) {
+        throw new InterruptedIOException();
+      }
+      if (finished) {
+        return false;
+      } else {
+        // The operation didn't time out, fail or finish, and therefore data must have been read.
+        readBuffer.flip();
+        Assertions.checkState(readBuffer.hasRemaining());
+        int bytesSkipped = (int) Math.min(readBuffer.remaining(), bytesToSkip);
+        readBuffer.position(readBuffer.position() + bytesSkipped);
+        bytesToSkip -= bytesSkipped;
+      }
+    }
+    return true;
+  }
+
   /**
    * Reads the whole response body.
    *
    * @return The response body.
-   * @throws HttpDataSourceException If an error occurs reading from the source.
+   * @throws IOException If an error occurs reading from the source.
    */
-  private byte[] readResponseBody() throws HttpDataSourceException {
+  private byte[] readResponseBody() throws IOException {
     byte[] responseBody = Util.EMPTY_BYTE_ARRAY;
     ByteBuffer readBuffer = getOrCreateReadBuffer();
     while (!finished) {
@@ -914,10 +925,10 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
    * the current {@code readBuffer} object so that it is not reused in the future.
    *
    * @param buffer The ByteBuffer into which the read data is stored. Must be a direct ByteBuffer.
-   * @throws HttpDataSourceException If an error occurs reading from the source.
+   * @throws IOException If an error occurs reading from the source.
    */
   @SuppressWarnings("ReferenceEquality")
-  private void readInternal(ByteBuffer buffer) throws HttpDataSourceException {
+  private void readInternal(ByteBuffer buffer) throws IOException {
     castNonNull(currentUrlRequest).read(buffer);
     try {
       if (!operation.block(readTimeoutMs)) {
@@ -930,23 +941,18 @@ public class CronetDataSource extends BaseDataSource implements HttpDataSource {
         readBuffer = null;
       }
       Thread.currentThread().interrupt();
-      throw new HttpDataSourceException(
+      throw new InterruptedIOException();
-          new InterruptedIOException(),
-          castNonNull(currentDataSpec),
-          HttpDataSourceException.TYPE_READ);
     } catch (SocketTimeoutException e) {
       // The operation is ongoing so replace buffer to avoid it being written to by this
       // operation during a subsequent request.
       if (buffer == readBuffer) {
         readBuffer = null;
       }
-      throw new HttpDataSourceException(
+      throw e;
-          e, castNonNull(currentDataSpec), HttpDataSourceException.TYPE_READ);
     }
 
     if (exception != null) {
-      throw new HttpDataSourceException(
+      throw exception;
-          exception, castNonNull(currentDataSpec), HttpDataSourceException.TYPE_READ);
     }
   }
 

extensions/cronet/src/test/java/com/google/android/exoplayer2/ext/cronet/CronetDataSourceTest.java

@@ -256,6 +256,7 @@ public final class CronetDataSourceTest {
   public void requestSetsRangeHeader() throws HttpDataSourceException {
     testDataSpec = new DataSpec(Uri.parse(TEST_URL), 1000, 5000);
     mockResponseStartSuccess();
+    mockReadSuccess(0, 1000);
 
     dataSourceUnderTest.open(testDataSpec);
     // The header value to add is current position to current position + length - 1.
@@ -287,8 +288,6 @@ public final class CronetDataSourceTest {
     testDataSpec =
         new DataSpec.Builder()
             .setUri(TEST_URL)
-            .setPosition(1000)
-            .setLength(5000)
             .setHttpRequestHeaders(dataSpecRequestProperties)
             .build();
     mockResponseStartSuccess();
@@ -1198,6 +1197,7 @@ public final class CronetDataSourceTest {
     dataSourceUnderTest.setRequestProperty("Content-Type", TEST_CONTENT_TYPE);
 
     mockSingleRedirectSuccess();
+    mockReadSuccess(0, 1000);
 
     testResponseHeader.put("Set-Cookie", "testcookie=testcookie; Path=/video");
 
@@ -1368,7 +1368,7 @@ public final class CronetDataSourceTest {
 
   @Test
   public void allowDirectExecutor() throws HttpDataSourceException {
-    testDataSpec = new DataSpec(Uri.parse(TEST_URL), 1000, 5000);
+    testDataSpec = new DataSpec(Uri.parse(TEST_URL));
     mockResponseStartSuccess();
 
     dataSourceUnderTest.open(testDataSpec);

extensions/okhttp/src/main/java/com/google/android/exoplayer2/ext/okhttp/OkHttpDataSource.java

@@ -168,8 +168,6 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
     }
   }
 
-  private static final byte[] SKIP_BUFFER = new byte[4096];
-
   private final Call.Factory callFactory;
   private final RequestProperties requestProperties;
 
@@ -183,10 +181,8 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
   @Nullable private InputStream responseByteStream;
   private boolean opened;
 
-  private long bytesToSkip;
-  private long bytesToRead;
-
   private long bytesSkipped;
+  private long bytesToRead;
   private long bytesRead;
 
   /** @deprecated Use {@link OkHttpDataSource.Factory} instead. */
@@ -332,7 +328,7 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
     // If we requested a range starting from a non-zero position and received a 200 rather than a
     // 206, then the server does not support partial requests. We'll need to manually skip to the
     // requested position.
-    bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
+    long bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
 
     // Determine the length of the data to be read, after skipping.
     if (dataSpec.length != C.LENGTH_UNSET) {
@@ -345,13 +341,21 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
     opened = true;
     transferStarted(dataSpec);
 
+    try {
+      if (!skipFully(bytesToSkip)) {
+        throw new DataSourceException(DataSourceException.POSITION_OUT_OF_RANGE);
+      }
+    } catch (IOException e) {
+      closeConnectionQuietly();
+      throw new HttpDataSourceException(e, dataSpec, HttpDataSourceException.TYPE_OPEN);
+    }
+
     return bytesToRead;
   }
 
   @Override
   public int read(byte[] buffer, int offset, int readLength) throws HttpDataSourceException {
     try {
-      skipInternal();
       return readInternal(buffer, offset, readLength);
     } catch (IOException e) {
       throw new HttpDataSourceException(
@@ -369,8 +373,8 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
   }
 
   /**
-   * Returns the number of bytes that have been skipped since the most recent call to
+   * Returns the number of bytes that were skipped during the most recent call to {@link
-   * {@link #open(DataSpec)}.
+   * #open(DataSpec)}.
    *
    * @return The number of bytes skipped.
    */
@@ -454,30 +458,32 @@ public class OkHttpDataSource extends BaseDataSource implements HttpDataSource {
   }
 
   /**
-   * Skips any bytes that need skipping. Else does nothing.
+   * Attempts to skip the specified number of bytes in full.
-   * <p>
-   * This implementation is based roughly on {@code libcore.io.Streams.skipByReading()}.
    *
+   * @param bytesToSkip The number of bytes to skip.
    * @throws InterruptedIOException If the thread is interrupted during the operation.
-   * @throws EOFException If the end of the input stream is reached before the bytes are skipped.
+   * @throws IOException If an error occurs reading from the source.
+   * @return Whether the bytes were skipped in full. If {@code false} then the data ended before the
+   *     specified number of bytes were skipped. Always {@code true} if {@code bytesToSkip == 0}.
    */
-  private void skipInternal() throws IOException {
+  private boolean skipFully(long bytesToSkip) throws IOException {
-    if (bytesSkipped == bytesToSkip) {
+    if (bytesToSkip == 0) {
-      return;
+      return true;
     }
-
+    byte[] skipBuffer = new byte[4096];
     while (bytesSkipped != bytesToSkip) {
-      int readLength = (int) min(bytesToSkip - bytesSkipped, SKIP_BUFFER.length);
+      int readLength = (int) min(bytesToSkip - bytesSkipped, skipBuffer.length);
-      int read = castNonNull(responseByteStream).read(SKIP_BUFFER, 0, readLength);
+      int read = castNonNull(responseByteStream).read(skipBuffer, 0, readLength);
       if (Thread.currentThread().isInterrupted()) {
         throw new InterruptedIOException();
       }
       if (read == -1) {
-        throw new EOFException();
+        return false;
       }
       bytesSkipped += read;
       bytesTransferred(read);
     }
+    return true;
   }
 
   /**

library/common/src/main/java/com/google/android/exoplayer2/upstream/DefaultHttpDataSource.java

@@ -46,7 +46,6 @@ import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.zip.GZIPInputStream;
-import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 
 /**
  * An {@link HttpDataSource} that uses Android's {@link HttpURLConnection}.
@@ -221,14 +220,11 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
   @Nullable private DataSpec dataSpec;
   @Nullable private HttpURLConnection connection;
   @Nullable private InputStream inputStream;
-  private byte @MonotonicNonNull [] skipBuffer;
   private boolean opened;
   private int responseCode;
 
-  private long bytesToSkip;
-  private long bytesToRead;
-
   private long bytesSkipped;
+  private long bytesToRead;
   private long bytesRead;
 
   /** @deprecated Use {@link DefaultHttpDataSource.Factory} instead. */
@@ -400,7 +396,7 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
     // If we requested a range starting from a non-zero position and received a 200 rather than a
     // 206, then the server does not support partial requests. We'll need to manually skip to the
     // requested position.
-    bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
+    long bytesToSkip = responseCode == 200 && dataSpec.position != 0 ? dataSpec.position : 0;
 
     // Determine the length of the data to be read, after skipping.
     boolean isCompressed = isCompressed(connection);
@@ -432,13 +428,21 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
     opened = true;
     transferStarted(dataSpec);
 
+    try {
+      if (!skipFully(bytesToSkip)) {
+        throw new DataSourceException(DataSourceException.POSITION_OUT_OF_RANGE);
+      }
+    } catch (IOException e) {
+      closeConnectionQuietly();
+      throw new HttpDataSourceException(e, dataSpec, HttpDataSourceException.TYPE_OPEN);
+    }
+
     return bytesToRead;
   }
 
   @Override
   public int read(byte[] buffer, int offset, int readLength) throws HttpDataSourceException {
     try {
-      skipInternal();
       return readInternal(buffer, offset, readLength);
     } catch (IOException e) {
       throw new HttpDataSourceException(
@@ -480,8 +484,8 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
   }
 
   /**
-   * Returns the number of bytes that have been skipped since the most recent call to
+   * Returns the number of bytes that were skipped during the most recent call to {@link
-   * {@link #open(DataSpec)}.
+   * #open(DataSpec)}.
    *
    * @return The number of bytes skipped.
    */
@@ -725,22 +729,19 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
   }
 
   /**
-   * Skips any bytes that need skipping. Else does nothing.
+   * Attempts to skip the specified number of bytes in full.
-   * <p>
-   * This implementation is based roughly on {@code libcore.io.Streams.skipByReading()}.
    *
+   * @param bytesToSkip The number of bytes to skip.
    * @throws InterruptedIOException If the thread is interrupted during the operation.
-   * @throws EOFException If the end of the input stream is reached before the bytes are skipped.
+   * @throws IOException If an error occurs reading from the source.
+   * @return Whether the bytes were skipped in full. If {@code false} then the data ended before the
+   *     specified number of bytes were skipped. Always {@code true} if {@code bytesToSkip == 0}.
    */
-  private void skipInternal() throws IOException {
+  private boolean skipFully(long bytesToSkip) throws IOException {
-    if (bytesSkipped == bytesToSkip) {
+    if (bytesToSkip == 0) {
-      return;
+      return true;
     }
-
+    byte[] skipBuffer = new byte[4096];
-    if (skipBuffer == null) {
-      skipBuffer = new byte[4096];
-    }
-
     while (bytesSkipped != bytesToSkip) {
       int readLength = (int) min(bytesToSkip - bytesSkipped, skipBuffer.length);
       int read = castNonNull(inputStream).read(skipBuffer, 0, readLength);
@@ -748,11 +749,12 @@ public class DefaultHttpDataSource extends BaseDataSource implements HttpDataSou
         throw new InterruptedIOException();
       }
       if (read == -1) {
-        throw new EOFException();
+        return false;
       }
       bytesSkipped += read;
       bytesTransferred(read);
     }
+    return true;
   }
 
   /**

testutils/src/main/java/com/google/android/exoplayer2/testutil/DataSourceContractTest.java

@@ -297,14 +297,8 @@ public abstract class DataSourceContractTest {
       DataSpec dataSpec =
           new DataSpec.Builder().setUri(resource.getUri()).setPosition(resourceLength + 1).build();
       try {
-        dataSource.open(dataSpec);
+        IOException exception = assertThrows(IOException.class, () -> dataSource.open(dataSpec));
-        // TODO: For any cases excluded from the requirement that a position-out-of-range exception
+        assertThat(DataSourceException.isCausedByPositionOutOfRange(exception)).isTrue();
-        // is thrown, decide what the allowed behavior should be for the first read, and assert it.
-      } catch (IOException e) {
-        // TODO: Decide whether to assert that a position-out-of-range exception must or must not be
-        // thrown (with exclusions if necessary), rather than just asserting it must be a
-        // position-out-of-range exception *if* one is thrown at all.
-        assertThat(DataSourceException.isCausedByPositionOutOfRange(e)).isTrue();
       } finally {
         dataSource.close();
       }