Java 类com.amazonaws.util.StringInputStream 实例源码

/**
 * This method converts a string (in n-triples format) into a Jena triple
 *
 * @param string containing single n-triples triple
 * @return Triple
 */
public static Triple toTriple(final String string) {
    // Create Jena model
    Model inputModel = createDefaultModel();
    try {
        // Load model with arg string (expecting n-triples)
        inputModel = inputModel.read(new StringInputStream(string), null, strLangNTriples);
    } catch (UnsupportedEncodingException e) {
        throw new RuntimeException(e);
    }

    // Since there is only one statement, get it
    final Statement stmt = inputModel.listStatements().nextStatement();

    // Return the Jena triple which the statement represents
    return stmt.asTriple();
}

@Test
public void testStoreValue()
    throws BinaryStoreException, UnsupportedEncodingException {
    String valueToStore = "value-to-store";

    expect(s3Client.doesObjectExist(eq(BUCKET), isA(String.class))).andReturn(false);
    Capture<ObjectMetadata> objMetaCapture = Capture.newInstance();
    expect(s3Client.putObject(eq(BUCKET), isA(String.class),
                              isA(InputStream.class), capture(objMetaCapture)))
        .andReturn(null);

    replayAll();

    s3BinaryStore.storeValue(new StringInputStream(valueToStore), false);
    ObjectMetadata objMeta = objMetaCapture.getValue();
    assertEquals(String.valueOf(false),
                 objMeta.getUserMetadata().get(s3BinaryStore.UNUSED_KEY));
}

@Test
public void testStoreValueExisting() throws BinaryStoreException, IOException {
    String valueToStore = "value-to-store";

    expect(s3Client.doesObjectExist(eq(BUCKET), isA(String.class))).andReturn(true);
    expect(s3Client.getObjectMetadata(eq(BUCKET), isA(String.class)))
        .andReturn(new ObjectMetadata());
    ObjectMetadata objMeta = new ObjectMetadata();
    Map<String, String> userMeta = new HashMap<>();
    userMeta.put(s3BinaryStore.UNUSED_KEY, String.valueOf(true));
    objMeta.setUserMetadata(userMeta);
    Capture<CopyObjectRequest> copyRequestCapture = Capture.newInstance();
    expect(s3Client.copyObject(capture(copyRequestCapture))).andReturn(null);

    replayAll();

    s3BinaryStore.storeValue(new StringInputStream(valueToStore), true);
    ObjectMetadata newObjMeta = copyRequestCapture.getValue().getNewObjectMetadata();
    assertEquals(String.valueOf(true),
                 newObjMeta.getUserMetadata().get(s3BinaryStore.UNUSED_KEY));
}

public static void main(String[] args) throws Exception {
    // create the AWS S3 Client
    AmazonS3 s3 = AWSS3Factory.getS3Client();

    // retrieve the object key and new object value from user
    System.out.println( "Enter the object key:" );
    String key = new BufferedReader( new InputStreamReader( System.in ) ).readLine();
    System.out.println( "Enter new object content:" );
    String content = new BufferedReader( new InputStreamReader( System.in ) ).readLine();

    // update the object in the demo bucket
    PutObjectRequest updateRequest = new PutObjectRequest(AWSS3Factory.S3_BUCKET, key,
            new StringInputStream(content), null);

    s3.putObject(updateRequest);

    // print out object key/value for validation
    System.out.println( String.format("update object [%s/%s] with new content: [%s]",
            AWSS3Factory.S3_BUCKET, key, content));
}

@Override
public S3Object getObject(String bucketName, String key) throws AmazonClientException, AmazonServiceException {
    assertEquals("mycamelbucket", bucketName);
    assertEquals("key", key);

    S3Object s3Object = new S3Object();
    s3Object.setBucketName(bucketName);
    s3Object.setKey(key);
    try {
        s3Object.setObjectContent(new StringInputStream("Camel rocks!"));
    } catch (UnsupportedEncodingException e) {
        // noop
    }

    return s3Object;
}

@Override
public S3Object getObject(String bucketName, String key) throws AmazonClientException, AmazonServiceException {
    assertEquals("mycamelbucket", bucketName);
    assertEquals("key", key);

    S3Object s3Object = new S3Object();
    s3Object.setBucketName(bucketName);
    s3Object.setKey(key);
    try {
        s3Object.setObjectContent(new StringInputStream("Camel rocks!"));
    } catch (UnsupportedEncodingException e) {
        // noop
    }

    return s3Object;
}

