= AWS S3 Storage Service Component
:doctitle: AWS S3 Storage Service
:shortname: aws2-s3
:artifactid: camel-aws2-s3
:description: Store and retrieve objects from AWS S3 Storage Service using AWS SDK version 2.x.
:since: 3.2
:supportlevel: Stable
:component-header: Both producer and consumer are supported
//Manually maintained attributes
:group: AWS
:camel-spring-boot-name: aws2-s3

*Since Camel {since}*

*{component-header}*

The AWS2 S3 component supports storing and retrieving objects from/to
https://aws.amazon.com/s3[Amazon's S3] service.

Prerequisites

You must have a valid Amazon Web Services developer account, and be
signed up to use Amazon S3. More information is available at
https://aws.amazon.com/s3[Amazon S3].

== URI Format

------------------------------
aws2-s3://bucketNameOrArn[?options]
------------------------------

The bucket will be created if it don't already exists. +
 You can append query options to the URI in the following format,
?options=value&option2=value&...


// component-configure options: START

// component-configure options: END

// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
// component options: END

// endpoint options: START

// endpoint options: END


Required S3 component options

You have to provide the amazonS3Client in the
Registry or your accessKey and secretKey to access
the https://aws.amazon.com/s3[Amazon's S3].

== Batch Consumer

This component implements the Batch Consumer.

This allows you for instance to know how many messages exists in this
batch and for instance let the Aggregator
aggregate this number of messages.

== Usage

For example in order to read file `hello.txt` from bucket `helloBucket`, use the following snippet:

[source,java]
--------------------------------------------------------------------------------
from("aws2-s3://helloBucket?accessKey=yourAccessKey&secretKey=yourSecretKey&prefix=hello.txt")
  .to("file:/var/downloaded");
--------------------------------------------------------------------------------

// component headers: START
include::partial$component-endpoint-headers.adoc[]
// component headers: END

=== S3 Producer operations

Camel-AWS2-S3 component provides the following operation on the producer side:

- copyObject
- deleteObject
- listBuckets
- deleteBucket
- listObjects
- getObject (this will return an S3Object instance)
- getObjectRange (this will return an S3Object instance)
- createDownloadLink

If you don't specify an operation explicitly the producer will do:
- a single file upload
- a multipart upload if multiPartUpload option is enabled

=== Advanced AmazonS3 configuration

If your Camel Application is running behind a firewall or if you need to
have more control over the `S3Client` instance configuration, you can
create your own instance and refer to it in your Camel aws2-s3 component configuration:

[source,java]
--------------------------------------------------------------------------------
from("aws2-s3://MyBucket?amazonS3Client=#client&delay=5000&maxMessagesPerPoll=5")
.to("mock:result");
--------------------------------------------------------------------------------

=== Use KMS with the S3 component

To use AWS KMS to encrypt/decrypt data by using AWS infrastructure you can use the options introduced in 2.21.x like in the following example

[source,java]
--------------------------------------------------------------------------------
from("file:tmp/test?fileName=test.txt")
     .setHeader(AWS2S3Constants.KEY, constant("testFile"))
     .to("aws2-s3://mybucket?amazonS3Client=#client&useAwsKMS=true&awsKMSKeyId=3f0637ad-296a-3dfe-a796-e60654fb128c");
--------------------------------------------------------------------------------

In this way you'll ask to S3, to use the KMS key 3f0637ad-296a-3dfe-a796-e60654fb128c, to encrypt the file test.txt. When you'll ask to download this file, the decryption will be done directly before the download.

=== Static credentials vs Default Credential Provider

You have the possibility of avoiding the usage of explicit static credentials, by specifying the useDefaultCredentialsProvider option and set it to true.

 - Java system properties - aws.accessKeyId and aws.secretKey
 - Environment variables - AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
 - Web Identity Token from AWS STS.
 - The shared credentials and config files.
 - Amazon ECS container credentials - loaded from the Amazon ECS if the environment variable AWS_CONTAINER_CREDENTIALS_RELATIVE_URI is set.
 - Amazon EC2 Instance profile credentials. 

