TransferManager.h

 
#pragma once
 
#include <aws/transfer/TransferHandle.h>
#include <aws/s3/S3Client.h>
#include <aws/s3/model/PutObjectRequest.h>
#include <aws/s3/model/GetObjectRequest.h>
#include <aws/s3/model/HeadObjectRequest.h>
#include <aws/s3/model/CreateMultipartUploadRequest.h>
#include <aws/s3/model/UploadPartRequest.h>
#include <aws/s3/model/CompletedPart.h>
#include <aws/core/utils/threading/Executor.h>
#include <aws/core/utils/memory/stl/AWSStreamFwd.h>
#include <aws/core/utils/ResourceManager.h>
#include <aws/core/client/AsyncCallerContext.h>
 
#include <memory>
 
namespace Aws
{    
    namespace Transfer
    {
        class TransferManager;
 
        typedef std::function<void(const TransferManager*, const std::shared_ptr<const TransferHandle>&)> UploadProgressCallback;
        typedef std::function<void(const TransferManager*, const std::shared_ptr<const TransferHandle>&)> DownloadProgressCallback;
        typedef std::function<void(const TransferManager*, const std::shared_ptr<const TransferHandle>&)> TransferStatusUpdatedCallback;
        typedef std::function<void(const TransferManager*, const std::shared_ptr<const TransferHandle>&, const Aws::Client::AWSError<Aws::S3::S3Errors>&)> ErrorCallback;
        typedef std::function<void(const TransferManager*, const std::shared_ptr<const TransferHandle>&)> TransferInitiatedCallback;
 
        const uint64_t MB5 = 5 * 1024 * 1024;
 
        struct TransferManagerConfiguration
        {
            TransferManagerConfiguration(Aws::Utils::Threading::Executor* executor)
              : s3Client(nullptr),
                transferExecutor(executor),
                computeContentMD5(false),
                transferBufferMaxHeapSize(10 * MB5),
                bufferSize(MB5)
            {
            }
            std::shared_ptr<Aws::S3::S3Client> s3Client;
            Aws::Utils::Threading::Executor* transferExecutor = nullptr;
 
            std::shared_ptr<Aws::Utils::Threading::Executor> spExecutor = nullptr;
 
            std::function<std::shared_ptr<Utils::Threading::Executor>()> executorCreateFn;
 
            bool computeContentMD5 = false;
            Aws::S3::Model::PutObjectRequest putObjectTemplate;
            Aws::S3::Model::GetObjectRequest getObjectTemplate;
            Aws::S3::Model::HeadObjectRequest headObjectTemplate;
            Aws::S3::Model::CreateMultipartUploadRequest createMultipartUploadTemplate;
            Aws::S3::Model::UploadPartRequest uploadPartTemplate;
            uint64_t transferBufferMaxHeapSize = 10 * MB5;
            uint64_t bufferSize = MB5;
 
            UploadProgressCallback uploadProgressCallback;
            DownloadProgressCallback downloadProgressCallback;
            TransferStatusUpdatedCallback transferStatusUpdatedCallback;
            TransferInitiatedCallback transferInitiatedCallback;
            ErrorCallback errorCallback;
            Aws::Map<Aws::String, Aws::String> customizedAccessLogTag;
 
            Aws::S3::Model::ChecksumAlgorithm checksumAlgorithm = S3::Model::ChecksumAlgorithm::CRC64NVME;
        };        
 
        class AWS_TRANSFER_API TransferManager : public std::enable_shared_from_this<TransferManager>
        {
        public:
            static std::shared_ptr<TransferManager> Create(const TransferManagerConfiguration& config);
 
            TransferStatus WaitUntilAllFinished(int64_t timeoutMs = std::numeric_limits<int64_t>::max());
 
            void CancelAll();
 
            ~TransferManager();
 
            std::shared_ptr<TransferHandle> UploadFile(const Aws::String& fileName, const Aws::String& bucketName,
                                                       const Aws::String& keyName, const Aws::String& contentType,
                                                       const Aws::Map<Aws::String, Aws::String>& metadata,
                                                       const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr,
                                                       const Aws::String& precalculatedChecksum = {});
 
            std::shared_ptr<TransferHandle> UploadFile(const std::shared_ptr<Aws::IOStream>& stream, const Aws::String& bucketName,
                                                       const Aws::String& keyName, const Aws::String& contentType,
                                                       const Aws::Map<Aws::String, Aws::String>& metadata,
                                                       const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr,
                                                       const Aws::String& precalculatedChecksum = {});
 
            std::shared_ptr<TransferHandle> DownloadFile(const Aws::String& bucketName,
                                                         const Aws::String& keyName,
                                                         const Aws::String& writeToFile,
                                                         const DownloadConfiguration& downloadConfig = DownloadConfiguration(),
                                                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr);
            std::shared_ptr<TransferHandle> DownloadFile(const Aws::String& bucketName, 
                                                         const Aws::String& keyName, 
                                                         CreateDownloadStreamCallback writeToStreamfn, 
                                                         const DownloadConfiguration& downloadConfig = DownloadConfiguration(),
                                                         const Aws::String& writeToFile = "",
                                                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr);
 
