TransferHandle.h

 
#pragma once
 
#include <aws/transfer/Transfer_EXPORTS.h>
#include <aws/core/utils/memory/stl/AWSString.h>
#include <aws/core/utils/memory/stl/AWSSet.h>
#include <aws/core/utils/memory/stl/AWSMap.h>
#include <aws/core/utils/UUID.h>
#include <aws/core/client/AWSError.h>
#include <aws/core/client/AsyncCallerContext.h>
#include <aws/s3/S3Errors.h>
#include <iostream>
#include <atomic>
#include <mutex>
#include <condition_variable>
 
namespace Aws
{
    namespace Utils
    {
        template < typename T > class Array;
    }
 
    namespace Transfer
    {
        class TransferHandle;
 
        typedef std::function<Aws::IOStream*(void)> CreateDownloadStreamCallback;
 
        static const char CLASS_TAG[] = "TransferManager";
 
        struct DownloadConfiguration
        {
            DownloadConfiguration() :
                versionId("")
            {}
 
            Aws::String versionId;
 
            // TBI: controls for in-memory parts vs. resumable file-based parts with state serialization to/from file
        };
 
        class AWS_TRANSFER_API PartState
        {
            public:
                PartState();
                PartState(int partId, uint64_t bestProgressInBytes, uint64_t sizeInBytes, bool lastPart = false);
 
                int GetPartId() const { return m_partId; }
 
                uint64_t GetBestProgressInBytes() const { return m_bestProgressInBytes; }
                void SetBestProgressInBytes(uint64_t progressInBytes) { m_bestProgressInBytes = progressInBytes; }
 
                uint64_t GetSizeInBytes() const { return m_sizeInBytes; }
                void SetSizeInBytes(uint64_t sizeInBytes) { m_sizeInBytes = sizeInBytes; }
 
                void Reset();
 
                void OnDataTransferred(uint64_t amount, const std::shared_ptr<TransferHandle> &transferHandle);
 
                void SetETag(const Aws::String& eTag) { m_eTag = eTag; }
                const Aws::String& GetETag() const { return m_eTag; }
 
                Aws::IOStream *GetDownloadPartStream() const { return m_downloadPartStream; }
                void SetDownloadPartStream(Aws::IOStream *downloadPartStream) { m_downloadPartStream = downloadPartStream; }
 
                unsigned char* GetDownloadBuffer() const { return m_downloadBuffer; }
                void SetDownloadBuffer(unsigned char* downloadBuffer) { m_downloadBuffer = downloadBuffer; }
 
                void SetRangeBegin(uint64_t rangeBegin) { m_rangeBegin = rangeBegin; }
                uint64_t GetRangeBegin() const { return m_rangeBegin; }
 
                bool IsLastPart() { return m_lastPart; }
                void SetLastPart() { m_lastPart = true; }
 
                Aws::String GetChecksum() const { return m_checksum; };
                void SetChecksum(const Aws::String& checksum) { m_checksum = checksum; }
            private:
 
                int m_partId = 0;
 
                Aws::String m_eTag;
                uint64_t m_currentProgressInBytes = 0;
                uint64_t m_bestProgressInBytes = 0;
                uint64_t m_sizeInBytes = 0;
                uint64_t m_rangeBegin = 0;
 
                std::atomic<Aws::IOStream *> m_downloadPartStream;
                std::atomic<unsigned char*> m_downloadBuffer;
                bool m_lastPart = false;
                Aws::String m_checksum;
        };
 
        using PartPointer = std::shared_ptr< PartState >;
        using PartStateMap = Aws::Map< int, PartPointer >;
 
        enum class TransferStatus
        {
            //this value is only used for directory synchronization
            EXACT_OBJECT_ALREADY_EXISTS,
            //Operation is still queued and has not begun processing
            NOT_STARTED,
            //Operation is now running
            IN_PROGRESS,
            //Operation was canceled. A Canceled operation can still be retried
            CANCELED,
            //Operation failed, A failed operation can still be retried.
            FAILED,
            //Operation was successful
            COMPLETED,
            //Operation either failed or was canceled and a user deleted the multipart upload from S3.
            ABORTED
        };
 
        enum class TransferDirection
        {
            UPLOAD,
            DOWNLOAD
        };
 
        class AWS_TRANSFER_API TransferHandle
        {
        public:
            TransferHandle(const Aws::String& bucketName, const Aws::String& keyName, uint64_t totalSize, const Aws::String& targetFilePath = "");
 
            TransferHandle(const Aws::String& bucketName, const Aws::String& keyName, const Aws::String& targetFilePath = "");
 
            TransferHandle(const Aws::String& bucketName, const Aws::String& keyName, CreateDownloadStreamCallback createDownloadStreamFn, const Aws::String& targetFilePath = "");
 
            TransferHandle(const Aws::String& bucketName, const Aws::String& keyName,
                const uint64_t fileOffset, const uint64_t downloadBytes,
                CreateDownloadStreamCallback createDownloadStreamFn, const Aws::String& targetFilePath = "");
 
 
            ~TransferHandle();
 