private List<UploadDocumentsRequest> createUploadDocumentsRequest(List<Document> docs) {
    List<String> parts = chunkedJson(docs);
    List<UploadDocumentsRequest> uploadDocumentRequests = new ArrayList<>(parts.size());
    for (String part : parts) {
        try (StringInputStream documents = new StringInputStream(part)) {
            UploadDocumentsRequest uploadDocumentsRequest = new UploadDocumentsRequest(). //
                    withDocuments(documents). //
                    withContentLength((long) part.length()). //
                    withContentType(Applicationjson);
            if (progressListener != null) {
                uploadDocumentsRequest.setGeneralProgressListener(progressListener);
            }
            uploadDocumentRequests.add(uploadDocumentsRequest);
        } catch (IOException e) {
            log.warn("this should never happen", e);
        }
    }
    return uploadDocumentRequests;
}

private Deployment cloneCurrentModel(){
    //need to clone the model
    JsonCodec jsonCodec=new JsonCodec();
    ByteArrayOutputStream baos=new ByteArrayOutputStream();
    jsonCodec.save(currentModel,baos);

    Deployment targetModel=new Deployment();
    try {
        String aString = new String(baos.toByteArray(),"UTF-8");
        InputStream is = new StringInputStream(aString);
        targetModel = (Deployment) jsonCodec.load(is);
    } catch (UnsupportedEncodingException e) {
        e.printStackTrace();
    }
    return targetModel;
}

private void assignContent(Request request, Object representation) {
    String contentString = new JSONObject(representation).toString();

    if (contentString == null) {
        throw new AmazonClientException("Unable to marshall representation to JSON: " + representation);
    }

    try {
        byte[] contentBytes = contentString.getBytes("UTF-8");

        request.setContent(new StringInputStream(contentString));
        request.addHeader("Content-Length", Integer.toString(contentBytes.length));
        request.addHeader("Content-Type", "application/json");
    } catch(Throwable t) {
        throw new AmazonClientException("Unable to marshall request to JSON: " + t.getMessage(), t);
    }
}

public static void storeObject(String contentType,
                               String content,
                               String destBucket,
                               String destKey) throws UnsupportedEncodingException {
    AmazonS3 s3Client = AmazonS3Provider.getS3Client();

    ObjectMetadata metadata = prepareObjectMetadata(contentType, content);

    s3Client.putObject(
            destBucket,
            destKey,
            new StringInputStream(content),
            metadata);
}

@Override
public InputStream adapt(String source) {
    if (source == null) {
        return null;
    }
    try {
        return new StringInputStream(source);
    } catch (UnsupportedEncodingException e) {
        throw new SdkClientException(e);
    }
}

@Test
public void query_params_in_uri_for_post_request_with_payload() throws IOException, URISyntaxException {
    final Request<Object> request = newDefaultRequest(HttpMethodName.POST);
    request.withParameter("foo", "bar");
    final String payload = "dummy string stream";
    request.setContent(new StringInputStream(payload));
    HttpRequestBase requestBase = requestFactory.create(request, settings);
    Assert.assertThat(requestBase, Matchers.instanceOf(HttpPost
            .class));
    Assert.assertEquals("foo=bar", requestBase.getURI().getQuery());
    Assert.assertThat(requestBase, Matchers.instanceOf(HttpPost
            .class));
    Assert.assertEquals(payload, IOUtils.toString(((HttpPost)requestBase).getEntity().getContent()));
}

@Test
public void request_has_default_content_type_set_when_not_explicitly_set() throws IOException,
        URISyntaxException {
    final Request<Object> request = newDefaultRequest(HttpMethodName.POST);
    request.setContent(new StringInputStream("dummy string stream"));
    HttpRequestBase requestBase = requestFactory.create(request, settings);
    assertContentTypeContains("application/x-www-form-urlencoded",
            requestBase.getHeaders(CONTENT_TYPE));
}

@Test
public void apache_request_has_content_type_set_when_not_explicitly_set() throws IOException,
        URISyntaxException {

    final Request<Object> request = newDefaultRequest(HttpMethodName.POST);
    final String testContentype = "testContentType";
    request.addHeader(HttpHeaders.CONTENT_TYPE, testContentype);
    request.setContent(new StringInputStream("dummy string stream"));
    HttpRequestBase requestBase = requestFactory.create(request, settings);
    assertContentTypeContains(testContentype,
            requestBase.getHeaders(CONTENT_TYPE));

}

private static void setEntity(HttpResponse response, String content) {
    try {
        BasicHttpEntity entity = new BasicHttpEntity();
        entity.setContent(new StringInputStream(content));
        response.setEntity(entity);
    } catch (UnsupportedEncodingException e) {
        throw new RuntimeException(e);
    }
}

@Before
public void setup() throws UnsupportedEncodingException {
    MockitoAnnotations.initMocks(this);
    when(errorCodeParser
                 .parseErrorCode((HttpResponse) anyObject(), (JsonContent) anyObject()))
            .thenReturn(ERROR_CODE);

    httpResponse = new HttpResponse(new DefaultRequest<String>(SERVICE_NAME), null);
    httpResponse.setContent(new StringInputStream("{}"));

    responseHandler = new JsonErrorResponseHandler(Arrays.asList(unmarshaller), errorCodeParser,
                                                   JsonErrorMessageParser.DEFAULT_ERROR_MESSAGE_PARSER,
                                                   new JsonFactory());
}

