Java 类com.amazonaws.services.s3.model.S3ObjectInputStream 实例源码

@Override
public S3Object getObject(GetObjectRequest getObjectRequest)
        throws AmazonClientException, AmazonServiceException {
    // in ESBlobStoreContainerTestCase.java, the prefix is empty,
    // so the key and blobName are equivalent to each other
    String blobName = getObjectRequest.getKey();

    if (!blobs.containsKey(blobName)) {
        throw new AmazonS3Exception("[" + blobName + "] does not exist.");
    }

    // the HTTP request attribute is irrelevant for reading
    S3ObjectInputStream stream = new S3ObjectInputStream(
            blobs.get(blobName), null, false);
    S3Object s3Object = new S3Object();
    s3Object.setObjectContent(stream);
    return s3Object;
}

@Test
public void getInventReportSuccess() throws Exception {
    testLocator.setMD5checksum(testMD5);
    testManifest.setFileSchema("storageClass, size");
    reportRetriever = new InventoryReportRetriever(mockS3Client, testLocator, testManifest);

    String expectedInventoryReportString = "testString";
    byte[] expectedInventoryReportBytes = inventReportBytes(expectedInventoryReportString);
    when(mockS3Object.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(expectedInventoryReportBytes), null));
    when(mockS3Client.getObject(getObjectRequestCaptor.capture())).thenReturn(mockS3Object);

    String result = reportRetriever.getInventoryReportToString();
    assertThat(result, is(expectedInventoryReportString));

    GetObjectRequest request = getObjectRequestCaptor.getValue();
    assertThat(request.getBucketName(), is("testBucket"));
    assertThat(request.getKey(), is("testInventReportKey"));
}

@Test
public void getInventoryManifestSuccess() throws Exception {
    InventoryManifest expectedManifest = manifest();
    byte[] expectedManifestBytes = manifestBytes(expectedManifest);
    when(mockS3JsonObject.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(expectedManifestBytes), null));

    String expectedChecksum = "a6121a6a788be627a68d7e9def9f6968";
    byte[] expectedChecksumBytes = expectedChecksum.getBytes(StandardCharsets.UTF_8);
    when(mockS3ChecksumObject.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(expectedChecksumBytes), null));

    when(mockS3Client.getObject(getObjectRequestCaptor.capture()))
            .thenReturn(mockS3JsonObject)
            .thenReturn(mockS3ChecksumObject);
    InventoryManifest result = retriever.getInventoryManifest();
    assertThat(result, is(expectedManifest));

    List<GetObjectRequest> request = getObjectRequestCaptor.getAllValues();
    assertThat(request.get(0).getBucketName(), is("testBucketName"));
    assertThat(request.get(0).getKey(), is("testBucketKey/manifest.json"));
    assertThat(request.get(1).getBucketName(), is("testBucketName"));
    assertThat(request.get(1).getKey(), is("testBucketKey/manifest.checksum"));
}

@Test (expected = ChecksumMismatchException.class)
public void getInventoryManifestMD5Mismatch() throws Exception {
    InventoryManifest expectedManifest = manifest();
    byte[] expectedManifestBytes = manifestBytes(expectedManifest);
    byte[] errorBytes = "ERROR".getBytes();
    byte[] wrongManifestBytes = ArrayUtils.addAll(expectedManifestBytes, errorBytes);
    when(mockS3JsonObject.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(wrongManifestBytes), null));
    String expectedChecksum = "37289f10a76751046658f6c5e0ab41d9";

    byte[] expectedChecksumBytes = expectedChecksum.getBytes(StandardCharsets.UTF_8);
    when(mockS3ChecksumObject.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(expectedChecksumBytes), null));
    when(mockS3Client.getObject(getObjectRequestCaptor.capture())).
            thenReturn(mockS3JsonObject)
            .thenReturn(mockS3ChecksumObject);
    retriever.getInventoryManifest();
}

@Test
public void contentTest() throws Exception {

    URL url = this.getClass().getResource("../../../../amazon-aws-logo.jpg");
    String tmpFileName = url.getFile();

    File file = new File(tmpFileName);
    String fileName = file.getName();
    InputStream is = url.openStream();
    String contentType = URLConnection.guessContentTypeFromStream(is);

    contentHelper.uploadContent(contentType, file.length(), bucketName, fileName, is);

    Thread.sleep(500);
    boolean doesObjectExist = s3Client.doesObjectExist(bucketName, fileName);
    Assert.assertTrue(doesObjectExist);

    S3ObjectInputStream inputStream = contentHelper.downloadContent(bucketName, fileName);
    Assert.assertNotNull(inputStream);

    contentHelper.deleteContent(bucketName, fileName);
    Thread.sleep(500);

    doesObjectExist = s3Client.doesObjectExist(bucketName, fileName);
    Assert.assertFalse(doesObjectExist);
}

/**
 * Reads bytes into {@link #target}, until either the end of {@link #target} or the end of the S3 object is
 * reached.
 *
 * @param s3Object the S3 object
 * @param getObjectRequest the S3 get-object request used for retrieving {@code s3Object}
 * @return the total size of the S3 object
 * @throws AmazonClientException if a call to {@link S3ObjectInputStream#read(byte[], int, int)} does not read
 *     any bytes even though it should have
 * @throws IOException if a call to {@link S3ObjectInputStream#read(byte[], int, int)} throws an I/O exception
 */