            inline bool IsMultipart() const { return m_isMultipart.load(); }
            inline void SetIsMultipart(bool value) { m_isMultipart.store(value); }
            inline const Aws::String GetMultiPartId() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_multipartId; }
            inline void SetMultipartId(const Aws::String& value) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_multipartId = value; }
            PartStateMap GetCompletedParts() const;
            void ChangePartToCompleted(const PartPointer& partState, const Aws::String &eTag);
            PartStateMap GetPendingParts() const;
            bool HasPendingParts() const;
            void AddPendingPart(const PartPointer& partState);
            PartStateMap GetQueuedParts() const;
            bool HasQueuedParts() const;
            void AddQueuedPart(const PartPointer& partState);
            PartStateMap GetFailedParts() const;
            bool HasFailedParts() const;
            void ChangePartToFailed(const PartPointer& partState);
            void GetAllPartsTransactional(PartStateMap& queuedParts, PartStateMap& pendingParts,
                    PartStateMap& failedParts, PartStateMap& completedParts);
            bool HasParts() const;
            bool ShouldContinue() const;
            void Cancel();
 
            void Restart();
            inline uint64_t GetBytesTransferred() const { return m_bytesTransferred.load(); }
            void UpdateBytesTransferred(uint64_t amount) { m_bytesTransferred.fetch_add(amount); }
 
            inline uint64_t GetBytesOffset() const { return m_offset; }
            inline uint64_t GetBytesTotalSize() const { return m_bytesTotalSize.load(); }
            inline void SetBytesTotalSize(uint64_t value) { m_bytesTotalSize.store(value); }
 
            inline uint64_t GetBytesAvailableFromStart() const { return m_bytesAvailableFromStart.load(std::memory_order_relaxed); }
 
            inline const Aws::String& GetBucketName() const { return m_bucket; }
            inline const Aws::String& GetKey() const { return m_key; }
            inline const Aws::String& GetTargetFilePath() const { return m_fileName; }
 
            const Aws::String GetVersionId() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_versionId; }
            void SetVersionId(const Aws::String& versionId) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_versionId = versionId; }
 
            inline TransferDirection GetTransferDirection() const { return m_direction; }
            inline const Aws::String GetContentType() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_contentType; }
            inline void SetContentType(const Aws::String& value) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_contentType = value; }
            inline const Aws::Map<Aws::String, Aws::String> GetMetadata() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_metadata; }
            inline void SetMetadata(const Aws::Map<Aws::String, Aws::String>& value) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_metadata = value; }
 
            inline void AddMetadataEntry(const Aws::String& key, const Aws::String& value) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_metadata[key] = value; }
 
            inline void SetContext(const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_context = context; }
 
            inline std::shared_ptr<const Aws::Client::AsyncCallerContext> GetContext() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_context; }
 
            TransferStatus GetStatus() const;
            void UpdateStatus(TransferStatus value);
 
            inline const Aws::Client::AWSError<Aws::S3::S3Errors> GetLastError() const { std::lock_guard<std::mutex> locker(m_getterSetterLock); return m_lastError; }
            inline void SetError(const Aws::Client::AWSError<Aws::S3::S3Errors>& error) { std::lock_guard<std::mutex> locker(m_getterSetterLock); m_lastError = error; }
            void WaitUntilFinished() const;
 
            const CreateDownloadStreamCallback& GetCreateDownloadStreamFunction() const { return m_createDownloadStreamFn; }
 
            Aws::String WritePartToDownloadStream(Aws::IOStream* partStream, uint64_t writeOffset);
 
            void ApplyDownloadConfiguration(const DownloadConfiguration& downloadConfig);
 
            bool LockForCompletion()
            {
                bool expected = false;
                return m_lastPart.compare_exchange_strong(expected, true/*desired*/);
            }
 
            /*
             * Returns a unique identifier tied to this particular transfer handle.
             */
            Aws::String GetId() const;
 
        private:
 
            void CleanupDownloadStream();
 
            std::atomic<bool> m_isMultipart;
            Aws::String m_multipartId;
            TransferDirection m_direction;
            PartStateMap m_completedParts;
            PartStateMap m_pendingParts;
            PartStateMap m_queuedParts;
            PartStateMap m_failedParts;
            std::atomic<uint64_t> m_bytesTransferred;
            std::atomic<bool> m_lastPart;
            std::atomic<uint64_t> m_bytesTotalSize;
            std::atomic<uint64_t> m_bytesAvailableFromStart;
            /* The next part number to watch, that is able to grow m_bytesAvailableFromStart. */
            uint32_t m_nextPartToWatch;
            uint64_t m_offset;
            Aws::String m_bucket;
            Aws::String m_key;
            Aws::String m_fileName;
            Aws::String m_contentType;
            Aws::String m_versionId;
            Aws::Map<Aws::String, Aws::String> m_metadata;
            TransferStatus m_status;
            Aws::Client::AWSError<Aws::S3::S3Errors> m_lastError;
            std::atomic<bool> m_cancel;
            std::shared_ptr<const Aws::Client::AsyncCallerContext> m_context;
            const Utils::UUID m_handleId;
 
            CreateDownloadStreamCallback m_createDownloadStreamFn;
            Aws::IOStream* m_downloadStream;
            /* in case customer stream is not based off 0 */
            uint64_t m_downloadStreamBaseOffset;
 
            mutable std::mutex m_downloadStreamLock;
            mutable std::mutex m_partsLock;
            mutable std::mutex m_statusLock;
            mutable std::condition_variable m_waitUntilFinishedSignal;
            mutable std::mutex m_getterSetterLock;
        };
 
        AWS_TRANSFER_API Aws::OStream& operator << (Aws::OStream& s, TransferStatus status);
    }
}