@Test
public void handle_EmptyContent_ReturnsGenericAmazonServiceException() throws Exception {
    httpResponse.setStatusCode(500);
    httpResponse.setContent(new StringInputStream(""));

    AmazonServiceException ase = responseHandler.handle(httpResponse);

    assertNotNull(ase);
}

@SuppressWarnings("rawtypes")
public static Request<?> getSampleRequestWithRepeatableContent(AmazonWebServiceRequest amazonWebServiceRequest) {
    DefaultRequest<?> request = new DefaultRequest(
            amazonWebServiceRequest, "non-existent-service");
    request.setEndpoint(URI.create("http://non-existent-service.amazonaws.com"));
    // StringInputStream#markSupported() returns true
    try {
        request.setContent(new StringInputStream("Some content that could be read for multiple times."));
    } catch (UnsupportedEncodingException e) {
        Assert.fail("Unable to set up the request content");
    }
    return request;
}

@Test
public void testGetInputStream() throws BinaryStoreException, IOException {
    S3Object s3Object = new S3Object();
    s3Object.setObjectContent(new StringInputStream(TEST_CONTENT));
    expect(s3Client.getObject(BUCKET, TEST_KEY)).andReturn(s3Object);

    replayAll();

    InputStream resultStream = s3BinaryStore.getInputStream(new BinaryKey(TEST_KEY));
    assertEquals(TEST_CONTENT, IOUtils.toString(resultStream));
}

public void putManyObjectsWithPrefix(AmazonS3 s3, String key, String content, String prefix, int numberOfObjects)
        throws java.io.IOException {

        for (int i=0; i<numberOfObjects;i++) {
                // create the object in the demo bucket
                s3.putObject(AWSS3Factory.S3_BUCKET, prefix + key + "_" + i, new StringInputStream(content), null);
                // print bucket key/value and content for validation
                System.out.println(String.format("created object [%s/%s] with content: [%s]",
                        AWSS3Factory.S3_BUCKET, key, content));
        }
}

public String createSampleObject(AmazonS3 client, String versionBucket) throws Exception {
    String prefix = "versionPrefix/";

    // create a few versions of the same key
    String key = prefix + "foo", content = "Hello Versions!";
    client.putObject(versionBucket, key, new StringInputStream(content), null);
    client.deleteObject(versionBucket, key);
    client.putObject(versionBucket, key, new StringInputStream(content), null);
    System.out.println("using prefix: " + prefix);
    return prefix;
}

public void testListAndReadVersions(AmazonS3 client, String versionBucket) throws Exception {
        String prefix = "versionPrefix/";
        // turn on versioning first

        // 1. Enable versioning on the bucket.
        BucketVersioningConfiguration configuration =
                new BucketVersioningConfiguration().withStatus("Enabled");

        SetBucketVersioningConfigurationRequest setBucketVersioningConfigurationRequest =
                new SetBucketVersioningConfigurationRequest(versionBucket,configuration);

        client.setBucketVersioningConfiguration(setBucketVersioningConfigurationRequest);

        // 2. Get bucket versioning configuration information.
        BucketVersioningConfiguration conf = client.getBucketVersioningConfiguration(versionBucket);
        System.out.println("bucket versioning configuration status:    " + conf.getStatus());


        /*
        client.setBucketVersioning(versionBucket,
                new VersioningConfiguration().withStatus(VersioningConfiguration.Status.Enabled));
*/

        // create a few versions of the same key
        String key = prefix + "foo", content = "Hello Versions!";
        client.putObject(versionBucket, key, new StringInputStream(content), null);
        client.deleteObject(versionBucket, key);
        client.putObject(versionBucket, key, new StringInputStream(content), null);

        VersionListing result = client.listVersions(versionBucket, prefix);

        List<S3VersionSummary> versions = result.getVersionSummaries();
        for (S3VersionSummary vs : versions) {
            System.out.println("key: " + vs.getKey() + "\t id: " + vs.getVersionId());
        }
    }

public static void main(String[] args) throws Exception {
    // create the AWS S3 Client
    AmazonS3 s3 = AWSS3Factory.getS3Client();

    // retrieve the object key and value from user
    System.out.println( "Enter the object key:" );
    String key = new BufferedReader( new InputStreamReader( System.in ) ).readLine();
    System.out.println( "Enter the object content:" );
    String content = new BufferedReader( new InputStreamReader( System.in ) ).readLine();

    //retrieve the object metadata key and value from user
    System.out.println( "Enter the metadata key:" );
    String metaKey = new BufferedReader( new InputStreamReader( System.in ) ).readLine();
    System.out.println( "Enter the metadata content:" );
    String metaValue = new BufferedReader( new InputStreamReader( System.in ) ).readLine();

    // create the metadata
    ObjectMetadata metadata = new ObjectMetadata();
    metadata.addUserMetadata(metaKey, metaValue);

    // Set the content type for streaming back to a browser.
    metadata.setContentType("text/plain");

    // create the object with the metadata in the demo bucket
    s3.putObject(AWSS3Factory.S3_BUCKET, key, new StringInputStream(content), metadata);

    // print out object key/value and metadata key/value for validation
    System.out.println( String.format("created object [%s/%s] with metadata [%s=%s] and content: [%s]",
            AWSS3Factory.S3_BUCKET, key, metaKey, metaValue, content));
}

