Diff: iothub_client.c

Revision:

80:db5f5237bc95

Parent:

78:74a8d3068204

Child:

82:f94e6bed4495

--- a/iothub_client.c	Fri Nov 17 13:57:39 2017 -0800
+++ b/iothub_client.c	Fri Dec 15 14:09:20 2017 -0800
@@ -62,6 +62,15 @@
     LOCK_HANDLE lockGarbage;
     int canBeGarbageCollected; /*flag indicating that the UPLOADTOBLOB_SAVED_DATA structure can be freed because the thread deadling with it finished*/
 }UPLOADTOBLOB_SAVED_DATA;
+
+typedef struct UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA_TAG
+{
+    char* destinationFileName;
+    IOTHUB_CLIENT_FILE_UPLOAD_GET_DATA_CALLBACK getDataCallback;
+    void* context;
+    THREAD_HANDLE uploadingThreadHandle;
+    IOTHUB_CLIENT_HANDLE iotHubClientHandle;
+}UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA;
 #endif
 
 #define USER_CALLBACK_TYPE_VALUES       \
@@ -405,6 +414,44 @@
 {
     size_t callbacks_length = VECTOR_size(call_backs);
     size_t index;
+
+    IOTHUB_CLIENT_DEVICE_TWIN_CALLBACK desired_state_callback = NULL;
+    IOTHUB_CLIENT_EVENT_CONFIRMATION_CALLBACK event_confirm_callback = NULL;
+    IOTHUB_CLIENT_REPORTED_STATE_CALLBACK reported_state_callback = NULL;
+    IOTHUB_CLIENT_CONNECTION_STATUS_CALLBACK connection_status_callback = NULL;
+    IOTHUB_CLIENT_DEVICE_METHOD_CALLBACK_ASYNC device_method_callback = NULL;
+    IOTHUB_CLIENT_INBOUND_DEVICE_METHOD_CALLBACK inbound_device_method_callback = NULL;
+    IOTHUB_CLIENT_MESSAGE_CALLBACK_ASYNC message_callback = NULL;
+    IOTHUB_CLIENT_HANDLE message_user_context_handle = NULL;
+    IOTHUB_CLIENT_HANDLE method_user_context_handle = NULL;
+
+    // Make a local copy of these callbacks, as we don't run with a lock held and iotHubClientInstance may change mid-run.
+    if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
+    {
+        LogError("failed locking for dispatch_user_callbacks");
+    }
+    else
+    {
+        desired_state_callback = iotHubClientInstance->desired_state_callback;
+        event_confirm_callback = iotHubClientInstance->event_confirm_callback;
+        reported_state_callback = iotHubClientInstance->reported_state_callback;
+        connection_status_callback = iotHubClientInstance->connection_status_callback;
+        device_method_callback = iotHubClientInstance->device_method_callback;
+        inbound_device_method_callback = iotHubClientInstance->inbound_device_method_callback;
+        message_callback = iotHubClientInstance->message_callback;
+        if (iotHubClientInstance->method_user_context)
+        {
+            method_user_context_handle = iotHubClientInstance->method_user_context->iotHubClientHandle;
+        }
+        if (iotHubClientInstance->message_user_context)
+        {
+            message_user_context_handle = iotHubClientInstance->message_user_context->iotHubClientHandle;
+        }
+        
+        (void)Unlock(iotHubClientInstance->LockHandle);
+    }
+
+    
     for (index = 0; index < callbacks_length; index++)
     {
         USER_CALLBACK_INFO* queued_cb = (USER_CALLBACK_INFO*)VECTOR_element(call_backs, index);
@@ -418,19 +465,6 @@
             {
                 case CALLBACK_TYPE_DEVICE_TWIN:
                 {
-                    IOTHUB_CLIENT_DEVICE_TWIN_CALLBACK desired_state_callback;
-
-                    if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
-                    {
-                        LogError("failed locking for dispatch_user_callbacks");
-                        desired_state_callback = NULL;
-                    }
-                    else
-                    {
-                        desired_state_callback = iotHubClientInstance->desired_state_callback;
-                        (void)Unlock(iotHubClientInstance->LockHandle);
-                    }
-
                     if (desired_state_callback)
                     {
                         desired_state_callback(queued_cb->iothub_callback.dev_twin_cb_info.update_state, queued_cb->iothub_callback.dev_twin_cb_info.payLoad, queued_cb->iothub_callback.dev_twin_cb_info.size, queued_cb->userContextCallback);
@@ -443,25 +477,25 @@
                     break;
                 }
                 case CALLBACK_TYPE_EVENT_CONFIRM:
-                    if (iotHubClientInstance->event_confirm_callback)
+                    if (event_confirm_callback)
                     {
-                        iotHubClientInstance->event_confirm_callback(queued_cb->iothub_callback.event_confirm_cb_info.confirm_result, queued_cb->userContextCallback);
+                        event_confirm_callback(queued_cb->iothub_callback.event_confirm_cb_info.confirm_result, queued_cb->userContextCallback);
                     }
                     break;
                 case CALLBACK_TYPE_REPORTED_STATE:
-                    if (iotHubClientInstance->reported_state_callback)
+                    if (reported_state_callback)
                     {
-                        iotHubClientInstance->reported_state_callback(queued_cb->iothub_callback.reported_state_cb_info.status_code, queued_cb->userContextCallback);
+                        reported_state_callback(queued_cb->iothub_callback.reported_state_cb_info.status_code, queued_cb->userContextCallback);
                     }
                     break;
                 case CALLBACK_TYPE_CONNECTION_STATUS:
-                    if (iotHubClientInstance->connection_status_callback)
+                    if (connection_status_callback)
                     {
-                        iotHubClientInstance->connection_status_callback(queued_cb->iothub_callback.connection_status_cb_info.connection_status, queued_cb->iothub_callback.connection_status_cb_info.status_reason, queued_cb->userContextCallback);
+                        connection_status_callback(queued_cb->iothub_callback.connection_status_cb_info.connection_status, queued_cb->iothub_callback.connection_status_cb_info.status_reason, queued_cb->userContextCallback);
                     }
                     break;
                 case CALLBACK_TYPE_DEVICE_METHOD:
-                    if (iotHubClientInstance->device_method_callback)
+                    if (device_method_callback)
                     {
                         const char* method_name = STRING_c_str(queued_cb->iothub_callback.method_cb_info.method_name);
                         const unsigned char* payload = BUFFER_u_char(queued_cb->iothub_callback.method_cb_info.payload);
@@ -469,12 +503,11 @@
 
                         unsigned char* payload_resp = NULL;
                         size_t response_size = 0;
-                        int status = iotHubClientInstance->device_method_callback(method_name, payload, payload_len, &payload_resp, &response_size, queued_cb->userContextCallback);
+                        int status = device_method_callback(method_name, payload, payload_len, &payload_resp, &response_size, queued_cb->userContextCallback);
 
                         if (payload_resp && (response_size > 0))
                         {
-                            IOTHUB_CLIENT_HANDLE handle = iotHubClientInstance->method_user_context->iotHubClientHandle;
-                            IOTHUB_CLIENT_RESULT result = IoTHubClient_DeviceMethodResponse(handle, queued_cb->iothub_callback.method_cb_info.method_id, (const unsigned char*)payload_resp, response_size, status);
+                            IOTHUB_CLIENT_RESULT result = IoTHubClient_DeviceMethodResponse(method_user_context_handle, queued_cb->iothub_callback.method_cb_info.method_id, (const unsigned char*)payload_resp, response_size, status);
                             if (result != IOTHUB_CLIENT_OK)
                             {
                                 LogError("IoTHubClient_LL_DeviceMethodResponse failed");
@@ -491,28 +524,27 @@
                     }
                     break;
                 case CALLBACK_TYPE_INBOUD_DEVICE_METHOD:
-                    if (iotHubClientInstance->inbound_device_method_callback)
+                    if (inbound_device_method_callback)
                     {
                         const char* method_name = STRING_c_str(queued_cb->iothub_callback.method_cb_info.method_name);
                         const unsigned char* payload = BUFFER_u_char(queued_cb->iothub_callback.method_cb_info.payload);
                         size_t payload_len = BUFFER_length(queued_cb->iothub_callback.method_cb_info.payload);
 
-                        iotHubClientInstance->inbound_device_method_callback(method_name, payload, payload_len, queued_cb->iothub_callback.method_cb_info.method_id, queued_cb->userContextCallback);
+                        inbound_device_method_callback(method_name, payload, payload_len, queued_cb->iothub_callback.method_cb_info.method_id, queued_cb->userContextCallback);
 
                         BUFFER_delete(queued_cb->iothub_callback.method_cb_info.payload);
                         STRING_delete(queued_cb->iothub_callback.method_cb_info.method_name);
                     }
                     break;
                 case CALLBACK_TYPE_MESSAGE:
-                    if (iotHubClientInstance->message_callback)
+                    if (message_callback)
                     {
-                        IOTHUBMESSAGE_DISPOSITION_RESULT disposition = iotHubClientInstance->message_callback(queued_cb->iothub_callback.message_cb_info->messageHandle, queued_cb->userContextCallback);
-                        IOTHUB_CLIENT_HANDLE handle = iotHubClientInstance->message_user_context->iotHubClientHandle;
+                        IOTHUBMESSAGE_DISPOSITION_RESULT disposition = message_callback(queued_cb->iothub_callback.message_cb_info->messageHandle, queued_cb->userContextCallback);
 
-                        if (Lock(handle->LockHandle) == LOCK_OK)
+                        if (Lock(message_user_context_handle->LockHandle) == LOCK_OK)
                         {
-                            IOTHUB_CLIENT_RESULT result = IoTHubClient_LL_SendMessageDisposition(handle->IoTHubClientLLHandle, queued_cb->iothub_callback.message_cb_info, disposition);
-                            (void)Unlock(handle->LockHandle);
+                            IOTHUB_CLIENT_RESULT result = IoTHubClient_LL_SendMessageDisposition(message_user_context_handle->IoTHubClientLLHandle, queued_cb->iothub_callback.message_cb_info, disposition);
+                            (void)Unlock(message_user_context_handle->LockHandle);
                             if (result != IOTHUB_CLIENT_OK)
                             {
                                 LogError("IoTHubClient_LL_SendMessageDisposition failed");
@@ -1065,11 +1097,6 @@
         }
         else
         {
-            if (iotHubClientInstance->created_with_transport_handle == 0)
-            {
-                iotHubClientInstance->event_confirm_callback = eventConfirmationCallback;
-            }
-
             /* Codes_SRS_IOTHUBCLIENT_01_025: [IoTHubClient_SendEventAsync shall be made thread-safe by using the lock created in IoTHubClient_Create.] */
             if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
             {
@@ -1079,6 +1106,11 @@
             }
             else
             {
+                if (iotHubClientInstance->created_with_transport_handle == 0)
+                {
+                    iotHubClientInstance->event_confirm_callback = eventConfirmationCallback;
+                }
+
                 if (iotHubClientInstance->created_with_transport_handle != 0 || eventConfirmationCallback == NULL)
                 {
                     result = IoTHubClient_LL_SendEventAsync(iotHubClientInstance->IoTHubClientLLHandle, eventMessageHandle, eventConfirmationCallback, userContextCallback);
@@ -1174,11 +1206,6 @@
         }
         else
         {
-            if (iotHubClientInstance->created_with_transport_handle == 0)
-            {
-                iotHubClientInstance->message_callback = messageCallback;
-            }
-
             /* Codes_SRS_IOTHUBCLIENT_01_027: [IoTHubClient_SetMessageCallback shall be made thread-safe by using the lock created in IoTHubClient_Create.] */
             if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
             {
@@ -1188,6 +1215,10 @@
             }
             else
             {
+                if (iotHubClientInstance->created_with_transport_handle == 0)
+                {
+                    iotHubClientInstance->message_callback = messageCallback;
+                }
                 if (iotHubClientInstance->message_user_context != NULL)
                 {
                     free(iotHubClientInstance->message_user_context);
@@ -1573,11 +1604,6 @@
         }
         else
         {
-            if (iotHubClientInstance->created_with_transport_handle == 0)
-            {
-                iotHubClientInstance->reported_state_callback = reportedStateCallback;
-            }
-
             /*Codes_SRS_IOTHUBCLIENT_10_021: [** `IoTHubClient_SendReportedState` shall be made thread-safe by using the lock created in IoTHubClient_Create. ]*/
             if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
             {
@@ -1587,6 +1613,11 @@
             }
             else
             {
+                if (iotHubClientInstance->created_with_transport_handle == 0)
+                {
+                    iotHubClientInstance->reported_state_callback = reportedStateCallback;
+                }
+
                 if (iotHubClientInstance->created_with_transport_handle != 0 || reportedStateCallback == NULL)
                 {
                     /*Codes_SRS_IOTHUBCLIENT_10_017: [** `IoTHubClient_SendReportedState` shall call `IoTHubClient_LL_SendReportedState`, while passing the `IoTHubClient_LL handle` created by `IoTHubClient_LL_Create` along with the parameters `reportedState`, `size`, `reportedStateCallback`, and `userContextCallback`. ]*/
@@ -1647,11 +1678,6 @@
         }
         else
         {
-            if (iotHubClientInstance->created_with_transport_handle == 0)
-            {
-                iotHubClientInstance->device_method_callback = deviceMethodCallback;
-            }
-
             /*Codes_SRS_IOTHUBCLIENT_12_018: [ IoTHubClient_SetDeviceMethodCallback shall be made thread-safe by using the lock created in IoTHubClient_Create. ]*/
             if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
             {
@@ -1661,6 +1687,11 @@
             }
             else
             {
+                if (iotHubClientInstance->created_with_transport_handle == 0)
+                {
+                    iotHubClientInstance->device_method_callback = deviceMethodCallback;
+                }
+
                 if (iotHubClientInstance->method_user_context)
                 {
                     free(iotHubClientInstance->method_user_context);
@@ -1730,11 +1761,6 @@
         }
         else
         {
-            if (iotHubClientInstance->created_with_transport_handle == 0)
-            {
-                iotHubClientInstance->inbound_device_method_callback = inboundDeviceMethodCallback;
-            }
-
             /*Codes_SRS_IOTHUBCLIENT_07_007: [ IoTHubClient_SetDeviceMethodCallback_Ex shall be made thread-safe by using the lock created in IoTHubClient_Create. ]*/
             if (Lock(iotHubClientInstance->LockHandle) != LOCK_OK)
             {
@@ -1744,6 +1770,11 @@
             }
             else
             {
+                if (iotHubClientInstance->created_with_transport_handle == 0)
+                {
+                    iotHubClientInstance->inbound_device_method_callback = inboundDeviceMethodCallback;
+                }
+
                 if (iotHubClientInstance->method_user_context)
                 {
                     free(iotHubClientInstance->method_user_context);
@@ -2031,4 +2062,86 @@
     }
     return result;
 }
+
+static int uploadMultipleBlock_thread(void* data)
+{
+    UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA *blocksData = (UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA *)data;
+    IOTHUB_CLIENT_LL_HANDLE llHandle = blocksData->iotHubClientHandle->IoTHubClientLLHandle;
+
+    /*Codes_SRS_IOTHUBCLIENT_99_078: [ The thread shall call `IoTHubClient_LL_UploadMultipleBlocksToBlob` passing the information packed in the structure. ]*/
+    IOTHUB_CLIENT_RESULT result = IoTHubClient_LL_UploadMultipleBlocksToBlob(llHandle, blocksData->destinationFileName, blocksData->getDataCallback, blocksData->context);
+
+    // Clean resources
+    free(blocksData->destinationFileName);
+    free(blocksData);
+    ThreadAPI_Exit(result);
+    return 0;
+}
+
+IOTHUB_CLIENT_RESULT IoTHubClient_UploadMultipleBlocksToBlobAsync(IOTHUB_CLIENT_HANDLE iotHubClientHandle, const char* destinationFileName, IOTHUB_CLIENT_FILE_UPLOAD_GET_DATA_CALLBACK getDataCallback, void* context)
+{
+    IOTHUB_CLIENT_RESULT result;
+
+    /*Codes_SRS_IOTHUBCLIENT_99_072: [ If `iotHubClientHandle` is `NULL` then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_INVALID_ARG`. ]*/
+    /*Codes_SRS_IOTHUBCLIENT_99_073: [ If `destinationFileName` is `NULL` then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_INVALID_ARG`. ]*/
+    /*Codes_SRS_IOTHUBCLIENT_99_074: [ If `getDataCallback` is `NULL` then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_INVALID_ARG`. ]*/
+    if (
+        (iotHubClientHandle == NULL) ||
+        (destinationFileName == NULL) ||
+        (getDataCallback == NULL)
+        )
+    {
+        LogError("invalid parameters iotHubClientHandle = %p , destinationFileName = %p, getDataCallback = %p",
+            iotHubClientHandle,
+            destinationFileName,
+            getDataCallback
+        );
+        result = IOTHUB_CLIENT_INVALID_ARG;
+    }
+    else
+    {
+        /*Codes_SRS_IOTHUBCLIENT_99_075: [ `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall copy the `destinationFileName`, `getDataCallback`, `context`  and `iotHubClientHandle` into a structure. ]*/
+        UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA *blocksData = (UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA *)malloc(sizeof(UPLOADTOBLOB_MULTIBLOCK_SAVED_DATA));
+        if (blocksData == NULL)
+        {
+            /*Codes_SRS_IOTHUBCLIENT_99_077: [ If copying to the structure or spawning the thread fails, then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_ERROR`. ]*/
+            LogError("unable to malloc");
+            result = IOTHUB_CLIENT_ERROR;
+        }
+        else
+        {
+            /*Codes_SRS_IOTHUBCLIENT_99_075: [ `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall copy the `destinationFileName`, `getDataCallback`, `context`  and `iotHubClientHandle` into a structure. ]*/
+            if (mallocAndStrcpy_s((char**)&blocksData->destinationFileName, destinationFileName) != 0)
+            {
+                /*Codes_SRS_IOTHUBCLIENT_99_077: [ If copying to the structure or spawning the thread fails, then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_ERROR`. ]*/
+                LogError("unable to mallocAndStrcpy_s");
+                free(blocksData);
+                result = IOTHUB_CLIENT_ERROR;
+            }
+            else
+            {
+                /*Codes_SRS_IOTHUBCLIENT_99_075: [ `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall copy the `destinationFileName`, `getDataCallback`, `context`  and `iotHubClientHandle` into a structure. ]*/
+                blocksData->getDataCallback = getDataCallback;
+                blocksData->context = context;
+                blocksData->iotHubClientHandle = iotHubClientHandle;
+                if (ThreadAPI_Create(&blocksData->uploadingThreadHandle, uploadMultipleBlock_thread, blocksData) != THREADAPI_OK)
+                {
+                    /*Codes_SRS_IOTHUBCLIENT_99_077: [ If copying to the structure or spawning the thread fails, then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall fail and return `IOTHUB_CLIENT_ERROR`. ]*/
+                    LogError("unable to ThreadAPI_Create");
+                    free(blocksData->destinationFileName);
+                    free(blocksData);
+                    result = IOTHUB_CLIENT_ERROR;
+                }
+                else
+                {
+                    /*Codes_SRS_IOTHUBCLIENT_99_077: [ If copying to the structure and spawning the thread succeeds, then `IoTHubClient_UploadMultipleBlocksToBlobAsync` shall return `IOTHUB_CLIENT_OK`. ]*/
+                    result = IOTHUB_CLIENT_OK;
+                }
+            }
+        }
+    }
+
+    return result;
+}
+
 #endif /*DONT_USE_UPLOADTOBLOB*/