private long readS3Object(@Nullable S3Object s3Object, GetObjectRequest getObjectRequest) throws IOException {
    long totalSize;
    if (s3Object == null) {
        totalSize = s3Client.getObjectMetadata(bucketName, key).getInstanceLength();
        if (offsetInS3Object < totalSize) {
            throw new AmazonClientException(String.format(
                "Could not read %s (range: %s), because AmazonS3#getClient() returned null.",
                key, Arrays.toString(getObjectRequest.getRange())
            ));
        }
    } else {
        totalSize = s3Object.getObjectMetadata().getInstanceLength();
        // Note that the (int) cast is safe because target.length is of type int.
        int remainingBytesToRead
            = (int) Math.max(0, Math.min(target.length - posInTarget, totalSize - offsetInS3Object));

        S3ObjectInputStream inputStream = s3Object.getObjectContent();
        int bytesRead;
        while (remainingBytesToRead > 0) {
            // read() promises to read "up to" remainingBytesToRead bytes. There is no guarantee that
            // this many bytes are read, even if enough bytes are available. In fact, experiments showed
            // that read() sometimes only returns 2^15 bytes.
            bytesRead = inputStream.read(target, posInTarget, remainingBytesToRead);
            posInTarget += bytesRead;
            remainingBytesToRead -= bytesRead;
            if (bytesRead <= 0) {
                // This should not happen and indicates a logical bug. We therefore fail here.
                throw new AmazonClientException(String.format(
                    "Could not read %s (range: %s). Requested %d bytes from input stream, but "
                        + "S3ObjectInputStream#read() returned %d.",
                    key, Arrays.toString(getObjectRequest.getRange()),
                    remainingBytesToRead, bytesRead
                ));
            }
        }
    }
    return totalSize;
}

@Test
public void testGet() {
    AmazonS3 client = mock(AmazonS3.class);
    S3StoreService service = new S3StoreService(client, S3_BUCKET, S3_PREFIX);

    String path = "path";
    String value = "value";

    ArgumentCaptor<GetObjectRequest> request = ArgumentCaptor.forClass(GetObjectRequest.class);

    S3Object s3Object = new S3Object();
    s3Object.setObjectContent(new S3ObjectInputStream(IOUtils.toInputStream(value), mock(HttpRequestBase.class)));

    when(client.getObject(request.capture())).thenReturn(s3Object);

    // invoke method under test
    Optional<String> result = service.get(path);

    assertTrue(result.isPresent());
    assertEquals(value, result.get());

    assertEquals(S3_BUCKET, request.getValue().getBucketName());
    assertEquals(S3_PREFIX + "/" + path, request.getValue().getKey());
}

@Test
public void testGetNoSuchKey() {
    AmazonS3 client = mock(AmazonS3.class);
    S3StoreService service = new S3StoreService(client, S3_BUCKET, S3_PREFIX);

    String path = "path";
    String value = "value";

    ArgumentCaptor<GetObjectRequest> request = ArgumentCaptor.forClass(GetObjectRequest.class);

    S3Object s3Object = new S3Object();
    s3Object.setObjectContent(new S3ObjectInputStream(IOUtils.toInputStream(value), mock(HttpRequestBase.class)));

    AmazonServiceException error = new AmazonServiceException("fake expected exception");
    error.setErrorCode("NoSuchKey");

    when(client.getObject(request.capture())).thenThrow(error);

    // invoke method under test
    Optional<String> result = service.get(path);

    assertFalse(result.isPresent());

    assertEquals(S3_BUCKET, request.getValue().getBucketName());
    assertEquals(S3_PREFIX + "/" + path, request.getValue().getKey());
}

@Test
public void testGetIOException() throws IOException {
    AmazonS3 client = mock(AmazonS3.class);
    S3StoreService service = new S3StoreService(client, S3_BUCKET, S3_PREFIX);

    InputStream is = mock(InputStream.class);
    when(is.read()).thenThrow(new IOException("fake exception"));

    S3Object s3Object = new S3Object();
    s3Object.setObjectContent(new S3ObjectInputStream(is, mock(HttpRequestBase.class)));

    when(client.getObject(any())).thenReturn(s3Object);

    try {
        // invoke method under test
        service.get("some-path");
        fail("expected exception not thrown");
    } catch (UnexpectedDataEncodingException ex) {
        assertTrue(ex.getMessage().contains("Unable to read contents of S3 object"));
    }
}

@Test
public void testAwsPrivateKeyStore() throws Exception {
    String bucketName = "my_bucket";
    String keyName = "my_key";
    String expected = "my_value";

    AmazonS3 s3 = Mockito.mock(AmazonS3.class);
    AWSKMS kms = Mockito.mock(AWSKMS.class);
    S3Object s3Object = Mockito.mock(S3Object.class);
    Mockito.when(s3.getObject(bucketName, keyName)).thenReturn(s3Object);
    InputStream is = new ByteArrayInputStream( expected.getBytes() );
    S3ObjectInputStream s3ObjectInputStream = new S3ObjectInputStream(is, null);
    Mockito.when(s3Object.getObjectContent()).thenReturn(s3ObjectInputStream);

    String result = expected;
    ByteBuffer buffer = ByteBuffer.wrap(result.getBytes());
    DecryptResult decryptResult = Mockito.mock(DecryptResult.class); 
    Mockito.when(kms.decrypt(Mockito.any(DecryptRequest.class))).thenReturn(decryptResult);
    Mockito.when(decryptResult.getPlaintext()).thenReturn(buffer);

    AwsPrivateKeyStore awsPrivateKeyStore = new AwsPrivateKeyStore(s3, kms);
    String actual = awsPrivateKeyStore.getApplicationSecret(bucketName, keyName);
    Assert.assertEquals(actual, expected);

}

