Merge pull request #441 from jaschrep-msft/samsung-asks

New options for blob downloads

Author: rickle-msft

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";

microsoft-azure-storage-test/src/com/microsoft/azure/storage/blob/BlobTestHelper.java

@@ -466,4 +466,17 @@ public static void disableSoftDelete() throws StorageException, URISyntaxExcepti
         bClient.uploadServiceProperties(serviceProperties);
         Thread.sleep(30000);
     }
+
+    public static HashMap<String, String> generateSampleMetadata(final int entries) {
+        if (entries < 0) {
+            throw new IllegalArgumentException();
+        }
+
+        HashMap<String, String> result = new HashMap<>(entries);
+        for (int i = 0; i < entries; i++) {
+            result.put("metadataKey_" + i, "metadataValue_" + i);
+        }
+
+        return result;
+    }
 }

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 {
@@ -2586,4 +2699,36 @@ public void testGetAccountInfo() throws StorageException, URISyntaxException, In
         assertNotNull(accountInformation.getAccountKind());
         assertNotNull(accountInformation.getSkuName());
     }
+
+    @Test
+    public void testSkipEtagCheck() throws StorageException, IOException, URISyntaxException {
+        final int blobSize = 2 * Constants.DEFAULT_MINIMUM_READ_SIZE_IN_BYTES; // so BlobInputStream doesn't read entire blob at once.
+
+        // setup
+        CloudBlockBlob blob = (CloudBlockBlob)BlobTestHelper.uploadNewBlob(
+                this.container, BlobType.BLOCK_BLOB, "testSkipEtagCheck", blobSize, null);
+
+        BlobRequestOptions options = new BlobRequestOptions();
+        options.setSkipEtagLocking(true); // Only request option to skip for these downloads. The rest is automatic.
+        BlobInputStream stream = blob.openInputStream(null, options, null);
+
+        // test
+        byte[] buffer = new byte[Constants.DEFAULT_MINIMUM_READ_SIZE_IN_BYTES];
+
+        assertEquals(Constants.DEFAULT_MINIMUM_READ_SIZE_IN_BYTES, stream.read(buffer)); // read 1st half of blob
+        blob.downloadAttributes();
+        String etag1 = blob.getProperties().getEtag();
+
+        blob.setMetadata(BlobTestHelper.generateSampleMetadata(1));
+        blob.uploadMetadata(); // change etag
+
+        assertEquals(Constants.DEFAULT_MINIMUM_READ_SIZE_IN_BYTES, stream.read(buffer)); // read 2nd half of blob
+        blob.downloadAttributes();
+        String etag2 = blob.getProperties().getEtag();
+
+        assertNotEquals(etag1, etag2); // assert etags were actually different if we get here without throwing
+
+        // cleanup
+        stream.close();
+    }
 }

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.
 
@@ -176,10 +222,15 @@ protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition acce
             }
         }
 
-        this.accessCondition = AccessCondition.generateIfMatchCondition(this.parentBlobRef.getProperties().getEtag());
+        this.accessCondition = new AccessCondition();
         this.accessCondition.setLeaseID(previousLeaseId);
+        if (!options.getSkipEtagLocking()) {
+            this.accessCondition.setIfMatch(this.parentBlobRef.getProperties().getEtag());
+        }
 
-        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 +242,7 @@ protected BlobInputStream(final CloudBlob parentBlob, final AccessCondition acce
             }
         }
 
-        this.reposition(0);
+        this.reposition(blobRangeOffset);
     }
 
     /**
@@ -238,7 +289,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 +495,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 +510,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 +530,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 +583,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();
         }