Range options now available for BlobInputStream

jaschrep-msft · jaschrep-msft · commit 0375367b5423 · 2019-02-28T16:45:20.000-08:00
CloudBlob::openInputStream now has an overload that takes offset and
optional length. BlobInputStream constructor has a similar overload
which will only download and stream the specified range.
diff --git a/microsoft-azure-storage-test/src/com/microsoft/azure/storage/TestHelper.java b/microsoft-azure-storage-test/src/com/microsoft/azure/storage/TestHelper.java
@@ -49,6 +49,7 @@
 import com.microsoft.aad.adal4j.AuthenticationContext;
 import com.microsoft.aad.adal4j.AuthenticationResult;
 import com.microsoft.aad.adal4j.ClientCredential;
+import org.junit.Assert;
 import org.junit.AssumptionViolatedException;
 import org.w3c.dom.DOMException;
 import org.w3c.dom.Document;
@@ -322,6 +323,27 @@ public static void assertURIsEqual(URI expected, URI actual, boolean ignoreQuery
         assertTrue(actualQueries.isEmpty());
     }
 
+    public static void expectedException(Runnable method, Class<? extends Exception> exceptionType) {
+        Throwable caughtException = null;
+        try {
+            method.run();
+        } catch (Exception e) {
+            Throwable th = e;
+            while (th != null) {
+                if (th.getClass().equals(exceptionType)) {
+                    caughtException = th;
+                    break;
+                }
+                th = th.getCause();
+            }
+        } finally {
+            if (caughtException == null)
+                Assert.fail(String.format(
+                        "Expected exception of type %s not encountered",
+                        exceptionType.getCanonicalName()));
+        }
+    }
+
     public static URI defiddler(URI uri) throws URISyntaxException {
         String fiddlerString = "ipv4.fiddler";
         String replacementString = "127.0.0.1";
diff --git a/microsoft-azure-storage-test/src/com/microsoft/azure/storage/blob/CloudBlockBlobTests.java b/microsoft-azure-storage-test/src/com/microsoft/azure/storage/blob/CloudBlockBlobTests.java
@@ -25,6 +25,7 @@
 import com.microsoft.azure.storage.file.SharedAccessFilePolicy;
 
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -1989,6 +1990,118 @@ public void testBlobInputStream() throws URISyntaxException, StorageException, I
         blobRef.delete();
     }
 