SignedDomain getSignedDomain(AmazonS3 s3, String domainName) {

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("AWSS3ChangeLog: getting signed domain {}", domainName);
        }

        SignedDomain signedDomain = null;
        try {
            S3Object object = s3.getObject(s3BucketName, domainName);
            try (S3ObjectInputStream s3is = object.getObjectContent()) {
                byte[] data = ByteStreams.toByteArray(s3is);
                signedDomain = JSON.fromBytes(data, SignedDomain.class);
            }
        } catch (Exception ex) {
            LOGGER.error("AWSS3ChangeLog: getSignedDomain - unable to get domain {} error: {}",
                    domainName, ex.getMessage());
        }
        return signedDomain;
    }

/**
 * S3 block read would be achieved through the AmazonS3 client. Following
 * are the steps to achieve: (1) Create the objectRequest from bucketName
 * and filePath. (2) Set the range to the above created objectRequest. (3)
 * Get the object portion through AmazonS3 client API. (4) Get the object
 * content from the above object portion.
 *
 * @param bytesFromCurrentOffset
 *          bytes read till now from current offset
 * @param bytesToFetch
 *          the number of bytes to be fetched
 * @return the number of bytes read, -1 if 0 bytes read
 * @throws IOException
 */

@Override
protected int readData(final long bytesFromCurrentOffset, final int bytesToFetch) throws IOException
{
  GetObjectRequest rangeObjectRequest = new GetObjectRequest(s3Params.bucketName, s3Params.filePath);
  rangeObjectRequest.setRange(offset + bytesFromCurrentOffset, offset + bytesFromCurrentOffset + bytesToFetch - 1);
  S3Object objectPortion = s3Params.s3Client.getObject(rangeObjectRequest);
  S3ObjectInputStream wrappedStream = objectPortion.getObjectContent();
  buffer = ByteStreams.toByteArray(wrappedStream);
  wrappedStream.close();
  int bufferLength = buffer.length;
  if (bufferLength <= 0) {
    return -1;
  }
  return bufferLength;
}

/**
 * S3 block read would be achieved through the AmazonS3 client. Following are the steps to achieve:
 * (1) Create the objectRequest from bucketName and filePath.
 * (2) Set the range to the above created objectRequest.
 * (3) Get the object portion through AmazonS3 client API.
 * (4) Get the object content from the above object portion.
 * @return the block entity
 * @throws IOException
 */
@Override
protected Entity readEntity() throws IOException
{
  entity.clear();
  GetObjectRequest rangeObjectRequest = new GetObjectRequest(
      bucketName, filePath);
  rangeObjectRequest.setRange(offset, blockMetadata.getLength() - 1);
  S3Object objectPortion = s3Client.getObject(rangeObjectRequest);
  S3ObjectInputStream wrappedStream = objectPortion.getObjectContent();
  byte[] record = ByteStreams.toByteArray(wrappedStream);
  entity.setUsedBytes(record.length);
  entity.setRecord(record);
  wrappedStream.close();
  return entity;
}

@Override
public InputStream readFileData(FileData fileData) {
    String path = FilenameUtils.separatorsToUnix(FilenameUtils.normalize(extraPath + fileData.getPath() + "/" + fileData.getFileName()));
    path = StringUtils.stripStart(path, "/");
    InputStream ret = null;
    S3ObjectInputStream objectContent = null;
    try {
        S3Object object = s3Client.getObject(bucketName, path);
        if (object != null) {
            ByteArrayOutputStream temp = new ByteArrayOutputStream();
            objectContent = object.getObjectContent();
            IOUtils.copy(objectContent, temp);
            ret = new ByteArrayInputStream(temp.toByteArray());
            if (compress) {
                ret = new GZIPInputStream(ret);
            }
        }
    } catch (Exception e) {
        LOG.error("Error getting File: " + e, e);
        throw new RuntimeException(e);
    } finally {
        IOUtils.closeQuietly(objectContent);
    }
    return ret;
}

/**
 * 
 * @param key
 * @return
 */
public InputStream getFile(String bucketName, String path) {
    InputStream ret = null;
    S3ObjectInputStream objectContent = null;
    try {
        S3Object object = s3Client.getObject(bucketName, path);
        if (object != null) {
            ByteArrayOutputStream temp = new ByteArrayOutputStream();
            objectContent = object.getObjectContent();
            IOUtils.copy(objectContent, temp);
            ret = new ByteArrayInputStream(temp.toByteArray());
        }
    } catch (Exception e) {
        LOG.error("Error getting File: " + e, e);
        throw new RuntimeException(e);
    } finally {
        IOUtils.closeQuietly(objectContent);
    }
    return ret;
}

