/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.camel.component.aws2.s3.stream;

import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.component.aws2.s3.AWS2S3Configuration;
import org.apache.camel.component.aws2.s3.AWS2S3Constants;
import org.apache.camel.component.aws2.s3.AWS2S3Endpoint;
import org.apache.camel.component.aws2.s3.utils.AWS2S3Utils;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.URISupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.awscore.AwsResponse;
import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.BucketCannedACL;
import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
import software.amazon.awssdk.services.s3.model.CompletedPart;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;
import software.amazon.awssdk.services.s3.model.S3Object;
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
import software.amazon.awssdk.services.s3.model.UploadPartResponse;
import software.amazon.awssdk.services.s3.paginators.ListObjectsV2Iterable;

/**
 * A Producer which sends messages to the Amazon Web Service Simple Storage Service
 * <a href="http://aws.amazon.com/s3/">AWS S3</a>
 */
public class AWS2S3StreamUploadProducer extends DefaultProducer {

    private static final Logger LOG = LoggerFactory.getLogger(AWS2S3StreamUploadProducer.class);
    private static final String TIMEOUT_CHECKER_EXECUTOR_NAME = "S3_Streaming_Upload_Timeout_Checker";
    private AtomicInteger part = new AtomicInteger();
    private UploadState uploadAggregate = null;
    private final Map<Long, UploadState> timestampBasedUploads = new ConcurrentHashMap<>();
    private final Lock lock = new ReentrantLock();
    private transient String s3ProducerToString;
    private ScheduledExecutorService timeoutCheckerExecutorService;
    private ChecksumAlgorithm algorithm = ChecksumAlgorithm.CRC32;

