= 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.
:since: 3.2
:supportlevel: Stable
:tabs-sync-option:
: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 doesn't already exist.

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].

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

== Usage

=== Batch Consumer

This component implements the Batch Consumer.

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

=== 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

== Examples

For example, 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");
--------------------------------------------------------------------------------

=== 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 S3 to use the KMS key 3f0637ad-296a-3dfe-a796-e60654fb128c, to encrypt the file test.txt.
When you ask to download this file, the decryption will be done directly before the download.

=== Static credentials, Default Credential Provider and Profile Credentials Provider

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

The order of evaluation for Default Credentials Provider is the following:

 - 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. 
 
You have also the possibility of using Profile Credentials Provider, by specifying the useProfileCredentialsProvider option to true and profileCredentialsName to the profile name.

Only one of static, default and profile credentials could be used at the same time.

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 copies 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 lists 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 deletes 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 gets 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 gets 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 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.

=== AWS S3 Producer minimum permissions

For making the producer work, you'll need at least PutObject and ListBuckets permissions. The following policy will be enough:

[source,json]
--------------------------------------------------------------------------------
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::*/*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::*"
        }
    ]
}
--------------------------------------------------------------------------------

A variation to the minimum permissions is related to the usage of Bucket autocreation. In that case the permissions will need to be increased with CreateBucket permission

[source,json]
--------------------------------------------------------------------------------
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:PutObject",
            "Resource": "arn:aws:s3:::*/*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:CreateBucket",
            "Resource": "arn:aws:s3:::*"
        }
    ]
}
--------------------------------------------------------------------------------

=== AWS S3 Consumer minimum permissions

For making the producer work, you'll need at least GetObject, ListBuckets and DeleteObject permissions. The following policy will be enough:

[source,json]
--------------------------------------------------------------------------------
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:ListBucket",
            "Resource": "arn:aws:s3:::*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::*/*"
        },
        {
            "Effect": "Allow",
            "Action": "s3:DeleteObject",
            "Resource": "arn:aws:s3:::*/*"
        }
    ]
}
--------------------------------------------------------------------------------

By Default the consumer will use the deleteAfterRead option, this means the object will be deleted once consumed, this is why the DeleteObject permission is required.

=== 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 strategies: 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 be 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 stop your producer route, the producer will take care of flushing the remaining buffered message and complete the upload.

In Streaming upload, you'll be able to 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 four files: camel.txt, camel-1.txt, camel-2.txt and camel-3.txt, the first three will have 20 messages, while the last one is only 10.
 - Restart the route
 - Send 25 messages
 - Stop the route
 - You'll now have two other files in your bucket: camel-5.txt and camel-6.txt, the first with 20 messages and the 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) in 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 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 Auto-creation

With the option `autoCreateBucket` users are able to avoid the auto-creation 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, remember to remove the bucketName header from the incoming exchange of the consumer, otherwise the file will always be 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 the 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 building an AWS Request can be complex because of multiple options. We introduce the possibility to use a POJO as the 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[]