private static void streamReadAndDownloadObject(
        final File workspace,
        final S3Object sessionObject,
        final String downloadedFileName) throws IOException {

    final File outputFile = new File(workspace, downloadedFileName);

    try (final S3ObjectInputStream objectContents = sessionObject.getObjectContent();
         final OutputStream outputStream = new FileOutputStream(outputFile)) {
        final int BUFFER_SIZE = 8192;
        final byte[] buffer = new byte[BUFFER_SIZE];

        int i;
        while ((i = objectContents.read(buffer)) != -1) {
            outputStream.write(buffer, 0, i);
        }
    }
}

@Test
public void testProcessSQS() throws Exception {
    final HttpRequestBase request = mock(HttpRequestBase.class);
    final S3ObjectInputStream stream = new S3ObjectInputStream(Resources
            .asByteSource(
                    Resources.getResource("fixtures/s3_object.txt.gz"))
            .openStream(), request);

    final ObjectMetadata metadata = new ObjectMetadata();
    metadata.setContentEncoding("gzip");
    final S3Object object = new S3Object();
    object.setObjectMetadata(metadata);
    object.setObjectContent(stream);

    when(broadcaster.isEmpty()).thenReturn(false);
    when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object);

    message.setBody(FixtureHelpers.fixture("fixtures/sqs_records.json"));
    final boolean actual = processor.test(message);

    verify(broadcaster, times(2)).isEmpty();
    verify(broadcaster, times(10)).test(anyString());
    verify(s3).fetch(any(S3EventNotificationRecord.class));
    verify(request, never()).abort();
    assertThat(actual).isTrue();
}

@Test
public void index() throws Exception {
    Context ctx = createContext();
    String content = "aaaaaaaaaaaaaa";
    ObjectMetadata meta = mock(ObjectMetadata.class);
    when(meta.getContentLength()).thenReturn(Long.valueOf(content.length()));
    S3Object obj = mock(S3Object.class);
    when(obj.getObjectMetadata()).thenReturn(meta);
    InputStream in = new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8));
    when(obj.getObjectContent()).thenReturn(new S3ObjectInputStream(in, null));
    when(s3.getObject(anyString(), anyString())).thenReturn(obj);

    input.seed = "index.html";
    Response resp = target.handleRequest(input, ctx);
    assertEquals(content, resp.getContent());
}

private void closeStream()
{
    if (in != null) {
        try {
            if (in instanceof S3ObjectInputStream) {
                ((S3ObjectInputStream) in).abort();
            }
            else {
                in.close();
            }
        }
        catch (IOException | AbortedException ignored) {
            // thrown if the current thread is in the interrupted state
        }
        in = null;
        STATS.connectionReleased();
    }
}

/**
 * Gets specified text file content from specified S3 bucket.
 *
 * @param bucketName
 * @param fileName
 * @return
 */
public String readTextFileContentFromBucket(String bucketName, String fileName) {

    final S3Object s3object = s3client.getObject(bucketName, fileName);
    final S3ObjectInputStream inputStream = s3object.getObjectContent();

    final StringWriter writer = new StringWriter();
    try {
        IOUtils.copy(inputStream, writer, "UTF-8");
    } catch (IOException ex) {
        log.error("Error copying file from s3: " + ex);
    }
    return writer.toString();
}

@Test (expected = ChecksumMismatchException.class)
public void getInventReportMD5Mismatch() throws Exception {
    testLocator.setMD5checksum("badChecksum");
    testManifest.setFileSchema("storageClass, size");
    reportRetriever = new InventoryReportRetriever(mockS3Client, testLocator, testManifest);

    String expectedInventoryReportString = "testString";
    byte[] expectedInventReportBytes = inventReportBytes(expectedInventoryReportString);
    when(mockS3Object.getObjectContent()).thenReturn(new S3ObjectInputStream(
            new ByteArrayInputStream(expectedInventReportBytes), null));
    when(mockS3Client.getObject(getObjectRequestCaptor.capture())).thenReturn(mockS3Object);
    reportRetriever.getInventoryReportToString();
}

@Test
public void testReadFileFromS3() throws IOException {
    final String testInput = "Test Input";
    final S3ObjectInputStream s3ObjectInputStream = new S3ObjectInputStream(
            new ByteArrayInputStream(testInput.getBytes()),
            mock(HttpRequestBase.class),
            false);

    final S3Object s3Object = mock(S3Object.class);
    when(s3Object.getObjectContent()).thenReturn(s3ObjectInputStream);

    when(amazonS3Client.getObject(any(GetObjectRequest.class))).thenReturn(s3Object);

    assertThat(awsHelperService.readFileFromS3("http://bucket.s3.amazonaws.com"), equalTo(testInput));
}

/**
 * Download an object data as a file
 *
 * @param remoteObjectName the name of object/key which contents should be downloaded
 * @param localFileName the location and file name on the local machine, where the file will be downloaded
 * @throws S3OperationException if there is an error during data transfer
 */