For more information about this you can look at https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html[AWS credentials documentation]

=== S3 Producer Operation examples

- Single Upload: This operation will upload a file to S3 based on the body content

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camel.txt");
          exchange.getIn().setBody("Camel rocks!");
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will upload the file camel.txt with the content "Camel rocks!" in the mycamelbucket bucket

- Multipart Upload: This operation will perform a multipart upload of a file to S3 based on the body content

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "empty.txt");
          exchange.getIn().setBody(new File("src/empty.txt"));
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&multiPartUpload=true&autoCreateBucket=true&partSize=1048576")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will perform a multipart upload of the file empty.txt with based on the content the file src/empty.txt in the mycamelbucket bucket

- CopyObject: this operation copy an object from one bucket to a different one

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.BUCKET_DESTINATION_NAME, "camelDestinationBucket");
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camelKey");
          exchange.getIn().setHeader(AWS2S3Constants.DESTINATION_KEY, "camelDestinationKey");
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=copyObject")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will copy the object with the name expressed in the header camelDestinationKey to the camelDestinationBucket bucket, from the bucket mycamelbucket.

- DeleteObject: this operation deletes an object from a bucket

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camelKey");
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=deleteObject")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will delete the object camelKey from the bucket mycamelbucket.

- ListBuckets: this operation list the buckets for this account in this region

[source,java]
--------------------------------------------------------------------------------
  from("direct:start")
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=listBuckets")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will list the buckets for this account

- DeleteBucket: this operation delete the bucket specified as URI parameter or header

[source,java]
--------------------------------------------------------------------------------
  from("direct:start")
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=deleteBucket")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will delete the bucket mycamelbucket

- ListObjects: this operation list object in a specific bucket

[source,java]
--------------------------------------------------------------------------------
  from("direct:start")
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=listObjects")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will list the objects in the mycamelbucket bucket

- GetObject: this operation get a single object in a specific bucket

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camelKey");
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=getObject")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will return an S3Object instance related to the camelKey object in mycamelbucket bucket.

- GetObjectRange: this operation get a single object range in a specific bucket

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camelKey");
          exchange.getIn().setHeader(AWS2S3Constants.RANGE_START, "0");
          exchange.getIn().setHeader(AWS2S3Constants.RANGE_END, "9");
      }
  })
  .to("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&operation=getObjectRange")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will return an S3Object instance related to the camelKey object in mycamelbucket bucket, containing a the bytes from 0 to 9.

- CreateDownloadLink: this operation will return a download link through S3 Presigner

[source,java]
--------------------------------------------------------------------------------
  from("direct:start").process(new Processor() {

      @Override
      public void process(Exchange exchange) throws Exception {
          exchange.getIn().setHeader(AWS2S3Constants.KEY, "camelKey");
      }
  })
  .to("aws2-s3://mycamelbucket?accessKey=xxx&secretKey=yyy&region=region&operation=createDownloadLink")
  .to("mock:result");
--------------------------------------------------------------------------------

This operation will return a download link url for the file camel-key in the bucket mycamelbucket and region region.
Parameters (`accessKey`, `secretKey` and `region`) are mandatory for this operation, if S3 client is autowired from the registry.

NOTE: If checksum validations are enabled, the url will no longer be browser compatible because it adds a signed header that must be included in the HTTP request.

== Streaming Upload mode

With the stream mode enabled users will be able to upload data to S3 without knowing ahead of time the dimension of the data, by leveraging multipart upload.
The upload will be completed when: the batchSize has been completed or the batchMessageNumber has been reached.
There are two possible naming strategy: progressive and random. With the progressive strategy each file will have the name composed by keyName option and a progressive counter, and eventually the file extension (if any), while with the random strategy a UUID will be added after keyName and eventually the file extension will appended.