            std::shared_ptr<TransferHandle> DownloadFile(const Aws::String& bucketName, 
                                                         const Aws::String& keyName, 
                                                         uint64_t fileOffset,
                                                         uint64_t downloadBytes,
                                                         CreateDownloadStreamCallback writeToStreamfn, 
                                                         const DownloadConfiguration& downloadConfig = DownloadConfiguration(),
                                                         const Aws::String& writeToFile = "",
                                                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr);
 
            std::shared_ptr<TransferHandle> RetryDownload(const std::shared_ptr<TransferHandle>& retryHandle);
 
            std::shared_ptr<TransferHandle> RetryUpload(const Aws::String& fileName, const std::shared_ptr<TransferHandle>& retryHandle);
            std::shared_ptr<TransferHandle> RetryUpload(const std::shared_ptr<Aws::IOStream>& stream, const std::shared_ptr<TransferHandle>& retryHandle);
            
            void AbortMultipartUpload(const std::shared_ptr<TransferHandle>& inProgressHandle);
 
            void UploadDirectory(const Aws::String& directory, const Aws::String& bucketName, const Aws::String& prefix, const Aws::Map<Aws::String, Aws::String>& metadata);
 
            void DownloadToDirectory(const Aws::String& directory, const Aws::String& bucketName, const Aws::String& prefix = Aws::String());
 
        private:
            TransferManager(const TransferManagerConfiguration& config);
 
            std::shared_ptr<TransferHandle> CreateUploadFileHandle(Aws::IOStream* fileStream, const Aws::String& bucketName,
                                                                   const Aws::String& keyName, const Aws::String& contentType,
                                                                   const Aws::Map<Aws::String, Aws::String>& metadata,
                                                                   const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context,
                                                                   const Aws::String& fileName = "", const Aws::String& checksum = "");
 
            std::shared_ptr<TransferHandle> SubmitUpload(const std::shared_ptr<TransferHandle>& handle, const std::shared_ptr<Aws::IOStream>& fileStream = nullptr);
            
            bool MultipartUploadSupported(uint64_t length) const;
            bool InitializePartsForDownload(const std::shared_ptr<TransferHandle>& handle);
 
            void DoMultiPartUpload(const std::shared_ptr<Aws::IOStream>& streamToPut, const std::shared_ptr<TransferHandle>& handle);
            void DoSinglePartUpload(const std::shared_ptr<Aws::IOStream>& streamToPut, const std::shared_ptr<TransferHandle>& handle);
 
            void DoMultiPartUpload(const std::shared_ptr<TransferHandle>& handle);
            void DoSinglePartUpload(const std::shared_ptr<TransferHandle>& handle);
 
            void DoDownload(const std::shared_ptr<TransferHandle>& handle);
            void DoSinglePartDownload(const std::shared_ptr<TransferHandle>& handle);
 
            void HandleGetObjectResponse(const Aws::S3::S3Client* client, 
                                         const Aws::S3::Model::GetObjectRequest& request,
                                         const Aws::S3::Model::GetObjectOutcome& outcome, 
                                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context);
 
            void WaitForCancellationAndAbortUpload(const std::shared_ptr<TransferHandle>& canceledHandle);
            void HandleUploadPartResponse(const Aws::S3::S3Client*, const Aws::S3::Model::UploadPartRequest&, const Aws::S3::Model::UploadPartOutcome&, const std::shared_ptr<const Aws::Client::AsyncCallerContext>&);
            void HandlePutObjectResponse(const Aws::S3::S3Client*, const Aws::S3::Model::PutObjectRequest&, const Aws::S3::Model::PutObjectOutcome&, const std::shared_ptr<const Aws::Client::AsyncCallerContext>&);
            void HandleListObjectsResponse(const Aws::S3::S3Client*, const Aws::S3::Model::ListObjectsV2Request&, const Aws::S3::Model::ListObjectsV2Outcome&, const std::shared_ptr<const Aws::Client::AsyncCallerContext>&);
 
            TransferStatus DetermineIfFailedOrCanceled(const TransferHandle&) const;
            void TriggerUploadProgressCallback(const std::shared_ptr<const TransferHandle>&) const;
            void TriggerDownloadProgressCallback(const std::shared_ptr<const TransferHandle>&) const;
            void TriggerTransferStatusUpdatedCallback(const std::shared_ptr<const TransferHandle>&) const;
            void TriggerErrorCallback(const std::shared_ptr<const TransferHandle>&, const Aws::Client::AWSError<Aws::S3::S3Errors>& error)const;
 
            void AddTask(std::shared_ptr<TransferHandle> handle);
            void RemoveTask(const std::shared_ptr<TransferHandle>& handle);
 
            void SetChecksumForAlgorithm(const std::shared_ptr<PartState>& state, Aws::S3::Model::CompletedPart& part);
 
            static Aws::String DetermineFilePath(const Aws::String& directory, const Aws::String& prefix, const Aws::String& keyName);
 
            Aws::Utils::ExclusiveOwnershipResourceManager<unsigned char*> m_bufferManager;
            TransferManagerConfiguration m_transferConfig;
 
            mutable Aws::UnorderedSet<std::shared_ptr<TransferHandle>> m_tasks;
            mutable std::condition_variable m_tasksSignal;
            mutable std::mutex m_tasksMutex;
 
        protected:
            static bool IsWithinParentDirectory(Aws::String parentDirectory, Aws::String filePath);
        };
 
        
    }
}