@Test
@Override
public void test() throws Exception {
    AmazonS3 amazonS3Client = TestUtils.getPropertyValue(this.s3MessageHandler, "transferManager.s3",
            AmazonS3.class);

    InputStream payload = new StringInputStream("a");
    Message<?> message = MessageBuilder.withPayload(payload)
            .setHeader("key", "myInputStream")
            .build();

    this.channels.input().send(message);

    ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor =
            ArgumentCaptor.forClass(PutObjectRequest.class);
    verify(amazonS3Client, atLeastOnce()).putObject(putObjectRequestArgumentCaptor.capture());

    PutObjectRequest putObjectRequest = putObjectRequestArgumentCaptor.getValue();
    assertThat(putObjectRequest.getBucketName(), equalTo(S3_BUCKET));
    assertThat(putObjectRequest.getKey(), equalTo("myInputStream"));
    assertNull(putObjectRequest.getFile());
    assertNotNull(putObjectRequest.getInputStream());

    ObjectMetadata metadata = putObjectRequest.getMetadata();
    assertThat(metadata.getContentMD5(), equalTo(Md5Utils.md5AsBase64(payload)));
    assertThat(metadata.getContentLength(), equalTo(1L));
    assertThat(metadata.getContentType(), equalTo(MediaType.APPLICATION_JSON_VALUE));
    assertThat(metadata.getContentDisposition(), equalTo("test.json"));
}

private String uploadSummary(Summary sum)throws Exception {
    logger.info("Sending summary file to S3");
    GenericMessage sumM = new GenericMessage(sum);
    Storage resStore = new Storage(Configuration.FILES_BUCKET_NAME);
    String conKey = resStore.putStream("summary_" + _uuid, new StringInputStream(sumM.toXML()));
    return conKey;
}

static public String getExpectedAuthorizationHeader(Request request) throws Exception {
    // create the signable request
    DefaultRequest signableRequest = new DefaultRequest(null, request.getServiceName());
    signableRequest.setEndpoint(new URI("http://" + request.getHost()));
    signableRequest.setResourcePath(request.getUri());
    signableRequest.setHttpMethod(HttpMethodName.valueOf(request.getHttpMethod()));
    signableRequest.setContent(new StringInputStream(request.getBody()));
    if (request.getHeaders() != null)
        signableRequest.setHeaders(request.getHeaders());
    if (request.getQueryParams() != null) {
        Map<String, List<String>> convertedQueryParams = new HashMap<>();
        for (String paramName : request.getQueryParams().keySet()) {
            convertedQueryParams.put(paramName, new ArrayList<>(request.getQueryParams().get(paramName)));
        }
        signableRequest.setParameters(convertedQueryParams);
    }

    /*
       Init the signer class

       Note: Double uri encoding is off simple before the signature does not match the expected signature of the test cases
       if it is enabled.  This was a bit unexpected because AWSElasticsearchClient (AWS SDK Class) enabled double URI encoding
       in the signer by default.  I can only assume that double encoding is needed when accessing the service but not when accessing
       elasticsearch.
     */
    AWS4Signer aws4Signer = new AWS4Signer(false);
    aws4Signer.setServiceName(request.getServiceName());
    aws4Signer.setRegionName(request.getRegion());
    Method method1 = AWS4Signer.class.getDeclaredMethod("setOverrideDate", Date.class);
    method1.setAccessible(true);
    method1.invoke(aws4Signer, request.getDate());
    aws4Signer.sign(signableRequest, request.getCredentialsProvider().getCredentials());

    return (String) signableRequest.getHeaders().get("Authorization");
}

private void writeResponse(Request request, Response response, String file) {
    OutputStream wrappedOutputStream;
    try {
        response.header("Content-Type", "text/html");
        response.status(200);
        wrappedOutputStream = GzipUtils.checkAndWrap(request.raw(), response.raw(), false);
        IOUtils.copy(new StringInputStream(file), wrappedOutputStream);
        wrappedOutputStream.flush();
        wrappedOutputStream.close();
    } catch (IOException e) {
        throw new RuntimeException("Failed to write HTTP response", e);
    }
}