    public AWS2S3StreamUploadProducer(final Endpoint endpoint) {
        super(endpoint);
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        if (getConfiguration().getStreamingUploadTimeout() > 0) {
            timeoutCheckerExecutorService
                    = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
                            TIMEOUT_CHECKER_EXECUTOR_NAME);
            timeoutCheckerExecutorService.scheduleAtFixedRate(new StreamingUploadTimeoutTask(),
                    getConfiguration().getStreamingUploadTimeout(), getConfiguration().getStreamingUploadTimeout(),
                    TimeUnit.MILLISECONDS);
        }
        if (getConfiguration().getRestartingPolicy().equals(AWSS3RestartingPolicyEnum.lastPart)) {
            setStartingPart();
        }
    }

    @Override
    protected void doStop() throws Exception {
        lock.lock();
        try {
            if (ObjectHelper.isNotEmpty(uploadAggregate)) {
                uploadPart(uploadAggregate);
                completeUpload(uploadAggregate);
            }

            // Complete any pending timestamp-based uploads
            for (UploadState state : timestampBasedUploads.values()) {
                if (ObjectHelper.isNotEmpty(state) && state.buffer.size() > 0) {
                    uploadPart(state);
                    completeUpload(state);
                }
            }
            timestampBasedUploads.clear();
        } finally {
            lock.unlock();
        }
        if (timeoutCheckerExecutorService != null) {
            getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(timeoutCheckerExecutorService);
            timeoutCheckerExecutorService = null;
        }
        super.doStop();

    }

    /**
     * Background task that triggers completion based on interval.
     */
    private final class StreamingUploadTimeoutTask implements Runnable {

        @Override
        public void run() {
            lock.lock();
            try {
                if (ObjectHelper.isNotEmpty(uploadAggregate)) {
                    uploadPart(uploadAggregate);
                    completeUpload(uploadAggregate);
                    uploadAggregate = null;
                }

                // Handle timestamp-based uploads timeout
                if (getConfiguration().isTimestampGroupingEnabled()) {
                    List<Long> keysToRemove = new ArrayList<>();
                    for (Map.Entry<Long, UploadState> entry : timestampBasedUploads.entrySet()) {
                        UploadState state = entry.getValue();
                        if (ObjectHelper.isNotEmpty(state) && state.buffer.size() > 0) {
                            uploadPart(state);
                            completeUpload(state);
                            keysToRemove.add(entry.getKey());
                        }
                    }
                    keysToRemove.forEach(timestampBasedUploads::remove);
                }
            } finally {
                lock.unlock();
            }
        }
    }

    @Override
    public void process(final Exchange exchange) throws Exception {
        if (getConfiguration().isTimestampGroupingEnabled()) {
            processWithTimestampGrouping(exchange);
        } else {
            processWithoutTimestampGrouping(exchange);
        }
    }

    private void processWithoutTimestampGrouping(final Exchange exchange) throws Exception {
        InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);

        UploadState state = null;
        int totalSize = 0;
        byte[] b;
        int maxRead = (getConfiguration().isMultiPartUpload()
                ? Math.toIntExact(getConfiguration().getPartSize()) : getConfiguration().getBufferSize());
        if (uploadAggregate != null) {
            uploadAggregate.index++;
            maxRead -= uploadAggregate.buffer.size();
        }

        while ((b = AWS2S3Utils.toByteArray(is, maxRead)) != null && b.length > 0) {
            totalSize += b.length;
            if (getConfiguration().isMultiPartUpload())
                maxRead -= b.length;
            lock.lock();
            try {
                // aggregate with previously received exchanges
                if (ObjectHelper.isNotEmpty(uploadAggregate)) {
                    uploadAggregate.buffer.write(b);
                    if (getConfiguration().isMultiPartUpload() &&
                            uploadAggregate.buffer.size() >= getConfiguration().getPartSize()) {
                        uploadPart(uploadAggregate);
                        maxRead = Math.toIntExact(getConfiguration().getPartSize());
                        continue;
                    }
                    if (uploadAggregate.buffer.size() >= getConfiguration().getBatchSize()
                            || (uploadAggregate.index >= getConfiguration().getBatchMessageNumber()
                                    && uploadAggregate.buffer.size() < getConfiguration().getPartSize())) {

                        if (uploadAggregate.buffer.size() > 0)
                            uploadPart(uploadAggregate);
                        CompleteMultipartUploadResponse uploadResult = completeUpload(uploadAggregate);
                        this.uploadAggregate = null;
                        Message message = getMessageForResponse(exchange);
                        message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
                        if (uploadResult.versionId() != null) {
                            message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
                        }
                        populateHttpResponseCode(uploadResult, message);
                    }
                    continue;
                }
            } finally {
                lock.unlock();
            }
            if (state == null) {
                state = new UploadState();
            } else {
                state.index = 1;
            }
            state.buffer.write(b);

            final String keyName = getConfiguration().getKeyName();
            final String fileName = AWS2S3Utils.determineFileName(keyName);
            final String extension = AWS2S3Utils.determineFileExtension(keyName);
            if (state.index == 1 && getConfiguration().getNamingStrategy().equals(AWSS3NamingStrategyEnum.random)) {
                state.id = UUID.randomUUID();
            }
            if (state.index == 1 && getConfiguration().getNamingStrategy().equals(AWSS3NamingStrategyEnum.timestamp)) {
                state.timestamp = System.currentTimeMillis();
            }
            state.dynamicKeyName = fileNameToUpload(fileName, getConfiguration().getNamingStrategy(), extension,
                    state.part, state.id, state.timestamp);
            CreateMultipartUploadRequest.Builder createMultipartUploadRequest
                    = CreateMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
                            .key(state.dynamicKeyName).checksumAlgorithm(algorithm);

            String storageClass = AWS2S3Utils.determineStorageClass(exchange, getConfiguration());
            if (storageClass != null) {
                createMultipartUploadRequest.storageClass(storageClass);
            }

            String cannedAcl = exchange.getIn().getHeader(AWS2S3Constants.CANNED_ACL, String.class);
            if (cannedAcl != null) {
                ObjectCannedACL objectAcl = ObjectCannedACL.valueOf(cannedAcl);
                createMultipartUploadRequest.acl(objectAcl);
            }

            BucketCannedACL acl = exchange.getIn().getHeader(AWS2S3Constants.ACL, BucketCannedACL.class);
            if (acl != null) {
                // note: if cannedacl and acl are both specified the last one will
                // be used. refer to
                // PutObjectRequest#setAccessControlList for more details
                createMultipartUploadRequest.acl(acl.toString());
            }

            AWS2S3Utils.setEncryption(createMultipartUploadRequest, getConfiguration());

            LOG.trace("Initiating multipart upload [{}] from exchange [{}]...", createMultipartUploadRequest, exchange);
            if (state.index == 1) {
                state.initResponse
                        = getEndpoint().getS3Client().createMultipartUpload(createMultipartUploadRequest.build());
            }

            try {
                if (totalSize >= getConfiguration().getBatchSize()
                        || state.buffer.size() >= getConfiguration().getBatchSize()
                        || state.index >= getConfiguration().getBatchMessageNumber()) {

                    uploadPart(state);
                    CompleteMultipartUploadResponse uploadResult = completeUpload(state);
                    Message message = getMessageForResponse(exchange);
                    message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
                    if (uploadResult.versionId() != null) {
                        message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
                    }
                    populateHttpResponseCode(uploadResult, message);
                    state = null;
                    continue;
                }
                if (getConfiguration().isMultiPartUpload() && state.buffer.size() >= getConfiguration().getPartSize()) {
                    uploadPart(state);
                    maxRead = Math.toIntExact(getConfiguration().getPartSize());
                }

            } catch (Exception e) {
                getEndpoint().getS3Client()
                        .abortMultipartUpload(AbortMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
                                .key(state.dynamicKeyName).uploadId(state.initResponse.uploadId()).build());
                throw e;
            }
        }

        if (ObjectHelper.isNotEmpty(state)) {
            // exchange wasn't large enough to send, batch it with subsequent exchanges.
            lock.lock();
            try {
                if (ObjectHelper.isEmpty(this.uploadAggregate)) {
                    this.uploadAggregate = state;
                } else {
                    // handle potential race condition.
                    this.uploadAggregate.buffer.write(state.buffer.toByteArray());
                }
            } finally {
                lock.unlock();
            }
        }
    }

    private void processWithTimestampGrouping(final Exchange exchange) throws Exception {
        InputStream is = exchange.getIn().getMandatoryBody(InputStream.class);

        // Extract timestamp from exchange
        Long messageTimestamp = extractTimestampFromExchange(exchange);
        if (messageTimestamp == null) {
            LOG.warn("No valid timestamp found in exchange header '{}', falling back to current time",
                    getConfiguration().getTimestampHeaderName());
            messageTimestamp = System.currentTimeMillis();
        }

        // Calculate the timestamp window
        long timestampWindow = getTimestampWindow(messageTimestamp);

        byte[] b;
        int maxRead = (getConfiguration().isMultiPartUpload()
                ? Math.toIntExact(getConfiguration().getPartSize()) : getConfiguration().getBufferSize());

        // Get or create upload state for this timestamp window
        UploadState state = timestampBasedUploads.get(timestampWindow);
        if (state != null) {
            state.index++;
            maxRead -= state.buffer.size();
        }

        while ((b = AWS2S3Utils.toByteArray(is, maxRead)) != null && b.length > 0) {
            if (getConfiguration().isMultiPartUpload())
                maxRead -= b.length;

            lock.lock();
            try {
                // Get or create upload state for this timestamp window
                if (!timestampBasedUploads.containsKey(timestampWindow)) {
                    UploadState newState = new UploadState();

                    // Generate timestamp-based filename
                    final String keyName = getConfiguration().getKeyName();
                    final String fileName = AWS2S3Utils.determineFileName(keyName);
                    final String extension = AWS2S3Utils.determineFileExtension(keyName);
                    newState.dynamicKeyName = generateTimestampBasedFileName(fileName, extension, timestampWindow);

                    // Initialize multipart upload
                    CreateMultipartUploadRequest.Builder createMultipartUploadRequest
                            = CreateMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
                                    .key(newState.dynamicKeyName).checksumAlgorithm(algorithm);

                    String storageClass = AWS2S3Utils.determineStorageClass(exchange, getConfiguration());
                    if (storageClass != null) {
                        createMultipartUploadRequest.storageClass(storageClass);
                    }

                    String cannedAcl = exchange.getIn().getHeader(AWS2S3Constants.CANNED_ACL, String.class);
                    if (cannedAcl != null) {
                        ObjectCannedACL objectAcl = ObjectCannedACL.valueOf(cannedAcl);
                        createMultipartUploadRequest.acl(objectAcl);
                    }

                    BucketCannedACL acl = exchange.getIn().getHeader(AWS2S3Constants.ACL, BucketCannedACL.class);
                    if (acl != null) {
                        createMultipartUploadRequest.acl(acl.toString());
                    }

                    AWS2S3Utils.setEncryption(createMultipartUploadRequest, getConfiguration());

                    LOG.trace("Initiating multipart upload [{}] for timestamp window {}", createMultipartUploadRequest,
                            timestampWindow);
                    newState.initResponse
                            = getEndpoint().getS3Client().createMultipartUpload(createMultipartUploadRequest.build());

                    timestampBasedUploads.put(timestampWindow, newState);
                }
                state = timestampBasedUploads.get(timestampWindow);

                state.buffer.write(b);

                if (getConfiguration().isMultiPartUpload() && state.buffer.size() >= getConfiguration().getPartSize()) {
                    uploadPart(state);
                    maxRead = Math.toIntExact(getConfiguration().getPartSize());
                    continue;
                }

                if (state.buffer.size() >= getConfiguration().getBatchSize()
                        || (state.index >= getConfiguration().getBatchMessageNumber()
                                && state.buffer.size() < getConfiguration().getPartSize())) {

                    if (state.buffer.size() > 0)
                        uploadPart(state);
                    CompleteMultipartUploadResponse uploadResult = completeUpload(state);
                    timestampBasedUploads.remove(timestampWindow);

                    Message message = getMessageForResponse(exchange);
                    message.setHeader(AWS2S3Constants.E_TAG, uploadResult.eTag());
                    if (uploadResult.versionId() != null) {
                        message.setHeader(AWS2S3Constants.VERSION_ID, uploadResult.versionId());
                    }
                    populateHttpResponseCode(uploadResult, message);
                }

            } catch (Exception e) {
                if (state != null && state.initResponse != null) {
                    getEndpoint().getS3Client()
                            .abortMultipartUpload(
                                    AbortMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
                                            .key(state.dynamicKeyName).uploadId(state.initResponse.uploadId()).build());
                    timestampBasedUploads.remove(timestampWindow);
                }
                throw e;
            } finally {
                lock.unlock();
            }
        }
    }

    private CompleteMultipartUploadResponse completeUpload(UploadState state) {
        CompletedMultipartUpload completeMultipartUpload
                = CompletedMultipartUpload.builder().parts(state.completedParts).build();
        CompleteMultipartUploadRequest compRequest
                = CompleteMultipartUploadRequest.builder().multipartUpload(completeMultipartUpload)
                        .bucket(getConfiguration().getBucketName()).key(state.dynamicKeyName)
                        .uploadId(state.initResponse.uploadId())
                        .build();

        try {
            final CompleteMultipartUploadResponse uploadResult
                    = getEndpoint().getS3Client().completeMultipartUpload(compRequest);

            // Converting the index to String can cause extra overhead
            if (LOG.isDebugEnabled()) {
                LOG.debug("Completed upload for the part {}, multipart {} with etag {} at index {}", part, state.multipartIndex,
                        uploadResult.eTag(),
                        state.index);
            }
            part.getAndIncrement();
            return uploadResult;
        } catch (Exception e) {
            LOG.warn("Error completing multipart upload - Multipart upload will be aborted", e);
            getEndpoint().getS3Client()
                    .abortMultipartUpload(AbortMultipartUploadRequest.builder().bucket(getConfiguration().getBucketName())
                            .key(state.dynamicKeyName).uploadId(state.initResponse.uploadId()).build());
            throw e;
        }
    }

    private void uploadPart(UploadState state) {
        UploadPartRequest uploadRequest = UploadPartRequest.builder().bucket(getConfiguration().getBucketName())
                .key(state.dynamicKeyName).uploadId(state.initResponse.uploadId())
                .partNumber(state.multipartIndex).checksumAlgorithm(algorithm).build();

        LOG.trace("Uploading part {}, multipart {} at index {} for {}", state.part, state.multipartIndex, state.index,
                getConfiguration().getKeyName());

        UploadPartResponse partResponse = getEndpoint().getS3Client()
                .uploadPart(uploadRequest, RequestBody.fromBytes(state.buffer.toByteArray()));
        CompletedPart partUpload = CompletedPart.builder().partNumber(state.multipartIndex)
                .checksumCRC32(partResponse.checksumCRC32()).eTag(partResponse.eTag()).build();
        state.completedParts.add(partUpload);
        state.buffer.reset();
        state.multipartIndex++;
    }

    private String fileNameToUpload(
            String fileName, AWSS3NamingStrategyEnum strategy, String ext, int part, UUID id, long timestamp) {
        String dynamicKeyName;
        switch (strategy) {
            case progressive:
                if (part > 0) {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + "-" + part + ext;
                    } else {
                        dynamicKeyName = fileName + "-" + part;
                    }
                } else {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + ext;
                    } else {
                        dynamicKeyName = fileName;
                    }
                }
                break;
            case random:
                if (part > 0) {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + "-" + id.toString() + ext;
                    } else {
                        dynamicKeyName = fileName + "-" + id.toString();
                    }
                } else {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + ext;
                    } else {
                        dynamicKeyName = fileName;
                    }
                }
                break;
            case timestamp:
                if (part > 0) {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + "-" + timestamp + ext;
                    } else {
                        dynamicKeyName = fileName + "-" + timestamp;
                    }
                } else {
                    if (ObjectHelper.isNotEmpty(ext)) {
                        dynamicKeyName = fileName + ext;
                    } else {
                        dynamicKeyName = fileName;
                    }
                }
                break;
            default:
                throw new IllegalArgumentException("Unsupported operation");
        }
        return dynamicKeyName;
    }

    private long getTimestampWindow(long timestamp) {
        long windowSize = getConfiguration().getTimestampWindowSizeMillis();
        return (timestamp / windowSize) * windowSize;
    }

    private String generateTimestampBasedFileName(String baseFileName, String extension, long timestampWindow) {
        Date windowStart = new Date(timestampWindow);
        Date windowEnd = new Date(timestampWindow + getConfiguration().getTimestampWindowSizeMillis());

        // Use seconds precision for windows smaller than 1 minute, minutes precision otherwise
        String datePattern, timePattern;
        if (getConfiguration().getTimestampWindowSizeMillis() < 60000) {
            datePattern = "yyyyMMdd_HHmmss";
            timePattern = "HHmmss";
        } else {
            datePattern = "yyyyMMdd_HHmm";
            timePattern = "HHmm";
        }

        SimpleDateFormat sdf = new SimpleDateFormat(datePattern);
        sdf.setTimeZone(TimeZone.getTimeZone("UTC"));

        SimpleDateFormat timeFormat = new SimpleDateFormat(timePattern);
        timeFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
        String timeRange = timeFormat.format(windowStart) + "-" + timeFormat.format(windowEnd);

        if (extension != null && !extension.isEmpty()) {
            return baseFileName + "_" + sdf.format(windowStart) + "_" + timeRange + extension;
        } else {
            return baseFileName + "_" + sdf.format(windowStart) + "_" + timeRange;
        }
    }

    private Long extractTimestampFromExchange(Exchange exchange) {
        String headerName = getConfiguration().getTimestampHeaderName();
        Object timestampObj = exchange.getIn().getHeader(headerName);

        if (timestampObj instanceof Long ts) {
            return ts;
        } else if (timestampObj instanceof Date date) {
            return date.getTime();
        } else if (timestampObj instanceof String str) {
            try {
                return Long.parseLong(str);
            } catch (NumberFormatException e) {
                LOG.warn("Cannot parse timestamp header '{}' with value '{}'", headerName, timestampObj);
                return null;
            }
        } else if (timestampObj != null) {
            LOG.warn("Unsupported timestamp header type: {} for header '{}'", timestampObj.getClass(), headerName);
        }

        return null;
    }

    private void setStartingPart() {
        if (getConfiguration().getNamingStrategy().equals(AWSS3NamingStrategyEnum.progressive)) {
            ArrayList<S3Object> list = new ArrayList<>();
            ListObjectsV2Request request = ListObjectsV2Request.builder().bucket(getConfiguration().getBucketName())
                    .prefix(AWS2S3Utils.determineFileName(getConfiguration().getKeyName())).build();
            ListObjectsV2Iterable listRes = getEndpoint().getS3Client().listObjectsV2Paginator(request);
            listRes.stream()
                    .flatMap(r -> r.contents().stream())
                    .forEach(content -> list.add(content));
            if (!list.isEmpty()) {
                list.sort(Comparator.comparing(S3Object::lastModified));
                int listSize = list.size();
                String fileName = AWS2S3Utils.determineFileName(list.get(listSize - 1).key());
                int position = fileName.lastIndexOf("-");
                if (position != -1) {
                    String partString = fileName.substring(position + 1);
                    if (ObjectHelper.isNotEmpty(partString)) {
                        part.getAndSet(Integer.parseInt(partString) + 1);
                    }
                } else {
                    part.getAndSet(1);
                }
            }
        } else {
            LOG.info("lastPart restarting policy can be used only with progressive naming strategy");
        }
    }

    protected AWS2S3Configuration getConfiguration() {
        return getEndpoint().getConfiguration();
    }

    @Override
    public String toString() {
        if (s3ProducerToString == null) {
            s3ProducerToString = "AWS2S3StreamUploadProducer[" + URISupport.sanitizeUri(getEndpoint().getEndpointUri()) + "]";
        }
        return s3ProducerToString;
    }

    @Override
    public AWS2S3Endpoint getEndpoint() {
        return (AWS2S3Endpoint) super.getEndpoint();
    }

    public static Message getMessageForResponse(final Exchange exchange) {
        return exchange.getMessage();
    }

    private static void populateHttpResponseCode(AwsResponse response, Message message) {
        if (response != null && response.sdkHttpResponse() != null) {
            message.setHeader(Exchange.HTTP_RESPONSE_CODE, response.sdkHttpResponse().statusCode());
        }
    }

    private class UploadState {
        int index;
        int multipartIndex;
        int part;
        List<CompletedPart> completedParts = new ArrayList<>();
        ByteArrayOutputStream buffer;
        String dynamicKeyName;
        UUID id;
        long timestamp;
        CreateMultipartUploadResponse initResponse;

        UploadState() {
            index = 1;
            multipartIndex = 1;
            part = AWS2S3StreamUploadProducer.this.part.get();
            buffer = new ByteArrayOutputStream();
        }
    }
}
