import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Test
public void testWriterWithExistingStream() throws Exception
{
    initialize("TestKinesisAppender/testWriterWithExistingStream.properties");

    MockKinesisClient mockClient = new MockKinesisClient();

    appender.setThreadFactory(new DefaultThreadFactory());
    appender.setWriterFactory(mockClient.newWriterFactory());

    logger.debug("example message");
    mockClient.allowWriterThread();

    assertEquals("describeStream: invocation count",        1,          mockClient.describeStreamInvocationCount);
    assertEquals("describeStream: stream name",             "argle",    mockClient.describeStreamStreamName);
    assertEquals("createStream: invocation count",          0,          mockClient.createStreamInvocationCount);
    assertEquals("putRecords: invocation count",            1,          mockClient.putRecordsInvocationCount);
    assertEquals("putRecords: source record count",         1,          mockClient.putRecordsSourceRecords.size());
    assertEquals("putRecords: source record partition key", "bargle",   mockClient.putRecordsSourceRecords.get(0).getPartitionKey());
    assertEquals("putRecords: source record content",       "example message\n",
                                                            new String(
                                                                BinaryUtils.copyAllBytesFrom(mockClient.putRecordsSourceRecords.get(0).getData()),
                                                                "UTF-8"));
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
/**
 * Step 2 of the AWS Signature version 4 calculation. Refer to
 * http://docs.aws
 * .amazon.com/general/latest/gr/sigv4-create-string-to-sign.html.
 */
protected String createStringToSign(String canonicalRequest,
        AWS4SignerRequestParams signerParams) {

    final StringBuilder stringToSignBuilder = new StringBuilder(
            signerParams.getSigningAlgorithm());
    stringToSignBuilder.append(LINE_SEPARATOR)
            .append(signerParams.getFormattedSigningDateTime())
            .append(LINE_SEPARATOR)
            .append(signerParams.getScope())
            .append(LINE_SEPARATOR)
            .append(BinaryUtils.toHex(hash(canonicalRequest)));

    final String stringToSign = stringToSignBuilder.toString();

    if (log.isDebugEnabled())
        log.debug("AWS4 String to Sign: '\"" + stringToSign + "\"");

    return stringToSign;
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
/**
 * Creates the authorization header to be included in the request.
 */
private String buildAuthorizationHeader(SignableRequest<?> request,
        byte[] signature, AWSCredentials credentials,
        AWS4SignerRequestParams signerParams) {
    final String signingCredentials = credentials.getAWSAccessKeyId() + "/"
            + signerParams.getScope();

    final String credential = "Credential="
            + signingCredentials;
    final String signerHeaders = "SignedHeaders="
            + getSignedHeadersString(request);
    final String signatureHeader = "Signature="
            + BinaryUtils.toHex(signature);

    final StringBuilder authHeaderBuilder = new StringBuilder();

    authHeaderBuilder.append(AWS4_SIGNING_ALGORITHM)
                     .append(" ")
                     .append(credential)
                     .append(", ")
                     .append(signerHeaders)
                     .append(", ")
                     .append(signatureHeader);

    return authHeaderBuilder.toString();
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private void finishSimpleUpload() {
    ObjectMetadata objectMetadata = new ObjectMetadata();
    objectMetadata.setContentLength(this.currentOutputStream.size());

    byte[] content = this.currentOutputStream.toByteArray();
    try {
        MessageDigest messageDigest = MessageDigest.getInstance("MD5");
        String md5Digest = BinaryUtils.toBase64(messageDigest.digest(content));
        objectMetadata.setContentMD5(md5Digest);
    } catch (NoSuchAlgorithmException e) {
        throw new IllegalStateException("MessageDigest could not be initialized because it uses an unknown algorithm", e);
    }

    SimpleStorageResource.this.amazonS3.putObject(SimpleStorageResource.this.bucketName, SimpleStorageResource.this.objectName,
            new ByteArrayInputStream(content), objectMetadata);

    //Release the memory early
    this.currentOutputStream = null;
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Test
public void testWriterWithNewStream() throws Exception
{
    initialize("TestKinesisAppender/testWriterWithNewStream.properties");

    MockKinesisClient mockClient = new MockKinesisClient();

    appender.setThreadFactory(new DefaultThreadFactory());
    appender.setWriterFactory(mockClient.newWriterFactory());

    logger.debug("example message");
    mockClient.allowWriterThread();

    // writer calls describeStream once to see if stream exists, a second time
    // to verify that it's active -- could perhaps combine those calls?

    assertEquals("describeStream: invocation count",        3,          mockClient.describeStreamInvocationCount);
    assertEquals("describeStream: stream name",             "foo",      mockClient.describeStreamStreamName);
    assertEquals("createStream: invocation count",          1,          mockClient.createStreamInvocationCount);
    assertEquals("createStream: stream name",               "foo",      mockClient.createStreamStreamName);
    assertEquals("putRecords: invocation count",            1,          mockClient.putRecordsInvocationCount);
    assertEquals("putRecords: source record count",         1,          mockClient.putRecordsSourceRecords.size());
    assertEquals("putRecords: source record partition key", "bar",      mockClient.putRecordsSourceRecords.get(0).getPartitionKey());
    assertEquals("putRecords: source record content",       "example message\n",
                                                            new String(
                                                                BinaryUtils.copyAllBytesFrom(mockClient.putRecordsSourceRecords.get(0).getData()),
                                                                "UTF-8"));
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Test
public void testStaticClientFactory() throws Exception
{
    initialize("TestKinesisAppender/testStaticClientFactory.properties");
    appender.setThreadFactory(new DefaultThreadFactory());
    appender.setWriterFactory(new KinesisWriterFactory());

    // first message triggers writer creation

    logger.debug("example message");
    waitForInitialization();
    AbstractLogWriter writer = (AbstractLogWriter)appender.getWriter();

    assertNotNull("factory was called to create client", staticFactoryMock);
    assertEquals("no initialization errors",             "",
                                                         writer.getInitializationMessage());
    assertEquals("factory method called",                "com.kdgregory.log4j.aws.TestKinesisAppender.createMockClient",
                                                         writer.getClientFactoryUsed());

    // although we should be happy at this point, we'll actually verify that the
    // message got written; assertions copied from testWriterWithExistingStream()

    staticFactoryMock.allowWriterThread();

    assertEquals("describeStream: invocation count",        1,          staticFactoryMock.describeStreamInvocationCount);
    assertEquals("describeStream: stream name",             "argle",    staticFactoryMock.describeStreamStreamName);
    assertEquals("createStream: invocation count",          0,          staticFactoryMock.createStreamInvocationCount);
    assertEquals("putRecords: invocation count",            1,          staticFactoryMock.putRecordsInvocationCount);
    assertEquals("putRecords: source record count",         1,          staticFactoryMock.putRecordsSourceRecords.size());
    assertEquals("putRecords: source record partition key", "bargle",   staticFactoryMock.putRecordsSourceRecords.get(0).getPartitionKey());
    assertEquals("putRecords: source record content",       "example message\n",
                                                            new String(
                                                                BinaryUtils.copyAllBytesFrom(staticFactoryMock.putRecordsSourceRecords.get(0).getData()),
                                                                "UTF-8"));
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
public RetrievedRecord(String shardId, Record record)
throws Exception
{
    this.shardId = shardId;
    this.partitionKey = record.getPartitionKey();
    this.message = new String(BinaryUtils.copyAllBytesFrom(record.getData()), "UTF-8").trim();
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private String getObjectMD5(byte[] objectData) {
  byte[] expectedMd5 = null;
  try {
    expectedMd5 = Md5Utils.computeMD5Hash(objectData);
  } catch (Exception e) {}
  return BinaryUtils.toHex(expectedMd5);
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private String getAwsMessageMD5(String message) {
  byte[] expectedMd5 = null;
  try {
    expectedMd5 = Md5Utils.computeMD5Hash(message.getBytes(UTF8));
  } catch (Exception e) {}
  return BinaryUtils.toHex(expectedMd5);
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
/**
 * Binary data should be placed as is, directly into the content.
 */
private void marshallBinaryPayload(Object val) {
    if (val instanceof ByteBuffer) {
        request.setContent(BinaryUtils.toStream((ByteBuffer) val));
    } else if (val instanceof InputStream) {
        request.setContent((InputStream) val);
    }
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Override
public StructuredJsonGenerator writeValue(ByteBuffer bytes) {
    try {
        writer.writeBlob(BinaryUtils.copyAllBytesFrom(bytes));
    } catch (IOException e) {
        throw new SdkClientException(e);
    }
    return this;
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Override
public StructuredJsonGenerator writeValue(ByteBuffer bytes) {
    try {
        generator.writeBinary(BinaryUtils.copyBytesFrom(bytes));
    } catch (IOException e) {
        throw new JsonGenerationException(e);
    }
    return this;
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Override
public void presignRequest(SignableRequest<?> request, AWSCredentials credentials,
        Date userSpecifiedExpirationDate) {

    // anonymous credentials, don't sign
    if (isAnonymous(credentials)) {
        return;
    }

    long expirationInSeconds = generateExpirationDate(userSpecifiedExpirationDate);

    addHostHeader(request);

    AWSCredentials sanitizedCredentials = sanitizeCredentials(credentials);
    if (sanitizedCredentials instanceof AWSSessionCredentials) {
        // For SigV4 pre-signing URL, we need to add "X-Amz-Security-Token"
        // as a query string parameter, before constructing the canonical
        // request.
        request.addParameter(X_AMZ_SECURITY_TOKEN,
                ((AWSSessionCredentials) sanitizedCredentials)
                        .getSessionToken());
    }

    final AWS4SignerRequestParams signerRequestParams = new AWS4SignerRequestParams(
            request, overriddenDate, regionName, serviceName,
            AWS4_SIGNING_ALGORITHM);

    // Add the important parameters for v4 signing
    final String timeStamp = signerRequestParams.getFormattedSigningDateTime();

    addPreSignInformationToRequest(request, sanitizedCredentials,
            signerRequestParams, timeStamp, expirationInSeconds);

    final String contentSha256 = calculateContentHashPresign(request);

    final String canonicalRequest = createCanonicalRequest(request,
            contentSha256);

    final String stringToSign = createStringToSign(canonicalRequest,
            signerRequestParams);

    final byte[] signingKey = deriveSigningKey(sanitizedCredentials,
            signerRequestParams);

    final byte[] signature = computeSignature(stringToSign, signingKey,
            signerRequestParams);

    request.addParameter(X_AMZ_SIGNATURE, BinaryUtils.toHex(signature));
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
/**
 * Calculate the hash of the request's payload. Subclass could override this
 * method to provide different values for "x-amz-content-sha256" header or
 * do any other necessary set-ups on the request headers. (e.g. aws-chunked
 * uses a pre-defined header value, and needs to change some headers
 * relating to content-encoding and content-length.)
 */
protected String calculateContentHash(SignableRequest<?> request) {
    InputStream payloadStream = getBinaryRequestPayloadStream(request);
    ReadLimitInfo info = request.getReadLimitInfo();
    payloadStream.mark(info == null ? -1 : info.getReadLimit());
    String contentSha256 = BinaryUtils.toHex(hash(payloadStream));
    try {
        payloadStream.reset();
    } catch (IOException e) {
        throw new SdkClientException(
                "Unable to reset stream after calculating AWS4 signature",
                e);
    }
    return contentSha256;
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private byte[] createSignedChunk(byte[] chunkData) {
    StringBuilder chunkHeader = new StringBuilder();
    // chunk-size
    chunkHeader.append(Integer.toHexString(chunkData.length));
    // sig-extension
    final String chunkStringToSign =
            CHUNK_STRING_TO_SIGN_PREFIX + "\n" +
            dateTime + "\n" +
            keyPath + "\n" +
            priorChunkSignature + "\n" +
            AbstractAWSSigner.EMPTY_STRING_SHA256_HEX + "\n" +
            BinaryUtils.toHex(sha256.digest(chunkData));
    final String chunkSignature =
        BinaryUtils.toHex(aws4Signer.signWithMac(chunkStringToSign, hmacSha256));
    priorChunkSignature = chunkSignature;
    chunkHeader.append(CHUNK_SIGNATURE_HEADER)
               .append(chunkSignature)
               .append(CRLF)
               ;
    try {
        byte[] header = chunkHeader.toString().getBytes(UTF8);
        byte[] trailer = CRLF.getBytes(UTF8);
        byte[] signedChunk = new byte[header.length + chunkData.length + trailer.length];
        System.arraycopy(header, 0, signedChunk, 0, header.length);
        System.arraycopy(chunkData, 0, signedChunk, header.length, chunkData.length);
        System.arraycopy(trailer, 0,
                signedChunk, header.length + chunkData.length,
                trailer.length);
        return signedChunk;
    } catch (Exception e) {
        throw new SdkClientException("Unable to sign the chunked data. " + e.getMessage(), e);
    }
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Override
public void setBucketCrossOriginConfiguration(
        SetBucketCrossOriginConfigurationRequest setBucketCrossOriginConfigurationRequest) {
    setBucketCrossOriginConfigurationRequest = beforeClientExecution(setBucketCrossOriginConfigurationRequest);
    rejectNull(setBucketCrossOriginConfigurationRequest,
            "The set bucket cross origin configuration request object must be specified.");

    String bucketName = setBucketCrossOriginConfigurationRequest.getBucketName();
    BucketCrossOriginConfiguration bucketCrossOriginConfiguration = setBucketCrossOriginConfigurationRequest.getCrossOriginConfiguration();

    rejectNull(bucketName,
            "The bucket name parameter must be specified when setting bucket cross origin configuration.");
    rejectNull(bucketCrossOriginConfiguration,
            "The cross origin configuration parameter must be specified when setting bucket cross origin configuration.");

    Request<SetBucketCrossOriginConfigurationRequest> request = createRequest(bucketName, null, setBucketCrossOriginConfigurationRequest, HttpMethodName.PUT);
    request.addParameter("cors", null);

    byte[] content = new BucketConfigurationXmlFactory().convertToXmlByteArray(bucketCrossOriginConfiguration);
    request.addHeader("Content-Length", String.valueOf(content.length));
    request.addHeader("Content-Type", "application/xml");
    request.setContent(new ByteArrayInputStream(content));
    try {
        byte[] md5 = Md5Utils.computeMD5Hash(content);
        String md5Base64 = BinaryUtils.toBase64(md5);
        request.addHeader("Content-MD5", md5Base64);
    } catch ( Exception e ) {
        throw new SdkClientException("Couldn't compute md5 sum", e);
    }

    invoke(request, voidResponseHandler, bucketName, null);
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
@Override
public void setBucketTaggingConfiguration(
        SetBucketTaggingConfigurationRequest setBucketTaggingConfigurationRequest) {
    setBucketTaggingConfigurationRequest = beforeClientExecution(setBucketTaggingConfigurationRequest);
    rejectNull(setBucketTaggingConfigurationRequest,
            "The set bucket tagging configuration request object must be specified.");

    String bucketName = setBucketTaggingConfigurationRequest.getBucketName();
    BucketTaggingConfiguration bucketTaggingConfiguration = setBucketTaggingConfigurationRequest.getTaggingConfiguration();

    rejectNull(bucketName,
            "The bucket name parameter must be specified when setting bucket tagging configuration.");
    rejectNull(bucketTaggingConfiguration,
            "The tagging configuration parameter must be specified when setting bucket tagging configuration.");

    Request<SetBucketTaggingConfigurationRequest> request = createRequest(bucketName, null, setBucketTaggingConfigurationRequest, HttpMethodName.PUT);
    request.addParameter("tagging", null);

    byte[] content = new BucketConfigurationXmlFactory().convertToXmlByteArray(bucketTaggingConfiguration);
    request.addHeader("Content-Length", String.valueOf(content.length));
    request.addHeader("Content-Type", "application/xml");
    request.setContent(new ByteArrayInputStream(content));
    try {
        byte[] md5 = Md5Utils.computeMD5Hash(content);
        String md5Base64 = BinaryUtils.toBase64(md5);
        request.addHeader("Content-MD5", md5Base64);
    } catch ( Exception e ) {
        throw new SdkClientException("Couldn't compute md5 sum", e);
    }

    invoke(request, voidResponseHandler, bucketName, null);
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private void setContent(Request<?> request, byte[] content, String contentType, boolean setMd5) {
    request.setContent(new ByteArrayInputStream(content));
    request.addHeader("Content-Length", Integer.toString(content.length));
    request.addHeader("Content-Type", contentType);
    if (setMd5) {
        try {
            byte[] md5 = Md5Utils.computeMD5Hash(content);
            String md5Base64 = BinaryUtils.toBase64(md5);
            request.addHeader("Content-MD5", md5Base64);
        } catch ( Exception e ) {
            throw new AmazonClientException("Couldn't compute md5 sum", e);
        }
    }
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
/**
 * If necessary, creates a chunk-encoding wrapper on the request payload.
 */
@Override
protected void processRequestPayload(SignableRequest<?> request, byte[] signature,
        byte[] signingKey, AWS4SignerRequestParams signerRequestParams) {
    if (useChunkEncoding(request)) {
        AwsChunkedEncodingInputStream chunkEncodededStream = new AwsChunkedEncodingInputStream(
                request.getContent(), signingKey,
                signerRequestParams.getFormattedSigningDateTime(),
                signerRequestParams.getScope(),
                BinaryUtils.toHex(signature), this);
        request.setContent(chunkEncodededStream);
    }
}
 

import com.amazonaws.util.BinaryUtils; //导入依赖的package包/类
private void uploadContents(String bucket, String key, byte[] contents)
        throws IOException {
    int failures = 0;
    boolean uploaded = false;
    while (!uploaded) {
        ObjectMetadata objectMetadata = new ObjectMetadata();
        objectMetadata.setContentType(MediaType.TEXT_PLAIN);
        objectMetadata.setContentLength(contents.length);
        objectMetadata.setContentMD5(BinaryUtils.toBase64(Hashing.md5().hashBytes(contents).asBytes()));

        try {
            _amazonS3.putObject(
                    new PutObjectRequest(bucket, key, new ByteArrayInputStream(contents), objectMetadata));
            uploaded = true;
        } catch (AmazonClientException e) {
            if (++failures == MAX_RETRIES) {
                throw new IOException(e);
            }
            try {
                Thread.sleep(_retryDelay.getMillis());
            } catch (InterruptedException e2) {
                // Stop retrying and propagate the original exception
                throw new IOException(e);
            }
        }
    }
}