private void setRequestPayload(Request<ListVoicesRequest> request, ListVoicesRequest listVoicesRequest) {
    try {
        StringWriter stringWriter = new StringWriter();
        JSONWriter jsonWriter = new JSONWriter(stringWriter);

        jsonWriter.object();
        if (listVoicesRequest.getVoice() != null) {
            Voice voice = listVoicesRequest.getVoice();

            jsonWriter.key(JSON_KEY_VOICE);
            jsonWriter.object();

            if (voice.getGender() != null) {
                jsonWriter.key(JSON_KEY_GENDER).value(voice.getGender());
            }
            if (voice.getLanguage() != null) {
                jsonWriter.key(JSON_KEY_LANGUAGE).value(voice.getLanguage());
            }
            if (voice.getName() != null) {
                jsonWriter.key(JSON_KEY_NAME).value(voice.getName());
            }

            jsonWriter.endObject();
        }
        jsonWriter.endObject();

        String snippet = stringWriter.toString();
        byte[] content = snippet.getBytes(UTF_8);
        request.setContent(new StringInputStream(snippet));
        request.addHeader("Content-Length", Integer.toString(content.length));
    } catch (Throwable t) {
        throw new AmazonClientException("Unable to marshall request to JSON: " + t.getMessage(), t);
    }
}

@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneViaS3() throws UnsupportedEncodingException {
    final AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
    final AmazonS3Client s3 = Mockito.mock(AmazonS3Client.class);
    final String queueName = "queue";
    final String s3Id = "123";
    Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
    final String receiptHandle = "abc";
    Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any())).thenReturn(
            new ReceiveMessageResult().withMessages(new Message().withReceiptHandle(receiptHandle).withBody(s3Id)));
    final String bucketName = "bucket";
    Mockito.when(s3.doesObjectExist(bucketName, s3Id)).thenAnswer(x -> true);
    final S3Object s3Object = mock(S3Object.class);
    Mockito.when(s3Object.getObjectContent())
            .thenReturn(new S3ObjectInputStream(new StringInputStream("body1"), null));
    final ObjectMetadata om = new ObjectMetadata();
    om.setLastModified(new Date(1001));
    Mockito.when(s3Object.getObjectMetadata()).thenReturn(om);
    Mockito.when(s3.getObject(bucketName, s3Id)).thenReturn(s3Object);
    Sqs.queueName(queueName) //
            .sqsFactory(() -> sqs) //
            .bucketName("bucket") //
            .s3Factory(() -> s3) //
            .messages() //
            .doOnNext(SqsMessage::deleteMessage) //
            .map(m -> m.message()) //
            .doOnError(Throwable::printStackTrace) //
            .take(1) //
            .test() //
            .awaitDone(10, TimeUnit.SECONDS) //
            .assertComplete() //
            .assertValues("body1");
    final InOrder inorder = Mockito.inOrder(sqs, s3, s3Object);
    inorder.verify(sqs, Mockito.atLeastOnce()).getQueueUrl(queueName);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id);
    inorder.verify(s3, Mockito.times(1)).getObject(bucketName, s3Id);
    inorder.verify(s3Object, Mockito.times(1)).getObjectContent();
    inorder.verify(s3Object, Mockito.times(1)).getObjectMetadata();
    inorder.verify(s3, Mockito.times(1)).deleteObject(bucketName, s3Id);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle);
    inorder.verify(sqs, Mockito.times(1)).shutdown();
    inorder.verify(s3, Mockito.times(1)).shutdown();
    inorder.verifyNoMoreInteractions();
}

@Test(timeout = 50000000)
public void testFirstCallToReceiveMessagesReturnsOneWithNoS3ObjectAndOneWithS3Object()
        throws UnsupportedEncodingException {
    final AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
    final AmazonS3Client s3 = Mockito.mock(AmazonS3Client.class);
    final String queueName = "queue";
    final String s3Id = "123";
    final String s3Id2 = "124";
    Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
    final String receiptHandle = "abc";
    final String receiptHandle2 = "abc2";
    Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
            .thenReturn(new ReceiveMessageResult()
                    .withMessages(new Message().withReceiptHandle(receiptHandle).withBody(s3Id)))
            .thenReturn(new ReceiveMessageResult()
                    .withMessages(new Message().withReceiptHandle(receiptHandle2).withBody(s3Id2)));
    final String bucketName = "bucket";
    Mockito.when(s3.doesObjectExist(bucketName, s3Id)).thenReturn(false);
    Mockito.when(s3.doesObjectExist(bucketName, s3Id2)).thenReturn(true);
    final S3Object s3Object = mock(S3Object.class);
    Mockito.when(s3Object.getObjectContent())
            .thenReturn(new S3ObjectInputStream(new StringInputStream("body2"), null));
    final ObjectMetadata om = new ObjectMetadata();
    om.setLastModified(new Date(1001));
    Mockito.when(s3Object.getObjectMetadata()).thenReturn(om);
    Mockito.when(s3.getObject(bucketName, s3Id2)).thenReturn(s3Object);
    Sqs.queueName(queueName) //
            .sqsFactory(() -> sqs) //
            .bucketName("bucket") //
            .s3Factory(() -> s3) //
            .messages() //
            .doOnNext(SqsMessage::deleteMessage) //
            .map(m -> m.message()) //
            .doOnError(Throwable::printStackTrace) //
            .take(1) //
            .test() //
            .awaitDone(10, TimeUnit.SECONDS) //
            .assertComplete() //
            .assertValues("body2");
    final InOrder inorder = Mockito.inOrder(sqs, s3, s3Object);
    inorder.verify(sqs, Mockito.atLeastOnce()).getQueueUrl(queueName);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id2);
    inorder.verify(s3, Mockito.times(1)).getObject(bucketName, s3Id2);
    inorder.verify(s3Object, Mockito.times(1)).getObjectContent();
    inorder.verify(s3Object, Mockito.times(1)).getObjectMetadata();
    inorder.verify(s3, Mockito.times(1)).deleteObject(bucketName, s3Id2);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle2);
    inorder.verify(sqs, Mockito.times(1)).shutdown();
    inorder.verify(s3, Mockito.times(1)).shutdown();
    inorder.verifyNoMoreInteractions();
}