@PublicAtsApi
public void download( String remoteObjectName, String localFileName ) throws S3OperationException,
                                                                      IllegalArgumentException {

    AmazonS3 s3Client = getClient();

    localFileName = IoUtils.normalizeFilePath(localFileName);
    String localDirName = IoUtils.getFilePath(localFileName);
    String localFileOnlyName = IoUtils.getFileName(localFileName);
    File localDir = new File(localDirName);
    if (localDir.exists()) {
        if (localDir.isFile()) {
            throw new IllegalArgumentException("Could not create file " + localFileOnlyName + " into existing file "
                                               + localDirName);
        }
        // else dir exists
    } else {
        LOG.debug("Creating target directory path " + localDirName);
        if (!localDir.mkdirs()) {
            throw new S3OperationException("Could not create local directory path '" + localDirName
                                           + "' for local file specified '" + localFileName + "'");
        }
    }

    S3Object obj = s3Client.getObject(bucketName, remoteObjectName);
    try (BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(new File(localFileName)));
            S3ObjectInputStream s3is = obj.getObjectContent();) {
        byte[] readBuffArr = new byte[4096];
        int readBytes = 0;
        while ( (readBytes = s3is.read(readBuffArr)) >= 0) {
            bos.write(readBuffArr, 0, readBytes);
        }
    } catch (Exception e) {
        handleExeption(e, "Error while downloading object " + remoteObjectName + " to local file " + localFileName
                          + ". If error persists check your endpoint, credentials and permissions.");
    }
    LOG.info("S3 object '" + remoteObjectName + "; is downloaded successfully from bucket '" + bucketName
             + "' to file " + localFileName);
}

/**
 * Adjusts the retrieved S3Object so that the object contents contain only the range of bytes
 * desired by the user.  Since encrypted contents can only be retrieved in CIPHER_BLOCK_SIZE
 * (16 bytes) chunks, the S3Object potentially contains more bytes than desired, so this method
 * adjusts the contents range.
 *
 * @param s3object
 *      The S3Object retrieved from S3 that could possibly contain more bytes than desired
 *      by the user.
 * @param range
 *      A two-element array of longs corresponding to the start and finish (inclusive) of a desired
 *      range of bytes.
 * @param instruction
 *      Instruction file in JSON or null if no instruction file is involved
 * @return
 *      The S3Object with adjusted object contents containing only the range desired by the user.
 *      If the range specified is invalid, then the S3Object is returned without any modifications.
 */
protected final S3ObjectWrapper adjustToDesiredRange(S3ObjectWrapper s3object,
        long[] range, Map<String,String> instruction) {
    if (range == null)
        return s3object;
    // Figure out the original encryption scheme used, which can be
    // different from the crypto scheme used for decryption.
    ContentCryptoScheme encryptionScheme = s3object.encryptionSchemeOf(instruction);
    // range get on data encrypted using AES_GCM
    final long instanceLen = s3object.getObjectMetadata().getInstanceLength();
    final long maxOffset = instanceLen - encryptionScheme.getTagLengthInBits() / 8 - 1;
    if (range[1] > maxOffset) {
        range[1] = maxOffset;
        if (range[0] > range[1]) {
            // Return empty content
            // First let's close the existing input stream to avoid resource
            // leakage
            closeQuietly(s3object.getObjectContent(), log);
            s3object.setObjectContent(new ByteArrayInputStream(new byte[0]));
            return s3object;
        }
    }
    if (range[0] > range[1]) {
        // Make no modifications if range is invalid.
        return s3object;
    }
    try {
        S3ObjectInputStream objectContent = s3object.getObjectContent();
        InputStream adjustedRangeContents = new AdjustedRangeInputStream(objectContent, range[0], range[1]);
        s3object.setObjectContent(new S3ObjectInputStream(adjustedRangeContents, objectContent.getHttpRequest()));
        return s3object;
    } catch (IOException e) {
        throw new SdkClientException("Error adjusting output to desired byte range: " + e.getMessage());
    }
}

/**
 * Returns an updated object where the object content input stream contains the decrypted contents.
 *
 * @param wrapper
 *      The object whose contents are to be decrypted.
 * @param cekMaterial
 *      The instruction that will be used to decrypt the object data.
 * @return
 *      The updated object where the object content input stream contains the decrypted contents.
 */
private S3ObjectWrapper decrypt(S3ObjectWrapper wrapper,
        ContentCryptoMaterial cekMaterial, long[] range) {
    S3ObjectInputStream objectContent = wrapper.getObjectContent();
    wrapper.setObjectContent(new S3ObjectInputStream(
            new CipherLiteInputStream(objectContent,
                cekMaterial.getCipherLite(),
                DEFAULT_BUFFER_SIZE),
                objectContent.getHttpRequest()));
    return wrapper;
}

/**
 * @see com.amazonaws.http.HttpResponseHandler#handle(com.amazonaws.http.HttpResponse)
 */
public AmazonWebServiceResponse<S3Object> handle(HttpResponse response) throws Exception {
    /*
     * TODO: It'd be nice to set the bucket name and key here, but the information isn't easy to
     * pull out of the response/request currently.
     */
    S3Object object = new S3Object();
    AmazonWebServiceResponse<S3Object> awsResponse = parseResponseMetadata(response);
    if (response.getHeaders().get(Headers.REDIRECT_LOCATION) != null) {
        object.setRedirectLocation(response.getHeaders().get(Headers.REDIRECT_LOCATION));
    }
    // If the requester is charged when downloading a object from an
    // Requester Pays bucket, then this header is set.
    if (response.getHeaders().get(Headers.REQUESTER_CHARGED_HEADER) != null) {
        object.setRequesterCharged(true);
    }

    if (response.getHeaders().get(Headers.S3_TAGGING_COUNT) != null) {
        object.setTaggingCount(Integer.parseInt(response.getHeaders().get(Headers.S3_TAGGING_COUNT)));
    }

    ObjectMetadata metadata = object.getObjectMetadata();
    populateObjectMetadata(response, metadata);

    object.setObjectContent(new S3ObjectInputStream(abortableIs(response), response.getHttpRequest()));

    awsResponse.setResult(object);
    return awsResponse;
}