As an example:

[source,java]
--------------------------------------------------------------------------------
from(kafka("topic1").brokers("localhost:9092"))
        .log("Kafka Message is: ${body}")
        .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));

from(kafka("topic2").brokers("localhost:9092"))
         .log("Kafka Message is: ${body}")
         .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.random).keyName("{{kafkaTopic2}}/{{kafkaTopic2}}.txt"));
--------------------------------------------------------------------------------

The default size for a batch is 1 Mb, but you can adjust it according to your requirements.

When you'll stop your producer route, the producer will take care of flushing the remaining buffered messaged and complete the upload.

In Streaming upload you'll be able restart the producer from the point where it left. It's important to note that this feature is critical only when using the progressive naming strategy.

By setting the restartingPolicy to lastPart, you will restart uploading files and contents from the last part number the producer left.

As example: 
 - Start the route with progressive naming strategy and keyname equals to camel.txt, with batchMessageNumber equals to 20, and restartingPolicy equals to lastPart
 - Send 70 messages.
 - Stop the route
 - On your S3 bucket you should now see 4 files: camel.txt, camel-1.txt, camel-2.txt and camel-3.txt, the first three will have 20 messages, while the last one only 10.
 - Restart the route
 - Send 25 messages
 - Stop the route
 - You'll now have 2 other files in your bucket: camel-5.txt and camel-6.txt, the first with 20 messages and second with 5 messages.
 - Go ahead

This won't be needed when using the random naming strategy.

On the opposite you can specify the override restartingPolicy. In that case you'll be able to override whatever you written before (for that particular keyName) on your bucket.

[NOTE]
====
In Streaming upload mode the only keyName option that will be taken into account is the endpoint option. Using the header will throw an NPE and this is done by design.
Setting the header means potentially change the file name on each exchange and this is against the aim of the streaming upload producer. The keyName needs to be fixed and static. 
The selected naming strategy will do the rest of the of the work.
====

Another possibility is specifying a streamingUploadTimeout with batchMessageNumber and batchSize options. With this option the user will be able to complete the upload of a file after a certain time passed.
In this way the upload completion will be passed on three tiers: the timeout, the number of messages and the batch size.

As an example:

[source,java]
--------------------------------------------------------------------------------
from(kafka("topic1").brokers("localhost:9092"))
        .log("Kafka Message is: ${body}")
        .to(aws2S3("camel-bucket").streamingUploadMode(true).batchMessageNumber(25).streamingUploadTimeout(10000).namingStrategy(AWS2S3EndpointBuilderFactory.AWSS3NamingStrategyEnum.progressive).keyName("{{kafkaTopic1}}/{{kafkaTopic1}}.txt"));
--------------------------------------------------------------------------------

In this case the upload will be completed after 10 seconds.

== Bucket Autocreation

With the option `autoCreateBucket` users are able to avoid the autocreation of an S3 Bucket in case it doesn't exist. The default for this option is `false`.
If set to false any operation on a not-existent bucket in AWS won't be successful and an error will be returned.

== Moving stuff between a bucket and another bucket

Some users like to consume stuff from a bucket and move the content in a different one without using the copyObject feature of this component.
If this is case for you, don't forget to remove the bucketName header from the incoming exchange of the consumer, otherwise the file will be always overwritten on the same
original bucket.

== MoveAfterRead consumer option

In addition to deleteAfterRead it has been added another option, moveAfterRead. With this option enabled the consumed object will be moved to a target destinationBucket instead of being only deleted.
This will require specifying the destinationBucket option. As example:

[source,java]
--------------------------------------------------------------------------------
  from("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&moveAfterRead=true&destinationBucket=myothercamelbucket")
  .to("mock:result");
--------------------------------------------------------------------------------

In this case the objects consumed will be moved to myothercamelbucket bucket and deleted from the original one (because of deleteAfterRead set to true as default).