public static void main(String[] args) throws Exception {
    // create the AWS S3 Client
    AmazonS3 s3 = AWSS3Factory.getS3Client();

    // retrieve object key/value from user
    System.out.println( "Enter the object key:" );
    String key = new BufferedReader( new InputStreamReader( System.in ) ).readLine();
    System.out.println( "Enter the object content:" );
    String content = new BufferedReader( new InputStreamReader( System.in ) ).readLine();

    String theBucket = AWSS3Factory.S3_BUCKET;

    PutObjectResult por = s3.putObject(theBucket, key, new StringInputStream(content), null);

    _01_CreateObject example = new _01_CreateObject();

    //PutObjectResult por = example.putObjectViaRequest(s3, theBucket, key, content);

    //PutObjectResult por = example.putManyObjectsWithPrefix(s3, key, content, "demo-obj/", 9);

    System.out.println(String.format("created object [%s/%s] with content: [%s]",
            theBucket, key, content));



}

public PutObjectResult putObjectViaRequest(AmazonS3 s3, String bucket, String key, String content) throws Exception {
    PutObjectRequest poReq = new PutObjectRequest(bucket, key, new StringInputStream(content), null);
    PutObjectResult por = s3.putObject(poReq);
    return por;
}

@Test(timeout = 5000)
public void testFirstCallToReceiveMessagesReturnsOneViaS3() throws UnsupportedEncodingException {
    AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
    AmazonS3Client s3 = Mockito.mock(AmazonS3Client.class);
    String queueName = "queue";
    String s3Id = "123";
    Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
    String receiptHandle = "abc";
    Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any())).thenReturn(
            new ReceiveMessageResult().withMessages(new Message().withReceiptHandle(receiptHandle).withBody(s3Id)));
    String bucketName = "bucket";
    Mockito.when(s3.doesObjectExist(bucketName, s3Id)).thenAnswer(x -> true);
    S3Object s3Object = mock(S3Object.class);
    Mockito.when(s3Object.getObjectContent())
            .thenReturn(new S3ObjectInputStream(new StringInputStream("body1"), null));
    ObjectMetadata om = new ObjectMetadata();
    om.setLastModified(new Date(1001));
    Mockito.when(s3Object.getObjectMetadata()).thenReturn(om);
    Mockito.when(s3.getObject(bucketName, s3Id)).thenReturn(s3Object);
    Sqs.queueName(queueName) //
            .sqsFactory(() -> sqs) //
            .bucketName("bucket") //
            .s3Factory(() -> s3) //
            .messages() //
            .doOnNext(SqsMessage::deleteMessage) //
            .map(m -> m.message()) //
            .doOnError(Throwable::printStackTrace) //
            .take(1) //
            .to(test()) //
            .awaitTerminalEvent() //
            .assertCompleted() //
            .assertValues("body1");
    InOrder inorder = Mockito.inOrder(sqs, s3, s3Object);
    inorder.verify(sqs, Mockito.atLeastOnce()).getQueueUrl(queueName);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id);
    inorder.verify(s3, Mockito.times(1)).getObject(bucketName, s3Id);
    inorder.verify(s3Object, Mockito.times(1)).getObjectContent();
    inorder.verify(s3Object, Mockito.times(1)).getObjectMetadata();
    inorder.verify(s3, Mockito.times(1)).deleteObject(bucketName, s3Id);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle);
    inorder.verify(sqs, Mockito.times(1)).shutdown();
    inorder.verify(s3, Mockito.times(1)).shutdown();
    inorder.verifyNoMoreInteractions();
}

