resource/csdk/connectivity/src/camessagehandler.c

   1 /******************************************************************
   2  *
   3  * Copyright 2014 Samsung Electronics All Rights Reserved.
   4  *
   5  *
   6  *
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  *
  19  ******************************************************************/
  20
  21 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <string.h>
  24 #include <stdint.h>
  25
  26 #include "cainterface.h"
  27 #include "camessagehandler.h"
  28 #include "caremotehandler.h"
  29 #include "cainterfacecontroller.h"
  30 #include "caprotocolmessage.h"
  31 #include "uqueue.h"
  32 #include "logger.h"
  33 #include "config.h" /* for coap protocol */
  34 #include "coap.h"
  35 #include "uthreadpool.h" /* for thread pool */
  36 #include "caqueueingthread.h"
  37 #include "umutex.h"
  38 #include "oic_malloc.h"
  39
  40 #define TAG PCF("CA")
  41
  42 #define MEMORY_ALLOCK_CHECK(arg) { if (arg == NULL) {OIC_LOG_V(DEBUG, TAG, "memory error"); goto memory_error_exit;} }
  43
  44 #define MAX_THREAD_POOL_SIZE    10
  45
  46 typedef enum
  47 {
  48     SEND_TYPE_MULTICAST = 0, SEND_TYPE_UNICAST
  49 } CASendDataType_t;
  50
  51 typedef struct
  52 {
  53     CASendDataType_t type;
  54     CARemoteEndpoint_t* remoteEndpoint;
  55     CARequestInfo_t* requestInfo;
  56     CAResponseInfo_t* responseInfo;
  57     CAHeaderOption_t* options;
  58     uint8_t numOptions;
  59 } CAData_t;
  60
  61 // thread pool handle
  62 static u_thread_pool_t gThreadPoolHandle = NULL;
  63
  64 // message handler main thread
  65 static CAQueueingThread_t gSendThread;
  66 static CAQueueingThread_t gReceiveThread;
  67
  68 // handler field
  69 static CARequestCallback gRequestHandler = NULL;
  70 static CAResponseCallback gResponseHandler = NULL;
  71
  72 static void CAReceiveThreadProcess(void* threadData)
  73 {
  74     // TODO
  75     // currently not support.
  76     // CAHandleRequestResponseCallbacks codes will move to this function.
  77 }
  78
  79 static void CASendThreadProcess(void* threadData)
  80 {
  81     CAData_t* data = (CAData_t*) threadData;
  82
  83     if (data == NULL)
  84     {
  85         OIC_LOG(DEBUG, TAG, "thread data error!!");
  86         return;
  87     }
  88
  89     if (NULL == data->remoteEndpoint)
  90     {
  91         OIC_LOG(DEBUG, TAG, "remoteEndpoint is null");
  92         return;
  93     }
  94
  95     CAResult_t res = CA_STATUS_FAILED;
  96
  97     CASendDataType_t type = data->type;
  98
  99     if (type == SEND_TYPE_UNICAST)
 100     {
 101         coap_pdu_t* pdu = NULL;
 102
 103         if (data->requestInfo != NULL)
 104         {
 105             OIC_LOG_V(DEBUG, TAG, "requestInfo is available..");
 106
 107             pdu = (coap_pdu_t*) CAGeneratePdu(data->remoteEndpoint->resourceUri,
 108                     data->requestInfo->method, data->requestInfo->info);
 109         }
 110         else if (data->responseInfo != NULL)
 111         {
 112             OIC_LOG_V(DEBUG, TAG, "responseInfo is available..");
 113
 114             pdu = (coap_pdu_t*) CAGeneratePdu(data->remoteEndpoint->resourceUri,
 115                     data->responseInfo->result, data->responseInfo->info);
 116         }
 117         else
 118         {
 119             OIC_LOG(DEBUG, TAG, "request info, response info is empty");
 120         }
 121
 122         // interface controller function call.
 123         if (NULL != pdu)
 124         {
 125             OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
 126
 127             OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
 128
 129             OIC_LOG_V(DEBUG, TAG, "PDU Maker - buffer data : %s", pdu->hdr);
 130
 131             res = CASendUnicastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 132         }
 133
 134     }
 135     else if (type == SEND_TYPE_MULTICAST)
 136     {
 137         OIC_LOG(DEBUG, TAG, "both requestInfo & responseInfo is not available");
 138
 139         coap_pdu_t* pdu = NULL;
 140         CAInfo_t info;
 141         memset(&info, 0, sizeof(CAInfo_t));
 142
 143         info.options = data->options;
 144         info.numOptions = data->numOptions;
 145
 146         pdu = (coap_pdu_t*) CAGeneratePdu(data->remoteEndpoint->resourceUri, CA_GET, info);
 147
 148         if (NULL != pdu)
 149         {
 150             OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
 151
 152             OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
 153
 154             OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
 155
 156             OIC_LOG_V(DEBUG, TAG, "PDU Maker - id : %d", pdu->hdr->id);
 157
 158             OIC_LOG_V(DEBUG, TAG, "PDU Maker - buffer data : %s", pdu->hdr);
 159
 160             res = CASendMulticastData(pdu->hdr, pdu->length);
 161         }
 162     }
 163     else
 164     {
 165         OIC_LOG(DEBUG, TAG, "unknown type!");
 166     }
 167
 168 }
 169
 170 static void CAReceivedPacketCallback(CARemoteEndpoint_t* endpoint, void* data, uint32_t dataLen)
 171 {
 172     OIC_LOG(DEBUG, TAG, "receivedPacketCallback in message handler!!");
 173
 174     if (NULL == data)
 175     {
 176         OIC_LOG(DEBUG, TAG, "received data is null");
 177         return;
 178     }
 179
 180     coap_pdu_t* pdu;
 181     uint32_t code = CA_NOT_FOUND;
 182     pdu = (coap_pdu_t*) CAParsePDU((const char*) data, &code);
 183     OICFree(data);
 184
 185     char uri[CA_MAX_URI_LENGTH] =
 186     { 0, };
 187
 188     if (code == CA_GET || code == CA_POST || code == CA_PUT || code == CA_DELETE)
 189     {
 190         CARequestInfo_t* ReqInfo;
 191         ReqInfo = (CARequestInfo_t*) OICMalloc(sizeof(CARequestInfo_t));
 192         memset(ReqInfo, 0, sizeof(CARequestInfo_t));
 193         CAGetRequestInfoFromPdu(pdu, ReqInfo, uri);
 194
 195         if (NULL != ReqInfo->info.options && NULL != endpoint)
 196         {
 197             uint32_t i;
 198             for (i = 0; i < ReqInfo->info.numOptions; i++)
 199             {
 200                 OIC_LOG_V(DEBUG, TAG, "Request- optionID: %d", ReqInfo->info.options[i].optionID);
 201
 202                 OIC_LOG_V(DEBUG, TAG, "Request- list: %s", ReqInfo->info.options[i].optionData);
 203             }
 204
 205             OIC_LOG_V(DEBUG, TAG, "Request- payload: %s", ReqInfo->info.payload);
 206
 207             OIC_LOG_V(DEBUG, TAG, "Request- code: %d", ReqInfo->method);
 208
 209             endpoint->resourceUri = (char*) OICMalloc(strlen(uri) + 1);
 210             memset(endpoint->resourceUri, 0, strlen(uri) + 1);
 211             memcpy(endpoint->resourceUri, uri, strlen(uri));
 212             OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
 213         }
 214
 215         // store the data at queue.
 216         CAData_t* cadata = NULL;
 217         cadata = (CAData_t*) OICMalloc(sizeof(CAData_t));
 218         memset(cadata, 0, sizeof(CAData_t));
 219
 220         cadata->type = SEND_TYPE_UNICAST;
 221         cadata->remoteEndpoint = endpoint;
 222         cadata->requestInfo = ReqInfo;
 223         cadata->responseInfo = NULL;
 224         CAQueueingThreadAddData(&gReceiveThread, cadata, sizeof(CAData_t));
 225     }
 226     else
 227     {
 228         CAResponseInfo_t* ResInfo;
 229         ResInfo = (CAResponseInfo_t*) OICMalloc(sizeof(CAResponseInfo_t));
 230         memset(ResInfo, 0, sizeof(CAResponseInfo_t));
 231         CAGetResponseInfoFromPdu(pdu, ResInfo, uri);
 232
 233         if (NULL != ResInfo->info.options && NULL != endpoint)
 234         {
 235             uint32_t i;
 236             for (i = 0; i < ResInfo->info.numOptions; i++)
 237             {
 238                 OIC_LOG_V(DEBUG, TAG, "Response- optionID: %d", ResInfo->info.options[i].optionID);
 239
 240                 OIC_LOG_V(DEBUG, TAG, "Response- list: %s", ResInfo->info.options[i].optionData);
 241             }
 242
 243             OIC_LOG_V(DEBUG, TAG, "Response- payload: %s", ResInfo->info.payload);
 244
 245             OIC_LOG_V(DEBUG, TAG, "Response- code: %d", ResInfo->result);
 246
 247             endpoint->resourceUri = (char*) OICMalloc(strlen(uri) + 1);
 248             memset(endpoint->resourceUri, 0, strlen(uri) + 1);
 249             memcpy(endpoint->resourceUri, uri, strlen(uri));
 250             OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
 251         }
 252
 253         // store the data at queue.
 254         CAData_t* cadata = NULL;
 255         cadata = (CAData_t*) OICMalloc(sizeof(CAData_t));
 256         memset(cadata, 0, sizeof(CAData_t));
 257
 258         cadata->type = SEND_TYPE_UNICAST;
 259         cadata->remoteEndpoint = endpoint;
 260         cadata->requestInfo = NULL;
 261         cadata->responseInfo = ResInfo;
 262
 263         CAQueueingThreadAddData(&gReceiveThread, cadata, sizeof(CAData_t));
 264     }
 265 }
 266
 267 static void CANetworkChangedCallback(CALocalConnectivity_t* info, CANetworkStatus_t status)
 268 {
 269     OIC_LOG(DEBUG, TAG, "networkChangeCallback in message handler!!");
 270 }
 271
 272 void CAHandleRequestResponseCallbacks()
 273 {
 274     OIC_LOG_V(DEBUG, TAG, "CAHandleRequestResponseCallbacks");
 275
 276     // parse the data and call the callbacks.
 277     // #1 parse the data
 278     // #2 get endpoint
 279
 280     u_mutex_lock(gReceiveThread.threadMutex);
 281
 282     u_queue_message_t* item = u_queue_get_element(gReceiveThread.dataQueue);
 283
 284     u_mutex_unlock(gReceiveThread.threadMutex);
 285
 286     if (item == NULL)
 287         return;
 288
 289     // get values
 290     void* msg = item->msg;
 291
 292     if (msg == NULL)
 293         return;
 294
 295     // get endpoint
 296     CAData_t* td = (CAData_t*) msg;
 297     CARemoteEndpoint_t* rep = td->remoteEndpoint;
 298
 299     if (rep == NULL)
 300         return;
 301
 302     if (td->requestInfo != NULL)
 303     {
 304         if (gRequestHandler)
 305         {
 306             gRequestHandler(rep, td->requestInfo);
 307         }
 308
 309         if(NULL != td->requestInfo->info.options)
 310         {
 311             OICFree(td->requestInfo->info.options);
 312         }
 313
 314         if(NULL != td->requestInfo->info.payload)
 315         {
 316             OICFree(td->requestInfo->info.payload);
 317         }
 318
 319         if(NULL != td->requestInfo->info.token)
 320         {
 321             OICFree(td->requestInfo->info.token);
 322         }
 323         OICFree(td->requestInfo);
 324     }
 325
 326     if (td->responseInfo != NULL)
 327     {
 328         if (gResponseHandler)
 329         {
 330             gResponseHandler(rep, td->responseInfo);
 331         }
 332
 333         if(NULL != td->responseInfo->info.options)
 334         {
 335             OICFree(td->responseInfo->info.options);
 336         }
 337
 338         if(NULL != td->responseInfo->info.payload)
 339         {
 340             OICFree(td->responseInfo->info.payload);
 341         }
 342
 343         if(NULL != td->responseInfo->info.token)
 344         {
 345             OICFree(td->responseInfo->info.token);
 346         }
 347         OICFree(td->responseInfo);
 348     }
 349
 350     if(NULL != rep->resourceUri)
 351     {
 352         OICFree(rep->resourceUri);
 353     }
 354     OICFree(rep);
 355 }
 356
 357 CAResult_t CADetachRequestMessage(const CARemoteEndpoint_t* object, const CARequestInfo_t* request)
 358 {
 359     OIC_LOG_V(DEBUG, TAG, "CADetachRequestMessage");
 360
 361     if (object == NULL || request == NULL)
 362     {
 363         return CA_STATUS_FAILED;
 364     }
 365
 366     CAData_t* data = (CAData_t*) OICMalloc(sizeof(CAData_t));
 367     MEMORY_ALLOCK_CHECK(data);
 368
 369     // initialize
 370     memset(data, 0, sizeof(CAData_t));
 371
 372     // clone remote endpoint
 373     CARemoteEndpoint_t* remoteEndpoint = CACloneRemoteEndpoint(object);
 374     MEMORY_ALLOCK_CHECK(remoteEndpoint);
 375
 376     // clone request info
 377     CARequestInfo_t* requestInfo = CACloneRequestInfo(request);
 378     MEMORY_ALLOCK_CHECK(requestInfo);
 379
 380     // save data
 381     data->type = SEND_TYPE_UNICAST;
 382     data->remoteEndpoint = remoteEndpoint;
 383     data->requestInfo = requestInfo;
 384     data->responseInfo = NULL;
 385
 386     // add thread
 387     CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
 388
 389     return CA_STATUS_OK;
 390
 391     // memory error label.
 392     memory_error_exit:
 393
 394     CADestroyRemoteEndpointInternal(remoteEndpoint);
 395
 396     CADestroyRequestInfoInternal(requestInfo);
 397
 398     if (data != NULL)
 399     {
 400         OICFree(data);
 401     }
 402
 403     return CA_MEMORY_ALLOC_FAILED;
 404 }
 405
 406 CAResult_t CADetachResponseMessage(const CARemoteEndpoint_t* object,
 407         const CAResponseInfo_t* response)
 408 {
 409     OIC_LOG_V(DEBUG, TAG, "CADetachResponseMessage");
 410
 411     if (object == NULL || response == NULL)
 412     {
 413         return CA_STATUS_FAILED;
 414     }
 415
 416     CAData_t* data = (CAData_t*) OICMalloc(sizeof(CAData_t));
 417     MEMORY_ALLOCK_CHECK(data);
 418
 419     // initialize
 420     memset(data, 0, sizeof(CAData_t));
 421
 422     // clone remote endpoint
 423     CARemoteEndpoint_t* remoteEndpoint = CACloneRemoteEndpoint(object);
 424     MEMORY_ALLOCK_CHECK(remoteEndpoint);
 425
 426     // clone response info
 427     CAResponseInfo_t* responseInfo = CACloneResponseInfo(response);
 428     MEMORY_ALLOCK_CHECK(responseInfo);
 429
 430     // save data
 431     data->type = SEND_TYPE_UNICAST;
 432     data->remoteEndpoint = remoteEndpoint;
 433     data->requestInfo = NULL;
 434     data->responseInfo = responseInfo;
 435
 436     // add thread
 437     CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
 438
 439     return CA_STATUS_OK;
 440
 441     // memory error label.
 442     memory_error_exit:
 443
 444     CADestroyRemoteEndpointInternal(remoteEndpoint);
 445
 446     CADestroyResponseInfoInternal(responseInfo);
 447
 448     if (data != NULL)
 449     {
 450         OICFree(data);
 451     }
 452
 453     return CA_MEMORY_ALLOC_FAILED;
 454 }
 455
 456 CAResult_t CADetachMessageResourceUri(const CAURI_t resourceUri, const CAHeaderOption_t* options,
 457         uint8_t numOptions)
 458 {
 459     if (resourceUri == NULL)
 460     {
 461         return CA_STATUS_FAILED;
 462     }
 463
 464     CAData_t* data = (CAData_t*) OICMalloc(sizeof(CAData_t));
 465     MEMORY_ALLOCK_CHECK(data);
 466
 467     // initialize
 468     memset(data, 0, sizeof(CAData_t));
 469
 470     CAAddress_t addr;
 471     memset(&addr, 0, sizeof(CAAddress_t));
 472     CARemoteEndpoint_t* remoteEndpoint = CACreateRemoteEndpointInternal(resourceUri, addr,
 473             CA_ETHERNET | CA_WIFI | CA_EDR | CA_LE);
 474
 475     // save data
 476     data->type = SEND_TYPE_MULTICAST;
 477     data->remoteEndpoint = remoteEndpoint;
 478     data->requestInfo = NULL;
 479     data->responseInfo = NULL;
 480     data->options = NULL;
 481     data->numOptions = 0;
 482
 483     if (options != NULL && numOptions > 0)
 484     {
 485         // copy data
 486         CAHeaderOption_t* temp = (CAHeaderOption_t*) OICMalloc(
 487                 sizeof(CAHeaderOption_t) * numOptions);
 488
 489         MEMORY_ALLOCK_CHECK(temp);
 490
 491         memset(temp, 0, sizeof(CAHeaderOption_t) * numOptions);
 492         memcpy(temp, options, sizeof(CAHeaderOption_t) * numOptions);
 493
 494         data->options = temp;
 495         data->numOptions = numOptions;
 496     }
 497
 498     // add thread
 499     CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
 500
 501     return CA_STATUS_OK;
 502
 503     // memory error label.
 504     memory_error_exit:
 505
 506     CADestroyRemoteEndpointInternal(remoteEndpoint);
 507
 508     if (data != NULL)
 509     {
 510         OICFree(data);
 511     }
 512
 513     return CA_MEMORY_ALLOC_FAILED;
 514 }
 515
 516 void CASetRequestResponseCallbacks(CARequestCallback ReqHandler, CAResponseCallback RespHandler)
 517 {
 518     OIC_LOG_V(DEBUG, TAG, "set request, response handler callback.");
 519
 520     gRequestHandler = ReqHandler;
 521     gResponseHandler = RespHandler;
 522 }
 523
 524 CAResult_t CAInitializeMessageHandler()
 525 {
 526         OIC_LOG(DEBUG, TAG, "CAInitializeMessageHandler - Entry");
 527     CASetPacketReceivedCallback(CAReceivedPacketCallback);
 528
 529     CASetNetworkChangeCallback(CANetworkChangedCallback);
 530
 531     // create thread pool
 532     CAResult_t res;
 533     res = u_thread_pool_init(MAX_THREAD_POOL_SIZE, &gThreadPoolHandle);
 534
 535     if (res != CA_STATUS_OK)
 536     {
 537         OIC_LOG_V(DEBUG, TAG, "thread pool initialize error.");
 538         return res;
 539     }
 540
 541     // send thread initialize
 542     CAQueueingThreadInitialize(&gSendThread, gThreadPoolHandle, CASendThreadProcess);
 543
 544     // start send thread
 545     res = CAQueueingThreadStart(&gSendThread);
 546
 547     if (res != CA_STATUS_OK)
 548     {
 549         OIC_LOG_V(DEBUG, TAG, "thread start error(send thread).");
 550         return res;
 551     }
 552
 553     // receive thread initialize
 554     CAQueueingThreadInitialize(&gReceiveThread, gThreadPoolHandle, CAReceiveThreadProcess);
 555
 556     // start receive thread
 557     // TODO
 558     // currently not support.
 559 //    res = CAThreadStart(gReceiveThread);
 560 //
 561 //    if (res != CA_STATUS_OK)
 562 //    {
 563 //        OIC_LOG_V(DEBUG, TAG, "thread start error(receive thread).");
 564 //        return res;
 565 //    }
 566
 567     // initialize interface adapters by controller
 568     CAInitializeAdapters(gThreadPoolHandle);
 569
 570     return CA_STATUS_OK;
 571 }
 572
 573 void CATerminateMessageHandler()
 574 {
 575     // terminate interface adapters by controller
 576     CATerminateAdapters();
 577
 578     // stop thread
 579     // delete thread data
 580     CAQueueingThreadStop(&gSendThread);
 581     CAQueueingThreadDestroy(&gSendThread);
 582
 583     // stop thread
 584     // delete thread data
 585     CAQueueingThreadStop(&gReceiveThread);
 586     CAQueueingThreadDestroy(&gReceiveThread);
 587
 588     // destroy thread pool
 589     u_thread_pool_free(gThreadPoolHandle);
 590
 591     OIC_LOG_V(DEBUG, TAG, "message handler terminate completed!");
 592 }
 593