You have also the possibility of using a key prefix/suffix while moving the file to a different bucket. The options are destinationBucketPrefix and destinationBucketSuffix.

Taking the above example, you could do something like:

[source,java]
--------------------------------------------------------------------------------
  from("aws2-s3://mycamelbucket?amazonS3Client=#amazonS3Client&moveAfterRead=true&destinationBucket=myothercamelbucket&destinationBucketPrefix=RAW(pre-)&destinationBucketSuffix=RAW(-suff)")
  .to("mock:result");
--------------------------------------------------------------------------------

In this case the objects consumed will be moved to myothercamelbucket bucket and deleted from the original one (because of deleteAfterRead set to true as default).

So if the file name is test, in the myothercamelbucket you should see a file called pre-test-suff.

== Using customer key as encryption

We introduced also the customer key support (an alternative of using KMS). The following code shows an example.

[source,java]
--------------------------------------------------------------------------------
String key = UUID.randomUUID().toString();
byte[] secretKey = generateSecretKey();
String b64Key = Base64.getEncoder().encodeToString(secretKey);
String b64KeyMd5 = Md5Utils.md5AsBase64(secretKey);

String awsEndpoint = "aws2-s3://mycamel?autoCreateBucket=false&useCustomerKey=true&customerKeyId=RAW(" + b64Key + ")&customerKeyMD5=RAW(" + b64KeyMd5 + ")&customerAlgorithm=" + AES256.name();

from("direct:putObject")
    .setHeader(AWS2S3Constants.KEY, constant("test.txt"))
    .setBody(constant("Test"))
    .to(awsEndpoint);
--------------------------------------------------------------------------------

== Using a POJO as body

Sometimes build an AWS Request can be complex, because of multiple options. We introduce the possibility to use a POJO as body.
In AWS S3 there are multiple operations you can submit, as an example for List brokers request, you can do something like:

[source,java]
------------------------------------------------------------------------------------------------------
from("direct:aws2-s3")
     .setBody(ListObjectsRequest.builder().bucket(bucketName).build())
     .to("aws2-s3://test?amazonS3Client=#amazonS3Client&operation=listObjects&pojoRequest=true")
------------------------------------------------------------------------------------------------------

In this way you'll pass the request directly without the need of passing headers and options specifically related to this operation.

== Create S3 client and add component to registry
Sometimes you would want to perform some advanced configuration using AWS2S3Configuration which also allows to set the S3 client.
You can create and set the S3 client in the component configuration as shown in the following example

[source,java]
--------------------------------------------------------------------------------
String awsBucketAccessKey = "your_access_key";
String awsBucketSecretKey = "your_secret_key";

S3Client s3Client = S3Client.builder().credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create(awsBucketAccessKey, awsBucketSecretKey)))
                .region(Region.US_EAST_1).build();

AWS2S3Configuration configuration = new AWS2S3Configuration();
configuration.setAmazonS3Client(s3Client);
configuration.setAutoDiscoverClient(true);
configuration.setBucketName("s3bucket2020");
configuration.setRegion("us-east-1");
--------------------------------------------------------------------------------

Now you can configure the S3 component (using the configuration object created above) and add it to the registry in the
configure method before initialization of routes.

[source,java]
--------------------------------------------------------------------------------
AWS2S3Component s3Component = new AWS2S3Component(getContext());
s3Component.setConfiguration(configuration);
s3Component.setLazyStartProducer(true);
camelContext.addComponent("aws2-s3", s3Component);
--------------------------------------------------------------------------------

Now your component will be used for all the operations implemented in camel routes.

== Dependencies

Maven users will need to add the following dependency to their pom.xml.

*pom.xml*

[source,xml]
---------------------------------------
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-aws2-s3</artifactId>
    <version>${camel-version}</version>
</dependency>
---------------------------------------

where `$\{camel-version}` must be replaced by the actual version of Camel.



include::spring-boot:partial$starter.adoc[]