@Test(timeout = 50000000)
public void testFirstCallToReceiveMessagesReturnsOneWithNoS3ObjectAndOneWithS3Object()
        throws UnsupportedEncodingException {
    AmazonSQSClient sqs = Mockito.mock(AmazonSQSClient.class);
    AmazonS3Client s3 = Mockito.mock(AmazonS3Client.class);
    String queueName = "queue";
    String s3Id = "123";
    String s3Id2 = "124";
    Mockito.when(sqs.getQueueUrl(queueName)).thenAnswer(x -> new GetQueueUrlResult().withQueueUrl(queueName));
    String receiptHandle = "abc";
    String receiptHandle2 = "abc2";
    Mockito.when(sqs.receiveMessage(Mockito.<ReceiveMessageRequest>any()))
            .thenReturn(new ReceiveMessageResult()
                    .withMessages(new Message().withReceiptHandle(receiptHandle).withBody(s3Id)))
            .thenReturn(new ReceiveMessageResult()
                    .withMessages(new Message().withReceiptHandle(receiptHandle2).withBody(s3Id2)));
    String bucketName = "bucket";
    Mockito.when(s3.doesObjectExist(bucketName, s3Id)).thenReturn(false);
    Mockito.when(s3.doesObjectExist(bucketName, s3Id2)).thenReturn(true);
    S3Object s3Object = mock(S3Object.class);
    Mockito.when(s3Object.getObjectContent())
            .thenReturn(new S3ObjectInputStream(new StringInputStream("body2"), null));
    ObjectMetadata om = new ObjectMetadata();
    om.setLastModified(new Date(1001));
    Mockito.when(s3Object.getObjectMetadata()).thenReturn(om);
    Mockito.when(s3.getObject(bucketName, s3Id2)).thenReturn(s3Object);
    Sqs.queueName(queueName) //
            .sqsFactory(() -> sqs) //
            .bucketName("bucket") //
            .s3Factory(() -> s3) //
            .messages() //
            .doOnNext(SqsMessage::deleteMessage) //
            .map(m -> m.message()) //
            .doOnError(Throwable::printStackTrace) //
            .take(1) //
            .to(test()) //
            .awaitTerminalEvent() //
            .assertCompleted() //
            .assertValues("body2");
    InOrder inorder = Mockito.inOrder(sqs, s3, s3Object);
    inorder.verify(sqs, Mockito.atLeastOnce()).getQueueUrl(queueName);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle);
    inorder.verify(sqs, Mockito.times(1)).receiveMessage(Mockito.<ReceiveMessageRequest>any());
    inorder.verify(s3, Mockito.times(1)).doesObjectExist(bucketName, s3Id2);
    inorder.verify(s3, Mockito.times(1)).getObject(bucketName, s3Id2);
    inorder.verify(s3Object, Mockito.times(1)).getObjectContent();
    inorder.verify(s3Object, Mockito.times(1)).getObjectMetadata();
    inorder.verify(s3, Mockito.times(1)).deleteObject(bucketName, s3Id2);
    inorder.verify(sqs, Mockito.times(1)).deleteMessage(queueName, receiptHandle2);
    inorder.verify(sqs, Mockito.times(1)).shutdown();
    inorder.verify(s3, Mockito.times(1)).shutdown();
    inorder.verifyNoMoreInteractions();
}

@Test
public void testRun()
        throws Exception
{
    String key = UUID.randomUUID().toString();


    Path outfile = folder.newFolder().toPath().resolve("out");

    createProject(projectDir);
    addWorkflow(projectDir, "acceptance/s3/s3_wait.dig");

    Id projectId = TestUtils.pushProject(server.endpoint(), projectDir);

    // Configure AWS credentials
    client.setProjectSecret(projectId, "aws.s3.access_key_id", TEST_S3_ACCESS_KEY_ID);
    client.setProjectSecret(projectId, "aws.s3.secret_access_key", TEST_S3_SECRET_ACCESS_KEY);
    client.setProjectSecret(projectId, "aws.s3.endpoint", TEST_S3_ENDPOINT);

    // Start workflow
    String projectName = projectDir.getFileName().toString();
    Id attemptId = startWorkflow(server.endpoint(), projectName, "s3_wait", ImmutableMap.of(
            "path", bucket + "/" + key,
            "outfile", outfile.toString()
    ));

    // Wait for s3 polling to show up in logs
    expect(Duration.ofSeconds(30), () -> {
        String attemptLogs = TestUtils.getAttemptLogs(client, attemptId);
        return attemptLogs.contains("s3_wait>: " + bucket + "/" + key);
    });

    // Verify that the dependent task has not been executed
    assertThat(Files.exists(outfile), is(false));

    // Verify that the attempt is not yet done
    RestSessionAttempt attempt = client.getSessionAttempt(attemptId);
    assertThat(attempt.getDone(), is(false));

    // Create the file that the workflow is waiting for
    String content = "hello world";
    s3.putObject(bucket, key, new StringInputStream(content), new ObjectMetadata());

    // Expect the attempt to finish and the dependent task to be executed
    expect(Duration.ofMinutes(2), attemptSuccess(server.endpoint(), attemptId));
    assertThat(Files.exists(outfile), is(true));

    JsonNode objectMetadata = MAPPER.readTree(Files.readAllBytes(outfile));
    int contentLength = objectMetadata.get("metadata").get("Content-Length").asInt();
    assertThat(contentLength, is(content.length()));
}

