Fix scp issue by forwarding data in chunks (#493)

ig15 · Ubuntu · web-flow · commit 9f027947603f · 2025-03-18T18:29:37.000+05:30
* Send data from local service to the IOT tunnel in chunks

---------

Co-authored-by: Ubuntu &lt;ubuntu@ip-172-31-27-251.us-west-2.compute.internal&gt;
diff --git a/source/fleetprovisioning/FleetProvisioning.cpp b/source/fleetprovisioning/FleetProvisioning.cpp
@@ -120,7 +120,8 @@ bool FleetProvisioning::WriteCSRCertToDirectory(CreateCertificateFromCsrResponse
 bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient identityClient)
 {
     LOG_INFO(TAG, "Provisioning new device certificate and private key using CreateKeysAndCertificate API");
-    auto onKeysAcceptedSubAck = [this](int ioErr) {
+    auto onKeysAcceptedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -132,7 +133,8 @@ bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient i
         keysAcceptedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onKeysRejectedSubAck = [this](int ioErr) {
+    auto onKeysRejectedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -144,7 +146,8 @@ bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient i
         keysRejectedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onKeysPublishSubAck = [this](int ioErr) {
+    auto onKeysPublishSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -156,7 +159,8 @@ bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient i
         keysPublishCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onKeysAccepted = [this](CreateKeysAndCertificateResponse *response, int ioErr) {
+    auto onKeysAccepted = [this](CreateKeysAndCertificateResponse *response, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_INFO(TAG, "CreateKeysAndCertificateResponse certificateId: %s.", response->CertificateId->c_str());
@@ -177,7 +181,8 @@ bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient i
         }
     };
 
-    auto onKeysRejected = [this](ErrorResponse *error, int ioErr) {
+    auto onKeysRejected = [this](ErrorResponse *error, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -252,7 +257,8 @@ bool FleetProvisioning::CreateCertificateAndKey(Iotidentity::IotIdentityClient i
 bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient identityClient)
 {
     LOG_INFO(TAG, "Provisioning new device certificate using CreateCertificateFromCsr API");
-    auto onCsrAcceptedSubAck = [this](int ioErr) {
+    auto onCsrAcceptedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -264,7 +270,8 @@ bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient
         csrAcceptedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onCsrRejectedSubAck = [this](int ioErr) {
+    auto onCsrRejectedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -276,7 +283,8 @@ bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient
         csrRejectedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onCsrPublishSubAck = [this](int ioErr) {
+    auto onCsrPublishSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -288,7 +296,8 @@ bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient
         csrPublishCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onCsrAccepted = [this](CreateCertificateFromCsrResponse *response, int ioErr) {
+    auto onCsrAccepted = [this](CreateCertificateFromCsrResponse *response, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_INFO(TAG, "CreateCertificateFromCsrResponse certificateId: %s. ***", response->CertificateId->c_str());
@@ -309,7 +318,8 @@ bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient
         }
     };
 
-    auto onCsrRejected = [this](ErrorResponse *error, int ioErr) {
+    auto onCsrRejected = [this](ErrorResponse *error, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -387,7 +397,8 @@ bool FleetProvisioning::CreateCertificateUsingCSR(Iotidentity::IotIdentityClient
 }
 bool FleetProvisioning::RegisterThing(Iotidentity::IotIdentityClient identityClient)
 {
-    auto onRegisterAcceptedSubAck = [this](int ioErr) {
+    auto onRegisterAcceptedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -399,7 +410,8 @@ bool FleetProvisioning::RegisterThing(Iotidentity::IotIdentityClient identityCli
         registerAcceptedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onRegisterRejectedSubAck = [this](int ioErr) {
+    auto onRegisterRejectedSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -411,7 +423,8 @@ bool FleetProvisioning::RegisterThing(Iotidentity::IotIdentityClient identityCli
         registerRejectedCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onRegisterPublishSubAck = [this](int ioErr) {
+    auto onRegisterPublishSubAck = [this](int ioErr)
+    {
         if (ioErr != AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
@@ -423,7 +436,8 @@ bool FleetProvisioning::RegisterThing(Iotidentity::IotIdentityClient identityCli
         registerPublishCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onRegisterAccepted = [this](RegisterThingResponse *response, int ioErr) {
+    auto onRegisterAccepted = [this](RegisterThingResponse *response, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_INFO(TAG, "RegisterThingResponse ThingName: %s.", response->ThingName->c_str());
@@ -438,7 +452,8 @@ bool FleetProvisioning::RegisterThing(Iotidentity::IotIdentityClient identityCli
         registerThingCompletedPromise.set_value(ioErr == AWS_OP_SUCCESS);
     };
 
-    auto onRegisterRejected = [this](ErrorResponse *error, int ioErr) {
+    auto onRegisterRejected = [this](ErrorResponse *error, int ioErr)
+    {
         if (ioErr == AWS_OP_SUCCESS)
         {
             LOGM_ERROR(
diff --git a/source/fleetprovisioning/FleetProvisioning.h b/source/fleetprovisioning/FleetProvisioning.h
@@ -278,6 +278,6 @@ namespace Aws
                     bool ObtainCertificateSerialID(const char *certPath);
                 };
             } // namespace FleetProvisioningNS
-        }     // namespace DeviceClient
-    }         // namespace Iot
+        } // namespace DeviceClient
+    } // namespace Iot
 } // namespace Aws
diff --git a/source/jobs/JobsFeature.cpp b/source/jobs/JobsFeature.cpp
@@ -425,7 +425,8 @@ void JobsFeature::publishUpdateJobExecutionStatusWithRetry(
         retryConfig.needStopFlag = nullptr;
     }
 
-    auto publishLambda = [this, data, statusInfo, statusDetails]() -> bool {
+    auto publishLambda = [this, data, statusInfo, statusDetails]() -> bool
+    {
         // We first need to make sure that we haven't previously leaked any promises into our map
         unique_lock<mutex> leakLock(updateJobExecutionPromisesLock);
         for (auto keyPromise = updateJobExecutionPromises.cbegin(); keyPromise != updateJobExecutionPromises.cend();
@@ -513,9 +514,9 @@ void JobsFeature::publishUpdateJobExecutionStatusWithRetry(
         this->updateJobExecutionPromises.erase(clientToken.c_str());
         return finished;
     };
-    std::thread updateJobExecutionThread([retryConfig, publishLambda, onCompleteCallback] {
-        Retry::exponentialBackoff(retryConfig, publishLambda, onCompleteCallback);
-    });
+    std::thread updateJobExecutionThread(
+        [retryConfig, publishLambda, onCompleteCallback]
+        { Retry::exponentialBackoff(retryConfig, publishLambda, onCompleteCallback); });
     updateJobExecutionThread.detach();
 }
 
@@ -608,7 +609,8 @@ bool JobsFeature::isDuplicateNotification(JobExecutionData job)
 
 void JobsFeature::initJob(const JobExecutionData &job)
 {
-    auto shutdownHandler = [this]() -> void {
+    auto shutdownHandler = [this]() -> void
+    {
         handlingJob.store(false);
         if (needStop.load())
         {
@@ -639,7 +641,8 @@ void JobsFeature::executeJob(const Iotjobs::JobExecutionData &job, const PlainJo
 {
     LOGM_INFO(TAG, "Executing job: %s", job.JobId->c_str());
 
-    auto shutdownHandler = [this]() -> void {
+    auto shutdownHandler = [this]() -> void
+    {
         handlingJob.store(false);
         if (needStop.load())
         {
@@ -648,7 +651,8 @@ void JobsFeature::executeJob(const Iotjobs::JobExecutionData &job, const PlainJo
         }
     };
     // TODO: Add support for checking condition
-    auto runJob = [this, job, jobDocument, shutdownHandler]() {
+    auto runJob = [this, job, jobDocument, shutdownHandler]()
+    {
         auto engine = createJobEngine();
         // execute all action steps in sequence as provided in job document
         int executionStatus = engine->exec_steps(jobDocument, jobHandlerDir);
diff --git a/source/jobs/JobsFeature.h b/source/jobs/JobsFeature.h
@@ -339,8 +339,8 @@ namespace Aws
                     virtual std::shared_ptr<JobEngine> createJobEngine();
                 };
             } // namespace Jobs
-        }     // namespace DeviceClient
-    }         // namespace Iot
+        } // namespace DeviceClient
+    } // namespace Iot
 } // namespace Aws
 
 #endif // DEVICE_CLIENT_JOBSFEATURE_H
diff --git a/source/tunneling/SecureTunnelingContext.cpp b/source/tunneling/SecureTunnelingContext.cpp
@@ -186,8 +186,41 @@ namespace Aws
 
                 void SecureTunnelingContext::OnTcpForwardDataReceive(const Crt::ByteBuf &data) const
                 {
+
+                    if (data.len == 0)
+                    {
+                        LOG_WARN(TAG, "Received empty data buffer in OnTcpForwardDataReceive");
+                        return;
+                    }
+
                     LOGM_DEBUG(TAG, "SecureTunnelingContext::OnTcpForwardDataReceive data.len=%zu", data.len);
-                    mSecureTunnel->SendData(aws_byte_cursor_from_buf(&data));
+                    const size_t MAX_CHUNK_SIZE = 32768; 
+                    size_t offset = 0;
+                    size_t total_sent = 0;
+                
+                    while (offset < data.len)
+                    {
+                        size_t chunk_size = std::min(MAX_CHUNK_SIZE, data.len - offset);
+                        Aws::Crt::ByteCursor chunk = aws_byte_cursor_from_array(data.buffer + offset, chunk_size);
+                        int result = mSecureTunnel->SendData(chunk);
+                        if (result != AWS_OP_SUCCESS)
+                        {
+                            LOGM_ERROR(TAG, "Failed to send data block to secure tunnel. Block size: %zu, Total bytes sent: %zu, Total size: %zu", 
+                                       chunk_size, total_sent, data.len);
+                            break;
+                        }
+                        offset += chunk_size;
+                        total_sent += chunk_size;
+                    }
+                
+                    if (total_sent == data.len)
+                    {
+                        LOGM_INFO(TAG, "Successfully sent data block. Total bytes sent: %zu", total_sent);
+                    }
+                    else
+                    {
+                        LOG_WARN(TAG, "Incomplete data block sent due to network issue");
+                    }
                 }
 
                 void SecureTunnelingContext::StopSecureTunnel()
