fix progress bug in resumable upload/download.

Author: huiguangjun

sdk/src/resumable/ResumableDownloader.cc

@@ -29,6 +29,11 @@
 
 using namespace AlibabaCloud::OSS;
 
+struct DownloaderTransferState {
+    int64_t transfered;
+    void *userData;
+};
+
 GetObjectOutcome ResumableDownloader::Download() 
 {
     OssError err;
@@ -79,9 +84,12 @@ GetObjectOutcome ResumableDownloader::Download()
                 getObjectReq.setRange(start, end);
                 getObjectReq.setFlags(getObjectReq.Flags() | REQUEST_FLAG_CHECK_CRC64 | REQUEST_FLAG_SAVE_CLIENT_CRC64);
 
+                DownloaderTransferState transferState;
                 auto process = request_.TransferProgress();
                 if (process.Handler) {
-                    TransferProgress uploadPartProcess = { DownloadPartProcessCallback, (void *)this };
+                    transferState.transfered = 0;
+                    transferState.userData = (void *)this;
+                    TransferProgress uploadPartProcess = { DownloadPartProcessCallback, (void *)&transferState };
                     getObjectReq.setTransferProgress(uploadPartProcess);
                 }
                 if (request_.RequestPayer() == RequestPayer::Requester) {
@@ -468,10 +476,14 @@ void ResumableDownloader::initRecord()
 
 void ResumableDownloader::DownloadPartProcessCallback(size_t increment, int64_t transfered, int64_t total, void *userData) 
 {
-    UNUSED_PARAM(transfered);
     UNUSED_PARAM(total);
+    auto transferState = (DownloaderTransferState *)userData;
+    auto downloader = (ResumableDownloader*)transferState->userData;
+    auto inc = transfered - transferState->transfered;
+    transferState->transfered = std::max(transfered, transferState->transfered);
+    inc = std::max(inc, static_cast<int64_t>(0));
+    increment = static_cast<size_t>(inc);
 
-    auto downloader = (ResumableDownloader*)userData;
     std::lock_guard<std::mutex> lck(downloader->lock_);
     downloader->consumedSize_ += increment;
 

sdk/src/resumable/ResumableUploader.cc

@@ -35,6 +35,10 @@
 
 using namespace AlibabaCloud::OSS;
 
+struct UploaderTransferState {
+    int64_t transfered;
+    void *userData;
+};
 
 ResumableUploader::ResumableUploader(const UploadObjectRequest& request, const OssClientImpl *client) :
     ResumableBaseWorker(request.ObjectSize(), request.PartSize()),
@@ -101,9 +105,12 @@ PutObjectOutcome ResumableUploader::Upload()
                 UploadPartRequest uploadPartRequest(request_.Bucket(), request_.Key(), part.PartNumber(), uploadID_, content);
                 uploadPartRequest.setContentLength(length);
 
+                UploaderTransferState transferState;
                 auto process = request_.TransferProgress();
                 if (process.Handler) {
-                    TransferProgress uploadPartProcess = { UploadPartProcessCallback, (void *)this };
+                    transferState.transfered = 0;
+                    transferState.userData = (void *)this;
+                    TransferProgress uploadPartProcess = { UploadPartProcessCallback, (void *)&transferState };
                     uploadPartRequest.setTransferProgress(uploadPartProcess);
                 }
                 if (request_.RequestPayer() == RequestPayer::Requester) {
@@ -409,10 +416,14 @@ void ResumableUploader::dumpRecordInfo(AlibabaCloud::OSS::Json::Value& root)
 
 void ResumableUploader::UploadPartProcessCallback(size_t increment, int64_t transfered, int64_t total, void *userData) 
 {
-    UNUSED_PARAM(transfered);
     UNUSED_PARAM(total);
+    auto transferState = (UploaderTransferState *)userData;
+    auto uploader = (ResumableUploader*)transferState->userData;
+    auto inc = transfered - transferState->transfered;
+    transferState->transfered = std::max(transfered, transferState->transfered);
+    inc = std::max(inc, static_cast<int64_t>(0));
+    increment = static_cast<size_t>(inc);
 
-    auto uploader = (ResumableUploader*)userData;
     std::lock_guard<std::mutex> lck(uploader->lock_);
     uploader->consumedSize_ += increment;
 

test/src/MultipartUpload/ResumableObjectTest.cc

@@ -664,6 +664,7 @@ class ResumableObjectTest : public::testing::Test {
         EXPECT_EQ(CreateDirectory(checkpointDir), true);
         EXPECT_EQ(IsDirectoryExist(checkpointDir), true);
 
+        std::cout << "this ptr:" << this << std::endl;
         TransferProgress progressCallback = { ProgressCallback, this };
         UploadObjectRequest request(BucketName, key, tmpFile, checkpointDir, 102400, 1);
         request.setTransferProgress(progressCallback);
@@ -1325,6 +1326,7 @@ class ResumableObjectTest : public::testing::Test {
         EXPECT_EQ(putObjectOutcome.isSuccess(), true);
         EXPECT_EQ(Client->DoesObjectExist(BucketName, sourceKey), true);
 
+        std::cout << "this ptr:" << this << std::endl;
         TransferProgress progressCallback = { ProgressCallback, this };
         DownloadObjectRequest request(BucketName, sourceKey, targetKey);
         request.setTransferProgress(progressCallback);
@@ -2354,7 +2356,7 @@ class ResumableObjectTest : public::testing::Test {
         std::string checkpointDir = TestUtils::GetTargetFileName("checkpoint");
         EXPECT_EQ(CreateDirectory(checkpointDir), true);
         EXPECT_EQ(IsDirectoryExist(checkpointDir), true);
-
+        std::cout << "this ptr:" << this << std::endl;
         TransferProgress progressCallback = { ProgressCallback, this };
         UploadObjectRequest request(BucketName, key, tmpFile, checkpointDir, 102400, 1);
         request.setTransferProgress(progressCallback);