+    @Test
+    @Category({ DevFabricTests.class, DevStoreTests.class })
+    public void testBlobInputStreamWithRange() throws StorageException, IOException, URISyntaxException {
+
+        final int blobLength = 4 * Constants.KB;
+        final String blobName = "testBlobInputStreamWithOffset" + UUID.randomUUID();
+
+        // setup
+        final CloudBlockBlob blob = this.container.getBlockBlobReference(blobName);
+        final byte[] blobData = BlobTestHelper.getRandomBuffer(blobLength);
+        blob.upload(new ByteArrayInputStream(blobData), blobData.length);
+
+        blob.downloadAttributes();
+
+        // test
+        doOpenInputStreamWithRangeTest(blob, blobData, 0, null);
+
+        doOpenInputStreamWithRangeTest(blob, blobData, Constants.KB, null);
+        doOpenInputStreamWithRangeTest(blob, blobData, Constants.KB, 2 * Constants.KB);
+        doOpenInputStreamWithRangeTest(blob, blobData, Constants.KB, 3 * Constants.KB);
+
+        doOpenInputStreamWithRangeTest(blob, blobData, Constants.KB, 4 * Constants.KB);
+
+        TestHelper.expectedException(
+                customRunnableWrapper(blob, blobData, -1 * Constants.KB, null),
+                IndexOutOfBoundsException.class);
+        TestHelper.expectedException(
+                customRunnableWrapper(blob, blobData, -1 * Constants.KB, 4 * Constants.KB),
+                IndexOutOfBoundsException.class);
+    }
+
+    /**
+     * Creates a Runnable wrapper for the above test with the below method. (Java 7 doesn't support lambdas).
+     * @param blob
+     * @param originalData
+     * @param offset
+     * @param length
+     * @return
+     */
+    private Runnable customRunnableWrapper(final CloudBlob blob, final byte[] originalData, final int offset, final Integer length) {
+        return new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    doOpenInputStreamWithRangeTest(blob, originalData, offset, length);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        };
+    }
+
+    /**
+     * Creates and reads from a BlobInputStream using offset and length parameters.
+     * Marks at the start and uses a reset halfway through to ensure mark and reset still work.
+     * @param blob
+     * @param originalData
+     * @param offset
+     * @param length
+     * @throws StorageException
+     * @throws IOException
+     */
+    private void doOpenInputStreamWithRangeTest(CloudBlob blob, byte[] originalData, int offset, Integer length)
+            throws StorageException, IOException {
+
+        // setup
+        final int readLength = length == null
+                ? (int) (blob.getProperties().getLength() - offset)
+                : Math.min(length, (int) (blob.getProperties().getLength() - offset));
+
+        BlobInputStream stream = blob.openInputStream(
+                offset,
+                length == null ? null : length.longValue(),
+                null, null, null);
+
+        // test
+        boolean hasResetToMark = false;
+        boolean hasMarked = false;
+        for (int i = 0; i < readLength; i++) {
+
+            if (!hasMarked && i == readLength / 4) { // if haven't marked, mark at 1/4 through 3/4
+                stream.mark(readLength / 2);
+                hasMarked = true;
+            }
+            if (!hasResetToMark && i == readLength / 2) { // call reset at halfway point ONCE
+                stream.reset();
+                i = readLength / 4;
+                hasResetToMark = true;
+            }
+            if (i == readLength * 4 / 5) { // call reset after mark expiry
+                boolean threw = false;
+                try {
+                    stream.reset();
+                } catch (IOException e) {
+                    threw = true;
+                } finally {
+                    if (!threw) {
+                        Assert.fail("Allowed reset after mark expired.");
+                    }
+                }
+            }
+
+            int data = stream.read();
+            assertTrue(String.format("%d is not greater than zero. i = %d", data, i), data >= 0);
+            assertEquals(originalData[i + offset], (byte) data);
+        }
+        assertEquals(-1, stream.read());
+
+        // cleanup
+        stream.close();
+    }
+
     @Test
     @Category({ DevFabricTests.class, DevStoreTests.class })
     public void testUploadFromByteArray() throws Exception {
diff --git a/microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobInputStream.java b/microsoft-azure-storage/src/com/microsoft/azure/storage/blob/BlobInputStream.java
@@ -69,7 +69,7 @@ public final class BlobInputStream extends InputStream {
     /**
      * Holds the stream length.
      */
-    private long streamLength = -1;
+    private long streamLength;
 
     /**
      * Holds the stream read size for both block and page blobs.
@@ -121,9 +121,14 @@ public final class BlobInputStream extends InputStream {
      */
     private AccessCondition accessCondition = null;
 
+    /**
+     * Offset of the source blob this class is configured to stream from.
+     */
+    private final long blobRangeOffset;
+
     /**
      * Initializes a new instance of the BlobInputStream class.
-     * 
+     *
      * @param parentBlob
      *            A {@link CloudBlob} object which represents the blob that this stream is associated with.
      * @param accessCondition
@@ -133,19 +138,50 @@ public final class BlobInputStream extends InputStream {
      *            request.
      * @param opContext
      *            An {@link OperationContext} object which is used to track the execution of the operation.
-     * 
+     *
      * @throws StorageException
      *             An exception representing any error which occurred during the operation.
      */
     @DoesServiceRequest
     protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition accessCondition,
-            final BlobRequestOptions options, final OperationContext opContext) throws StorageException {
+                              final BlobRequestOptions options, final OperationContext opContext) throws StorageException {
+        this(0, null, parentBlob, accessCondition, options, opContext);
+    }
+
+    /**
+     * Initializes a new instance of the BlobInputStream class.
+     * Note that if {@code blobRangeOffset} is not {@code 0} or {@code blobRangeLength} is not {@code null}, there will
+     * be no content MD5 verification.
+     *
+     * @param blobRangeOffset
+     *            The offset of blob data to begin stream.
+     * @param blobRangeLength
+     *            How much data the stream should return after blobRangeOffset.
+     * @param parentBlob
+     *            A {@link CloudBlob} object which represents the blob that this stream is associated with.
+     * @param accessCondition
+     *            An {@link AccessCondition} object which represents the access conditions for the blob.
+     * @param options
+     *            A {@link BlobRequestOptions} object which represents that specifies any additional options for the
+     *            request.
+     * @param opContext
+     *            An {@link OperationContext} object which is used to track the execution of the operation.
+     * 
+     * @throws StorageException
+     *             An exception representing any error which occurred during the operation.
+     */
+    @DoesServiceRequest
+    protected BlobInputStream(long blobRangeOffset, Long blobRangeLength, final CloudBlob parentBlob,
+            final AccessCondition accessCondition, final BlobRequestOptions options, final OperationContext opContext)
+            throws StorageException {
+
+        this.blobRangeOffset = blobRangeOffset;
         this.parentBlobRef = parentBlob;
         this.parentBlobRef.assertCorrectBlobType();
         this.options = new BlobRequestOptions(options);
         this.opContext = opContext;
         this.streamFaulted = false;
-        this.currentAbsoluteReadPosition = 0;
+        this.currentAbsoluteReadPosition = blobRangeOffset;
         this.readSize = parentBlob.getStreamMinimumReadSizeInBytes();
 
         if (options.getUseTransactionalContentMD5() && this.readSize > 4 * Constants.MB) {
@@ -154,12 +190,22 @@ protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition acce
 
         parentBlob.downloadAttributes(accessCondition, this.options, this.opContext);
 
+        Utility.assertInBounds("blobRangeOffset", blobRangeOffset, 0, parentBlob.getProperties().getLength() - 1);
+        if (blobRangeLength != null) {
+            Utility.assertGreaterThanOrEqual("blobRangeLength", blobRangeLength, 0);
+        }
+
         this.retrievedContentMD5Value = parentBlob.getProperties().getContentMD5();
 
         // Will validate it if it was returned
         this.validateBlobMd5 = !options.getDisableContentMD5Validation()
                 && !Utility.isNullOrEmpty(this.retrievedContentMD5Value);
 
+        // Need the whole blob to validate MD5. If we download a range, don't bother trying.
+        if (blobRangeOffset != 0 || blobRangeLength != null) {
+            this.validateBlobMd5 = false;
+        }
+
         // Validates the first option, and sets future requests to use if match
         // request option.
 
@@ -179,7 +225,9 @@ protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition acce
         this.accessCondition = AccessCondition.generateIfMatchCondition(this.parentBlobRef.getProperties().getEtag());
         this.accessCondition.setLeaseID(previousLeaseId);
 
-        this.streamLength = parentBlob.getProperties().getLength();
+        this.streamLength = blobRangeLength == null
+                ? this.parentBlobRef.getProperties().getLength() - this.blobRangeOffset
+                : Math.min(this.parentBlobRef.getProperties().getLength() - this.blobRangeOffset, blobRangeLength);
 
         if (this.validateBlobMd5) {
             try {
@@ -191,7 +239,7 @@ protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition acce
             }
         }
 
-        this.reposition(0);
+        this.reposition(blobRangeOffset);
     }
 
     /**
@@ -238,7 +286,7 @@ public synchronized void close() throws IOException {
     }
 
     /**
-     * Dispatches a read operation of N bytes. When using sparspe page blobs the page ranges are evaluated and zero
+     * Dispatches a read operation of N bytes. When using sparse page blobs, the page ranges are evaluated and zero
      * bytes may be generated on the client side for some ranges that do not exist.
      * 
      * @param readLength
@@ -444,8 +492,8 @@ private synchronized int readInternal(final byte[] b, final int off, int len) th
 
         // if buffer is empty do next get operation
         if ((this.currentBuffer == null || this.currentBuffer.available() == 0)
-                && this.currentAbsoluteReadPosition < this.streamLength) {
-            this.dispatchRead((int) Math.min(this.readSize, this.streamLength - this.currentAbsoluteReadPosition));
+                && this.currentAbsoluteReadPosition < this.streamLength + this.blobRangeOffset) {
+            this.dispatchRead((int) Math.min(this.readSize, this.streamLength + this.blobRangeOffset - this.currentAbsoluteReadPosition));
         }
 
         len = Math.min(len, this.readSize);
@@ -459,7 +507,7 @@ private synchronized int readInternal(final byte[] b, final int off, int len) th
             if (this.validateBlobMd5) {
                 this.md5Digest.update(b, off, numberOfBytesRead);
 
-                if (this.currentAbsoluteReadPosition == this.streamLength) {
+                if (this.currentAbsoluteReadPosition == this.streamLength + this.blobRangeOffset) {
                     // Reached end of stream, validate md5.
                     final String calculatedMd5 = Base64.encode(this.md5Digest.digest());
                     if (!calculatedMd5.equals(this.retrievedContentMD5Value)) {
@@ -479,7 +527,7 @@ private synchronized int readInternal(final byte[] b, final int off, int len) th
 
         // update markers
         if (this.markExpiry > 0 && this.markedPosition + this.markExpiry < this.currentAbsoluteReadPosition) {
-            this.markedPosition = 0;
+            this.markedPosition = this.blobRangeOffset;
             this.markExpiry = 0;
         }
 
@@ -532,7 +580,7 @@ public synchronized long skip(final long n) throws IOException {
             return 0;
         }
 
-        if (n < 0 || this.currentAbsoluteReadPosition + n > this.streamLength) {
+        if (n < 0 || this.currentAbsoluteReadPosition + n > this.streamLength + this.blobRangeOffset) {
             throw new IndexOutOfBoundsException();
         }
 
diff --git a/microsoft-azure-storage/src/com/microsoft/azure/storage/blob/CloudBlob.java b/microsoft-azure-storage/src/com/microsoft/azure/storage/blob/CloudBlob.java
@@ -2464,7 +2464,7 @@ public final boolean isSnapshot() {
      */
     @DoesServiceRequest
     public final BlobInputStream openInputStream() throws StorageException {
-        return this.openInputStream(null /* accessCondition */, null /* options */, null /* opContext */);
+        return this.openInputStream(0 /* offset */, null /* range */, null /* accessCondition */, null /* options */, null /* opContext */);
     }
 
     /**
@@ -2491,14 +2491,50 @@ public final BlobInputStream openInputStream() throws StorageException {
     @DoesServiceRequest
     public final BlobInputStream openInputStream(final AccessCondition accessCondition, BlobRequestOptions options,
             OperationContext opContext) throws StorageException {
+        return this.openInputStream(0, null, accessCondition, options, opContext);
+    }
+
+    /**
+     * Opens a blob input stream to download the blob using the specified request options and operation context.
+     * <p>
+     * Use {@link #setStreamMinimumReadSizeInBytes(int)} to configure the read size.
+     *
+     * @param offset
+     *            A <code>long</code> which represents the offset to use as the starting point for the source.
+     * @param length
+     *            A {@link Long} which represents the number of bytes to read or <code>null</code>.
+     * @param accessCondition
+     *            An {@link AccessCondition} object that represents the access conditions for the blob.
+     * @param options
+     *            A {@link BlobRequestOptions} object that specifies any additional options for the request. Specifying
+     *            <code>null</code> will use the default request options from the associated service client (
+     *            {@link CloudBlobClient}).
+     * @param opContext
+     *            An {@link OperationContext} object that represents the context for the current operation. This object
+     *            is used to track requests to the storage service, and to provide additional runtime information about
+     *            the operation.
+     *
+     * @return An <code>InputStream</code> object that represents the stream to use for reading from the blob.
+     *
+     * @throws StorageException
+     *             If a storage service error occurred.
+     */
+    @DoesServiceRequest
+    public final BlobInputStream openInputStream(final long offset, final Long length, final AccessCondition accessCondition, BlobRequestOptions options,
+            OperationContext opContext) throws StorageException {
+
+        if (offset < 0 || (length != null && length <= 0)) {
+            throw new IndexOutOfBoundsException();
+        }
+
         if (opContext == null) {
             opContext = new OperationContext();
         }
 
         options = BlobRequestOptions.populateAndApplyDefaults(options, this.properties.getBlobType(), this.blobServiceClient, 
                 false /* setStartTime */);
 
-        return new BlobInputStream(this, accessCondition, options, opContext);
+        return new BlobInputStream(offset, length, this, accessCondition, options, opContext);
     }
     
     /**