@Test
public void passwordUsesTheS3Bucket() throws Exception {
    S3Object mockS3Object = mock(S3Object.class);
    AmazonS3Client mockClient = mock(AmazonS3Client.class);
    when(mockClientBuilder.build()).thenReturn(mockClient);
    when(mockClient.getObject(any(GetObjectRequest.class))).thenReturn(mockS3Object);
    AWSKMSClient mockKmsClient = mock(AWSKMSClient.class);
    when(mockKmsClientBuilder.build()).thenReturn(mockKmsClient);

    S3ObjectInputStream mockS3ObjectInputStream = mock(S3ObjectInputStream.class);
    when(mockS3Object.getObjectContent()).thenReturn(mockS3ObjectInputStream);
    when(mockS3ObjectInputStream.read(new byte[anyInt()], anyInt(), anyByte()))
            .thenAnswer(new WriteBufferAnswer("encryptedPassword".getBytes()))
            .thenReturn(-1);

    DecryptResult result = new DecryptResult();
    CharsetEncoder charsetEncoder = Charset.forName("UTF-8").newEncoder();
    result.setPlaintext(charsetEncoder.encode(CharBuffer.wrap("password")));
    when(mockKmsClient.decrypt(any(DecryptRequest.class))).thenReturn(result);

    Secret secret = test.getPassword();

    // have we got the expected password
    assertThat(secret.getPlainText()).isEqualTo("password");

    // have we used the bucket
    ArgumentCaptor<GetObjectRequest> capturedObjectRequest = ArgumentCaptor.forClass(GetObjectRequest.class);
    verify(mockClient).getObject(capturedObjectRequest.capture());
    assertThat(capturedObjectRequest.getValue().getBucketName()).isEqualTo("bucketUri");
    assertThat(capturedObjectRequest.getValue().getS3ObjectId().getKey()).isEqualTo("/bucketPath");

    // have we used kms to decrypt
    ArgumentCaptor<DecryptRequest> capturedDecryptRequest = ArgumentCaptor.forClass(DecryptRequest.class);
    verify(mockKmsClient).decrypt(capturedDecryptRequest.capture());
    assertThat(capturedDecryptRequest.getValue().getEncryptionContext()).containsEntry("someEncryptContextKey", "kmsEncryptContextValue");
    ByteBuffer ciphertextBlob = capturedDecryptRequest.getValue().getCiphertextBlob();
    assertThat(new String(Charset.forName("UTF-8").decode(ciphertextBlob).array())).isEqualTo("encryptedPassword");

}

@Test
public void closesIfIOExceptionWhileReading() throws Exception {
    AmazonS3Client mockClient = mock(AmazonS3Client.class);
    when(mockClientBuilder.build()).thenReturn(mockClient);

    AWSKMSClient mockKmsClient = mock(AWSKMSClient.class);
    when(mockKmsClientBuilder.build()).thenReturn(mockKmsClient);

    S3Object mockS3Object = mock(S3Object.class);
    when(mockClient.getObject(any(GetObjectRequest.class))).thenReturn(mockS3Object);
    S3ObjectInputStream mockS3ObjectInputStream = mock(S3ObjectInputStream.class);
    when(mockS3Object.getObjectContent()).thenReturn(mockS3ObjectInputStream);
    when(mockS3ObjectInputStream.read(new byte[anyInt()], anyInt(), anyByte()))
            .thenAnswer(new WriteBufferAnswer("encryptedPassword".getBytes()))
            .thenThrow(new IOException("something went wrong"))
            .thenReturn(-1);

    DecryptResult result = new DecryptResult();
    CharsetEncoder charsetEncoder = Charset.forName("UTF-8").newEncoder();
    result.setPlaintext(charsetEncoder.encode(CharBuffer.wrap("password")));
    when(mockKmsClient.decrypt(any(DecryptRequest.class))).thenReturn(result);

    Secret secret = null;
    try {
        secret = test.getPassword();
        TestCase.fail("should have thrown exception");
    } catch (AwsBucketReadingException e) {
        assertThat(e.getCause()).isInstanceOf(IOException.class);
    }

    // have we used the bucket
    ArgumentCaptor<GetObjectRequest> capturedObjectRequest = ArgumentCaptor.forClass(GetObjectRequest.class);
    verify(mockClient).getObject(capturedObjectRequest.capture());
    assertThat(capturedObjectRequest.getValue().getBucketName()).isEqualTo("bucketUri");
    assertThat(capturedObjectRequest.getValue().getS3ObjectId().getKey()).isEqualTo("/bucketPath");

    // and we have closed it even if there was an exception
    verify(mockS3Object).close();
}

public String getString(String bucket, String key) {
    GetObjectRequest request = new GetObjectRequest(bucket, key);
    S3Object response = client.getObject(request);
    try (S3ObjectInputStream is = response.getObjectContent()) {
        return CharStreams.toString(new InputStreamReader(is, Charsets.UTF_8));
    } catch (IOException ex) {
        throw new RuntimeException(ex);
    }
}

