iothubtransport.c

00001 // Copyright (c) Microsoft. All rights reserved.
00002 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
00003 
00004 #include <stdlib.h> 
00005 #ifdef _CRTDBG_MAP_ALLOC
00006 #include <crtdbg.h>
00007 #endif
00008 
00009 #include "azure_c_shared_utility/gballoc.h"
00010 
00011 #include <stdlib.h>
00012 #include <signal.h>
00013 #include <stddef.h>
00014 #include "azure_c_shared_utility/crt_abstractions.h"
00015 #include "iothubtransport.h"
00016 #include "iothub_client.h"
00017 #include "iothub_client_private.h"
00018 #include "azure_c_shared_utility/threadapi.h"
00019 #include "azure_c_shared_utility/lock.h"
00020 #include "azure_c_shared_utility/iot_logging.h"
00021 #include "azure_c_shared_utility/vector.h"
00022 
00023 typedef struct TRANSPORT_HANDLE_DATA_TAG
00024 {
00025     TRANSPORT_LL_HANDLE transportLLHandle;
00026     THREAD_HANDLE workerThreadHandle;
00027     LOCK_HANDLE lockHandle;
00028     sig_atomic_t stopThread;
00029     TRANSPORT_PROVIDER_FIELDS;
00030     VECTOR_HANDLE clients;
00031 } TRANSPORT_HANDLE_DATA;
00032 
00033 /* Used for Unit test */
00034 const size_t IoTHubTransport_ThreadTerminationOffset = offsetof(TRANSPORT_HANDLE_DATA, stopThread);
00035 
00036 TRANSPORT_HANDLE  IoTHubTransport_Create(IOTHUB_CLIENT_TRANSPORT_PROVIDER protocol, const char* iotHubName, const char* iotHubSuffix)
00037 {
00038     TRANSPORT_HANDLE_DATA * result;
00039 
00040     if (protocol == NULL || iotHubName == NULL || iotHubSuffix == NULL)
00041     {
00042         /*Codes_SRS_IOTHUBTRANSPORT_17_002: [ If protocol is NULL, this function shall return NULL. ]*/
00043         /*Codes_SRS_IOTHUBTRANSPORT_17_003: [ If iotHubName is NULL, this function shall return NULL. ]*/
00044         /*Codes_SRS_IOTHUBTRANSPORT_17_004: [ If iotHubSuffix is NULL, this function shall return NULL. ]*/
00045         LogError("Invalid NULL argument, protocol [%p], name [%p], suffix [%p].", protocol, iotHubName, iotHubSuffix);
00046         result = NULL;
00047     }
00048     else
00049     {
00050         /*Codes_SRS_IOTHUBTRANSPORT_17_032: [ IoTHubTransport_Create shall allocate memory for the transport data. ]*/
00051         result = (TRANSPORT_HANDLE_DATA*)malloc(sizeof(TRANSPORT_HANDLE_DATA));
00052         if (result == NULL)
00053         {
00054             /*Codes_SRS_IOTHUBTRANSPORT_17_040: [ If memory allocation fails, IoTHubTransport_Create shall return NULL. ]*/
00055             LogError("Transport handle was not allocated.");
00056         }
00057         else
00058         {
00059             TRANSPORT_PROVIDER * transportProtocol = (TRANSPORT_PROVIDER*)(protocol());
00060             IOTHUB_CLIENT_CONFIG upperConfig;
00061             upperConfig.deviceId = NULL;
00062             upperConfig.deviceKey = NULL;
00063             upperConfig.iotHubName = iotHubName;
00064             upperConfig.iotHubSuffix = iotHubSuffix;
00065             upperConfig.protocol = protocol;
00066             upperConfig.protocolGatewayHostName = NULL;
00067 
00068             IOTHUBTRANSPORT_CONFIG transportLLConfig;
00069             transportLLConfig.upperConfig = &upperConfig;
00070             transportLLConfig.waitingToSend = NULL;
00071 
00072             /*Codes_SRS_IOTHUBTRANSPORT_17_005: [ IoTHubTransport_Create shall create the lower layer transport by calling the protocol's IoTHubTransport_Create function. ]*/
00073             result->transportLLHandle = transportProtocol->IoTHubTransport_Create(&transportLLConfig);
00074             if (result->transportLLHandle == NULL)
00075             {
00076                 /*Codes_SRS_IOTHUBTRANSPORT_17_006: [ If the creation of the transport fails, IoTHubTransport_Create shall return NULL. ]*/
00077                 LogError("Lower Layer transport not created.");
00078                 free(result);
00079                 result = NULL;
00080             }
00081             else
00082             {
00083                 /*Codes_SRS_IOTHUBTRANSPORT_17_007: [ IoTHubTransport_Create shall create the transport lock by Calling Lock_Init. ]*/
00084                 result->lockHandle = Lock_Init();
00085                 if (result->lockHandle == NULL)
00086                 {
00087                     /*Codes_SRS_IOTHUBTRANSPORT_17_008: [ If the lock creation fails, IoTHubTransport_Create shall return NULL. ]*/
00088                     LogError("transport Lock not created.");
00089                     transportProtocol->IoTHubTransport_Destroy(result->transportLLHandle);
00090                     free(result);
00091                     result = NULL;
00092                 }
00093                 else
00094                 {
00095                     /*Codes_SRS_IOTHUBTRANSPORT_17_038: [ IoTHubTransport_Create shall call VECTOR_Create to make a list of IOTHUB_CLIENT_HANDLE using this transport. ]*/
00096                     result->clients = VECTOR_create(sizeof(IOTHUB_CLIENT_HANDLE));
00097                     if (result->clients == NULL)
00098                     {
00099                         /*Codes_SRS_IOTHUBTRANSPORT_17_039: [ If the Vector creation fails, IoTHubTransport_Create shall return NULL. ]*/
00100                         /*Codes_SRS_IOTHUBTRANSPORT_17_009: [ IoTHubTransport_Create shall clean up any resources it creates if the function does not succeed. ]*/
00101                         LogError("clients list not created.");
00102                         Lock_Deinit(result->lockHandle);
00103                         transportProtocol->IoTHubTransport_Destroy(result->transportLLHandle);
00104                         free(result);
00105                         result = NULL;
00106                     }
00107                     else
00108                     {
00109                         /*Codes_SRS_IOTHUBTRANSPORT_17_001: [ IoTHubTransport_Create shall return a non-NULL handle on success.]*/
00110                         result->stopThread = 1;
00111                         result->workerThreadHandle = NULL; /* create thread when work needs to be done */
00112                         result->IoTHubTransport_GetHostname = transportProtocol->IoTHubTransport_GetHostname;
00113                         result->IoTHubTransport_SetOption = transportProtocol->IoTHubTransport_SetOption;
00114                         result->IoTHubTransport_Create = transportProtocol->IoTHubTransport_Create;
00115                         result->IoTHubTransport_Destroy = transportProtocol->IoTHubTransport_Destroy;
00116                         result->IoTHubTransport_Register = transportProtocol->IoTHubTransport_Register;
00117                         result->IoTHubTransport_Unregister = transportProtocol->IoTHubTransport_Unregister;
00118                         result->IoTHubTransport_Subscribe = transportProtocol->IoTHubTransport_Subscribe;
00119                         result->IoTHubTransport_Unsubscribe = transportProtocol->IoTHubTransport_Unsubscribe;
00120                         result->IoTHubTransport_DoWork = transportProtocol->IoTHubTransport_DoWork;
00121                         result->IoTHubTransport_GetSendStatus = transportProtocol->IoTHubTransport_GetSendStatus;
00122                     }
00123                 }
00124             }
00125         }
00126     }
00127     
00128     return result;
00129 }
00130 
00131 static int transport_worker_thread(void* threadArgument)
00132 {
00133     TRANSPORT_HANDLE_DATA* transportData = (TRANSPORT_HANDLE_DATA*)threadArgument;
00134 
00135     while (1)
00136     {
00137         /*Codes_SRS_IOTHUBTRANSPORT_17_030: [ All calls to lower layer transport DoWork shall be protected by the lock created in IoTHubTransport_Create. ]*/
00138         if (Lock(transportData->lockHandle) == LOCK_OK)
00139         {
00140             /*Codes_SRS_IOTHUBTRANSPORT_17_031: [ If acquiring the lock fails, lower layer transport DoWork shall not be called. ]*/
00141             if (transportData->stopThread)
00142             {
00143                 /*Codes_SRS_IOTHUBTRANSPORT_17_028: [ The thread shall exit when IoTHubTransport_EndWorkerThread has been called for each clientHandle which invoked IoTHubTransport_StartWorkerThread. ]*/
00144                 (void)Unlock(transportData->lockHandle);
00145                 break;
00146             }
00147             else
00148             {
00149                 (transportData->IoTHubTransport_DoWork)(transportData->transportLLHandle, NULL);
00150                 (void)Unlock(transportData->lockHandle);
00151             }
00152         }
00153         /*Codes_SRS_IOTHUBTRANSPORT_17_029: [ The thread shall call lower layer transport DoWork every 1 ms. ]*/
00154         ThreadAPI_Sleep(1);
00155     }
00156 
00157     return 0;
00158 }
00159 
00160 static bool find_by_handle(const void* element, const void* value)
00161 {
00162     /* data stored at element is device handle */
00163     const IOTHUB_CLIENT_HANDLE * guess = (const IOTHUB_CLIENT_HANDLE *)element;
00164     const IOTHUB_CLIENT_HANDLE match = (const IOTHUB_CLIENT_HANDLE)value;
00165     return (*guess == match);
00166 }
00167 
00168 static IOTHUB_CLIENT_RESULT start_worker_if_needed(TRANSPORT_HANDLE_DATA * transportData, IOTHUB_CLIENT_HANDLE clientHandle)
00169 {
00170     IOTHUB_CLIENT_RESULT result;
00171     if (transportData->workerThreadHandle == NULL)
00172     {
00173         /*Codes_SRS_IOTHUBTRANSPORT_17_018: [ If the worker thread does not exist, IoTHubTransport_StartWorkerThread shall start the thread using ThreadAPI_Create. ]*/
00174         transportData->stopThread = 0;
00175         if (ThreadAPI_Create(&transportData->workerThreadHandle, transport_worker_thread, transportData) != THREADAPI_OK)
00176         {
00177             transportData->workerThreadHandle = NULL;
00178         }
00179     }
00180     if (transportData->workerThreadHandle != NULL)
00181     {
00182         /*Codes_SRS_IOTHUBTRANSPORT_17_020: [ IoTHubTransport_StartWorkerThread shall search for IoTHubClient clientHandle in the list of IoTHubClient handles. ]*/
00183         bool addToList = ((VECTOR_size(transportData->clients) == 0) || (VECTOR_find_if(transportData->clients, find_by_handle, clientHandle) == NULL));
00184         if (addToList)
00185         {
00186             /*Codes_SRS_IOTHUBTRANSPORT_17_021: [ If handle is not found, then clientHandle shall be added to the list. ]*/
00187             if (VECTOR_push_back(transportData->clients, &clientHandle, 1) != 0)
00188             {
00189                 /*Codes_SRS_IOTHUBTRANSPORT_17_042: [ If Adding to the client list fails, IoTHubTransport_StartWorkerThread shall return IOTHUB_CLIENT_ERROR. ]*/
00190                 result = IOTHUB_CLIENT_ERROR;
00191             }
00192             else
00193             {
00194                 result = IOTHUB_CLIENT_OK;
00195             }
00196         }
00197         else
00198         {
00199             result = IOTHUB_CLIENT_OK;
00200         }
00201     }
00202     else
00203     {
00204         result = IOTHUB_CLIENT_ERROR;
00205     }
00206     return result;
00207 }
00208 
00209 static void stop_worker_thread(TRANSPORT_HANDLE_DATA * transportData)
00210 {
00211     /*Codes_SRS_IOTHUBTRANSPORT_17_043: [** IoTHubTransport_SignalEndWorkerThread shall signal the worker thread to end.*/
00212     transportData->stopThread = 1;
00213 }
00214 
00215 static void wait_worker_thread(TRANSPORT_HANDLE_DATA * transportData)
00216 {
00217     if (transportData->workerThreadHandle != NULL)
00218     {
00219         int res;
00220         /*Codes_SRS_IOTHUBTRANSPORT_17_027: [ If handle list is empty, IoTHubTransport_EndWorkerThread shall be joined. ]*/
00221         if (ThreadAPI_Join(transportData->workerThreadHandle, &res) != THREADAPI_OK)
00222         {
00223             LogError("ThreadAPI_Join failed");
00224         }
00225         else
00226         {
00227             transportData->workerThreadHandle = NULL;
00228         }
00229     }
00230 }
00231 
00232 static bool signal_end_worker_thread(TRANSPORT_HANDLE_DATA * transportData, IOTHUB_CLIENT_HANDLE clientHandle)
00233 {
00234     bool okToJoin;
00235     void* element = VECTOR_find_if(transportData->clients, find_by_handle, clientHandle);
00236     if (element != NULL)
00237     {
00238         /*Codes_SRS_IOTHUBTRANSPORT_17_026: [ IoTHubTransport_EndWorkerThread shall remove clientHandlehandle from handle list. ]*/
00239         VECTOR_erase(transportData->clients, element, 1);
00240     }
00241     /*Codes_SRS_IOTHUBTRANSPORT_17_025: [ If the worker thread does not exist, then IoTHubTransport_EndWorkerThread shall return. ]*/
00242     if (transportData->workerThreadHandle != NULL)
00243     {
00244         if (VECTOR_size(transportData->clients) == 0)
00245         {
00246             stop_worker_thread(transportData);
00247             okToJoin = true;
00248         }
00249         else
00250         {
00251             okToJoin = false;
00252         }
00253     }
00254     else
00255     {
00256         okToJoin = false;
00257     }
00258     return okToJoin;
00259 }
00260 
00261 void IoTHubTransport_Destroy(TRANSPORT_HANDLE transportHandle)
00262 {
00263     /*Codes_SRS_IOTHUBTRANSPORT_17_011: [ IoTHubTransport_Destroy shall do nothing if transportHandle is NULL. ]*/
00264     if (transportHandle != NULL)
00265     {
00266         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00267         /*Codes_SRS_IOTHUBTRANSPORT_17_033: [ IoTHubTransport_Destroy shall lock the transport lock. ]*/
00268         if (Lock(transportData->lockHandle) != LOCK_OK)
00269         {
00270             LogError("Unable to lock - will still attempt to end thread without thread safety");
00271             stop_worker_thread(transportData);
00272         }
00273         else
00274         {
00275             stop_worker_thread(transportData);
00276             (void)Unlock(transportData->lockHandle);
00277         }
00278         wait_worker_thread(transportData);
00279         /*Codes_SRS_IOTHUBTRANSPORT_17_010: [ IoTHubTransport_Destroy shall free all resources. ]*/
00280         Lock_Deinit(transportData->lockHandle);
00281         (transportData->IoTHubTransport_Destroy)(transportData->transportLLHandle);
00282         VECTOR_destroy(transportData->clients);
00283         free(transportHandle);
00284     }
00285 }
00286 
00287 LOCK_HANDLE IoTHubTransport_GetLock(TRANSPORT_HANDLE transportHandle)
00288 {
00289     LOCK_HANDLE lock;
00290     if (transportHandle == NULL)
00291     {
00292         /*Codes_SRS_IOTHUBTRANSPORT_17_013: [ If transportHandle is NULL, IoTHubTransport_GetLock shall return NULL. ]*/
00293         lock = NULL;
00294     }
00295     else
00296     {
00297         /*Codes_SRS_IOTHUBTRANSPORT_17_012: [ IoTHubTransport_GetLock shall return a handle to the transport lock. ]*/
00298         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00299         lock = transportData->lockHandle;
00300     }
00301     return lock;
00302 }
00303 
00304 TRANSPORT_LL_HANDLE IoTHubTransport_GetLLTransport(TRANSPORT_HANDLE transportHandle)
00305 {
00306     TRANSPORT_LL_HANDLE llTransport;
00307     if (transportHandle == NULL)
00308     {
00309         /*Codes_SRS_IOTHUBTRANSPORT_17_015: [ If transportHandle is NULL, IoTHubTransport_GetLLTransport shall return NULL. ]*/
00310         llTransport = NULL;
00311     }
00312     else
00313     {
00314         /*Codes_SRS_IOTHUBTRANSPORT_17_014: [ IoTHubTransport_GetLLTransport shall return a handle to the lower layer transport. ]*/
00315         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00316         llTransport = transportData->transportLLHandle;
00317     }
00318     return llTransport;
00319 }
00320 
00321 IOTHUB_CLIENT_RESULT IoTHubTransport_StartWorkerThread(TRANSPORT_HANDLE transportHandle, IOTHUB_CLIENT_HANDLE clientHandle)
00322 {
00323     IOTHUB_CLIENT_RESULT result;
00324     if (transportHandle == NULL || clientHandle == NULL)
00325     {
00326         /*Codes_SRS_IOTHUBTRANSPORT_17_016: [ If transportHandle is NULL, IoTHubTransport_StartWorkerThread shall return IOTHUB_CLIENT_INVALID_ARG. ]*/
00327         /*Codes_SRS_IOTHUBTRANSPORT_17_017: [ If clientHandle is NULL, IoTHubTransport_StartWorkerThread shall return IOTHUB_CLIENT_INVALID_ARG. ]*/
00328         result = IOTHUB_CLIENT_INVALID_ARG;
00329     }
00330     else
00331     {
00332         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00333 
00334         if ((result = start_worker_if_needed(transportData, clientHandle)) != IOTHUB_CLIENT_OK)
00335         {
00336             /*Codes_SRS_IOTHUBTRANSPORT_17_019: [ If thread creation fails, IoTHubTransport_StartWorkerThread shall return IOTHUB_CLIENT_ERROR. */
00337             LogError("Unable to start thread safely");
00338         }
00339         else
00340         {
00341             /*Codes_SRS_IOTHUBTRANSPORT_17_022: [ Upon success, IoTHubTransport_StartWorkerThread shall return IOTHUB_CLIENT_OK. ]*/
00342             result = IOTHUB_CLIENT_OK;
00343         }
00344     }
00345     return result;
00346 }
00347 
00348 bool IoTHubTransport_SignalEndWorkerThread(TRANSPORT_HANDLE transportHandle, IOTHUB_CLIENT_HANDLE clientHandle)
00349 {
00350     bool okToJoin;
00351     /*Codes_SRS_IOTHUBTRANSPORT_17_023: [ If transportHandle is NULL, IoTHubTransport_EndWorkerThread shall return. ]*/
00352     /*Codes_SRS_IOTHUBTRANSPORT_17_024: [ If clientHandle is NULL, IoTHubTransport_EndWorkerThread shall return. ]*/
00353     if (!(transportHandle == NULL || clientHandle == NULL))
00354     {
00355         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00356         okToJoin = signal_end_worker_thread(transportData, clientHandle);
00357     }
00358     else
00359     {
00360         okToJoin = false;
00361     }
00362     return okToJoin;
00363 }
00364 
00365 void IoTHubTransport_JoinWorkerThread(TRANSPORT_HANDLE transportHandle, IOTHUB_CLIENT_HANDLE clientHandle)
00366 {
00367     /*Codes_SRS_IOTHUBTRANSPORT_17_044: [ If transportHandle is NULL, IoTHubTransport_JoinWorkerThread shall do nothing. ]*/
00368     /*Codes_SRS_IOTHUBTRANSPORT_17_045: [ If clientHandle is NULL, IoTHubTransport_JoinWorkerThread shall do nothing. ]*/
00369     if (!(transportHandle == NULL || clientHandle == NULL))
00370     {
00371         TRANSPORT_HANDLE_DATA * transportData = (TRANSPORT_HANDLE_DATA*)transportHandle;
00372         /*Codes_SRS_IOTHUBTRANSPORT_17_027: [ The worker thread shall be joined. ]*/
00373         wait_worker_thread(transportData);
00374     }
00375 }