@Before
public void setUp()
        throws Exception
{
    assumeThat(GCP_CREDENTIAL, not(isEmptyOrNullString()));

    proxyServer = TestUtils.startRequestFailingProxy(2, new ConcurrentHashMap<>(), HttpResponseStatus.INTERNAL_SERVER_ERROR,
            (req, reqCount) -> {
                // io.digdag.standards.operator.gcp.BqJobRunner sends "CONNECT www.googleapis.com" frequently. It can easily cause infinite retry.
                // So the following custom logic should be used for that kind of requests.
                if (req.getMethod().equals(HttpMethod.CONNECT)) {
                    return Optional.of(reqCount % 5 == 0);
                }
                return Optional.absent();
            });

    server = TemporaryDigdagServer.builder()
            .environment(ImmutableMap.of(
                    "https_proxy", "http://" + proxyServer.getListenAddress().getHostString() + ":" + proxyServer.getListenAddress().getPort())
            )
            .withRandomSecretEncryptionKey()
            .build();
    server.start();

    projectDir = folder.getRoot().toPath();
    createProject(projectDir);
    projectName = projectDir.getFileName().toString();
    projectId = pushProject(server.endpoint(), projectDir, projectName);

    outfile = folder.newFolder().toPath().resolve("outfile");

    digdagClient = DigdagClient.builder()
            .host(server.host())
            .port(server.port())
            .build();

    digdagClient.setProjectSecret(projectId, "gcp.credential", GCP_CREDENTIAL);

    gcpCredential = GoogleCredential.fromStream(new StringInputStream(GCP_CREDENTIAL));

    assertThat(gcpProjectId, not(isEmptyOrNullString()));

    jsonFactory = new JacksonFactory();
    transport = GoogleNetHttpTransport.newTrustedTransport();
    gcs = gcsClient(gcpCredential);
    bq = bqClient(gcpCredential);
}

@Before
public void setUp()
        throws Exception
{
    assumeThat(GCP_CREDENTIAL, not(isEmptyOrNullString()));
    assumeThat(GCS_TEST_BUCKET, not(isEmptyOrNullString()));

    proxyServer = TestUtils.startRequestFailingProxy(1);

    server = TemporaryDigdagServer.builder()
            .environment(ImmutableMap.of(
                    "https_proxy", "http://" + proxyServer.getListenAddress().getHostString() + ":" + proxyServer.getListenAddress().getPort())
            )
            .withRandomSecretEncryptionKey()
            .build();
    server.start();

    projectDir = folder.getRoot().toPath();
    createProject(projectDir);
    projectName = projectDir.getFileName().toString();
    projectId = pushProject(server.endpoint(), projectDir, projectName);

    outfile = folder.newFolder().toPath().resolve("outfile");

    digdagClient = DigdagClient.builder()
            .host(server.host())
            .port(server.port())
            .build();

    digdagClient.setProjectSecret(projectId, "gcp.credential", GCP_CREDENTIAL);

    gcpCredential = GoogleCredential.fromStream(new StringInputStream(GCP_CREDENTIAL));

    gcpProjectId = DigdagClient.objectMapper().readTree(GCP_CREDENTIAL).get("project_id").asText();
    assertThat(gcpProjectId, not(isEmptyOrNullString()));

    jsonFactory = new JacksonFactory();
    transport = GoogleNetHttpTransport.newTrustedTransport();
    gcs = gcsClient(gcpCredential);

    client = DigdagClient.builder()
            .host(server.host())
            .port(server.port())
            .build();
}

@Test
public void testZeroPaddedDigestMD5() throws Exception {
    inputStream2 = new StringInputStream(INPUT_STRING_NO2);
    String result = DigestHelper.digest(MD5, inputStream2).toString();
    Assert.assertEquals(ZERO_PADDED_MD5_CHECKSUM, result);
}

@Test
public void testZeroPaddedDigestSHA256() throws Exception {
    inputStream2 = new StringInputStream(INPUT_STRING_NO3);
    String result = DigestHelper.digest(SHA_256, inputStream2).toString();
    Assert.assertEquals(ZERO_PADDED_SHA256_CHECKSUM, result);
}

@BeforeClass
public static void init() throws UnsupportedEncodingException {
    inputStream = new StringInputStream(INPUT_STRING);
}