/**
 * Proccess the log from s3 and breaks it down to a list of events
 * @param stream The stream from s3
 * @return a list of CloudFrontLogEvent to be processed by the processors
 * @throws IOException
 */
protected List<CloudFrontLogEvent> ingestLogStream(S3ObjectInputStream stream) throws IOException {
    List<CloudFrontLogEvent> logEvents = new LinkedList<>();

    try {
        GZIPInputStream gzipInputStream = new GZIPInputStream(stream);
        Reader decoder = new InputStreamReader(gzipInputStream);
        BufferedReader bufferedReader = new BufferedReader(decoder);

        String request;
        while ((request = bufferedReader.readLine()) != null) {
            // ignore comment lines
            if (request.startsWith("#")) {
                if (request.contains("Version")) {
                    // This lambda was written for V1 log format lets explode if the version ever gets bumped
                    assert request.contains("Version: 1.0");
                }
                continue;
            }
            logEvents.add(new CloudFrontLogEvent(request));
        }
    } finally {
        if (stream != null) {
            stream.close();
        }
    }

    return logEvents;
}

private byte[] readBase64BucketData(S3Object object) throws IOException {
    try (S3ObjectInputStream is = object.getObjectContent();
         ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
        byte[] buf = new byte[8192];
        int bytesRead = is.read(buf);
        while (bytesRead > 0) {
            baos.write(buf, 0, bytesRead);
            bytesRead = is.read(buf);
        }
        return Base64.getDecoder().decode(baos.toByteArray());
    }
}

public String getString(String bucket, String key) {
    GetObjectRequest request = new GetObjectRequest(bucket, key);
    S3Object response = client.getObject(request);
    try (S3ObjectInputStream is = response.getObjectContent()) {
        return CharStreams.toString(new InputStreamReader(is, Charsets.UTF_8));
    } catch (IOException ex) {
        throw new RuntimeException(ex);
    }
}

String getDecryptedData(final String bucketName, final String keyName) {

    String keyValue = "";
    S3Object s3Object = s3.getObject(bucketName, keyName);

    if (LOG.isDebugEnabled()) {
        LOG.debug("retrieving appName {}, key {}", bucketName, keyName);
    }

    if (null == s3Object) {
        LOG.error("error retrieving key {}, from bucket {}", keyName, bucketName);
        return keyValue;
    }

    try (S3ObjectInputStream s3InputStream = s3Object.getObjectContent(); 
            ByteArrayOutputStream result = new ByteArrayOutputStream();) {

        byte[] buffer = new byte[1024];
        int length;
        while ((length = s3InputStream.read(buffer)) != -1) {
            result.write(buffer, 0, length);
        }

        // if key should be decrypted, do so with KMS

        if (kmsDecrypt) {
            DecryptRequest req = new DecryptRequest().withCiphertextBlob(ByteBuffer.wrap(result.toByteArray()));
            ByteBuffer plainText = kms.decrypt(req).getPlaintext();
            keyValue = new String(plainText.array());
        } else {
            keyValue = result.toString();
        }

    } catch (IOException e) {
        LOG.error("error getting application secret.", e);
    }

    return keyValue.trim();
}

/**
 * Stream an {@link S3Object} object and process each line with the
 * processor.
 * 
 * @param object
 *            S3Object to download and process
 * @return number of events processed
 * @throws IOException
 *             if unable to stream the object
 */
private int streamObject(@Nonnull final S3Object object)
        throws IOException {

    final AtomicInteger eventCount = new AtomicInteger(0);
    try (S3ObjectInputStream input = object.getObjectContent()) {

        final BufferedReader reader;
        if (AmazonS3Downloader.isGZipped(object)) {
            reader = new BufferedReader(new InputStreamReader(
                    new StreamingGZIPInputStream(input),
                    StandardCharsets.UTF_8));
        } else {
            reader = new BufferedReader(
                    new InputStreamReader(input, StandardCharsets.UTF_8));
        }

        // failed will be true if we did not successfully broadcast all
        // of the events because of no consumers
        final boolean failed = reader.lines()
                .peek(event -> eventCount.incrementAndGet())
                .anyMatch(broadcaster::test);

        if (failed) {
            // abort the current S3 download
            input.abort();
            LOGGER.error(
                    "Partial events broadcast ({} sent) from key: {}/{}",
                    eventCount.get(), object.getBucketName(),
                    object.getKey());
            throw new IOException("aborting download");
        }
    }
    return eventCount.get();
}

@Test
public void testProcessSNS() throws Exception {
    final HttpRequestBase request = mock(HttpRequestBase.class);
    final S3ObjectInputStream stream = new S3ObjectInputStream(Resources
            .asByteSource(
                    Resources.getResource("fixtures/s3_object.txt.gz"))
            .openStream(), request);

    final ObjectMetadata metadata = new ObjectMetadata();
    metadata.setContentEncoding("gzip");
    final S3Object object = new S3Object();
    object.setObjectMetadata(metadata);
    object.setObjectContent(stream);

    when(broadcaster.isEmpty()).thenReturn(false);
    when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object);

    message.setBody(
            FixtureHelpers.fixture("fixtures/sns_notification.json"));
    final boolean actual = processor.test(message);

    verify(broadcaster, times(2)).isEmpty();
    verify(broadcaster, times(10)).test(anyString());
    verify(s3).fetch(any(S3EventNotificationRecord.class));
    verify(request, never()).abort();
    assertThat(actual).isTrue();
}

@Test
public void testProcessNoConnectionsDuringDownload() throws Exception {
    final HttpRequestBase request = mock(HttpRequestBase.class);
    final S3ObjectInputStream stream = new S3ObjectInputStream(Resources
            .asByteSource(
                    Resources.getResource("fixtures/s3_object.txt.gz"))
            .openStream(), request);

    final ObjectMetadata metadata = new ObjectMetadata();
    metadata.setContentEncoding("gzip");
    final S3Object object = new S3Object();
    object.setObjectMetadata(metadata);
    object.setObjectContent(stream);

    when(broadcaster.test(anyString())).thenReturn(false, false, false,
            false, true);
    when(s3.fetch(any(S3EventNotificationRecord.class))).thenReturn(object);

    message.setBody(
            FixtureHelpers.fixture("fixtures/sns_notification.json"));
    final boolean actual = processor.test(message);

    verify(broadcaster, times(2)).isEmpty();
    verify(broadcaster, times(5)).test(anyString());
    verify(s3).fetch(any(S3EventNotificationRecord.class));
    verify(request).abort();
    assertThat(actual).isFalse();
}

@Override
public StorageObject open(String key)
    throws StorageFileNotFoundException
{
    checkArgument(key != null, "key is null");

    String errorMessage = "opening file bucket " + bucket + " key " + key;
    GetObjectRequest req = new GetObjectRequest(bucket, key);

    S3Object obj = getWithRetry(errorMessage, () -> client.getObject(req));

    final long actualSize = obj.getObjectMetadata().getContentLength();

    // override close to call abort instead because close skips all remaining bytes so that
    // s3 client can reuse the TCP connection. but close of a fully opened file is occasionally
    // used to skip remaing work (e.g. finally block when exception is thrown). Unlike openRange,
    // performance impact could be significantly large.
    InputStream stream = overrideCloseToAbort(obj.getObjectContent());

    InputStream resumable = new ResumableInputStream(stream, (offset, closedCause) -> {
            try {
                S3ObjectInputStream raw = getWithRetry(errorMessage, () -> {
                        req.setRange(offset, actualSize - offset - 1);
                        return client.getObject(req);
                    })
                .getObjectContent();
                return overrideCloseToAbort(raw);
            }
            catch (StorageFileNotFoundException ex) {
                throw new IOException(ex);
            }
        });

    return new StorageObject(resumable, actualSize);
}

private InputStream overrideCloseToAbort(final S3ObjectInputStream raw)
{
    return new FilterInputStream(raw)
    {
        @Override
        public void close() throws IOException
        {
            raw.abort();
        }
    };
}

@Test
public void testCreateObject() throws Exception {
  AmazonS3ExecutorConfig config = getConfig();
  config.taskConfig.taskType = TaskType.CREATE_NEW_OBJECT;
  config.taskConfig.content = "${record:value('/content')}";

  AmazonS3Executor executor = new AmazonS3Executor(config);
  TargetRunner runner = new TargetRunner.Builder(AmazonS3DExecutor.class, executor)
    .build();
  runner.runInit();

  try {
    runner.runWrite(ImmutableList.of(getTestRecord()));

    //Make sure the prefix is empty
    ObjectListing objectListing = s3client.listObjects(BUCKET_NAME, objectName);
    Assert.assertEquals(1, objectListing.getObjectSummaries().size());

    S3Object object = s3client.getObject(BUCKET_NAME, objectName);
    S3ObjectInputStream objectContent = object.getObjectContent();

    List<String> stringList = IOUtils.readLines(objectContent);
    Assert.assertEquals(1, stringList.size());
    Assert.assertEquals("Secret", stringList.get(0));
  } finally {
    runner.runDestroy();
  }
}

@Test
public void testWriteTextDataWithCompression() throws Exception {

  String prefix = "testWriteTextDataWithCompression";
  String suffix = "";
  AmazonS3Target amazonS3Target = createS3targetWithTextData(prefix, true, suffix);
  TargetRunner targetRunner = new TargetRunner.Builder(AmazonS3DTarget.class, amazonS3Target).build();
  targetRunner.runInit();

  List<Record> logRecords = TestUtil.createStringRecords(BUCKET_NAME);

  //Make sure the prefix is empty
  ObjectListing objectListing = s3client.listObjects(BUCKET_NAME, prefix);
  Assert.assertTrue(objectListing.getObjectSummaries().isEmpty());

  targetRunner.runWrite(logRecords);
  targetRunner.runDestroy();

  //check that prefix contains 1 file
  objectListing = s3client.listObjects(BUCKET_NAME, prefix);
  Assert.assertEquals(1, objectListing.getObjectSummaries().size());
  S3ObjectSummary objectSummary = objectListing.getObjectSummaries().get(0);

  //get contents of file and check data - should have 9 lines
  S3Object object = s3client.getObject(BUCKET_NAME, objectSummary.getKey());
  S3ObjectInputStream objectContent = object.getObjectContent();

  Assert.assertTrue(object.getKey().endsWith(".gz"));

  List<String> stringList = IOUtils.readLines(new GZIPInputStream(objectContent));
  Assert.assertEquals(9, stringList.size());
  for(int i = 0 ; i < 9; i++) {
    Assert.assertEquals(TestUtil.TEST_STRING + i, stringList.get(i));
  }
}