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 "caretransmission.h"
  38 #include "umutex.h"
  39 #include "oic_malloc.h"
  40 #include "caadapterutils.h"
  41 #include "canetworkconfigurator.h"
  42
  43 #define TAG PCF("CA")
  44 #define SINGLE_HANDLE
  45
  46 #define CA_MEMORY_ALLOC_CHECK(arg) { if (arg == NULL) {OIC_LOG(ERROR, TAG, "Out of memory"); \
  47 goto memory_error_exit;} }
  48
  49 #define MAX_THREAD_POOL_SIZE    20
  50
  51 typedef enum
  52 {
  53     SEND_TYPE_MULTICAST = 0, SEND_TYPE_UNICAST
  54 } CASendDataType_t;
  55
  56 typedef struct
  57 {
  58     CASendDataType_t type;
  59     CARemoteEndpoint_t *remoteEndpoint;
  60     CARequestInfo_t *requestInfo;
  61     CAResponseInfo_t *responseInfo;
  62     CAHeaderOption_t *options;
  63     uint8_t numOptions;
  64 } CAData_t;
  65
  66 // thread pool handle
  67 static u_thread_pool_t g_threadPoolHandle = NULL;
  68
  69 // message handler main thread
  70 static CAQueueingThread_t g_sendThread;
  71 static CAQueueingThread_t g_receiveThread;
  72
  73 static CARetransmission_t g_retransmissionContext;
  74
  75 // handler field
  76 static CARequestCallback g_requestHandler = NULL;
  77 static CAResponseCallback g_responseHandler = NULL;
  78
  79 static void CATimeoutCallback(const CARemoteEndpoint_t *endpoint, const void *pdu, uint32_t size)
  80 {
  81     CARemoteEndpoint_t* ep = CACloneRemoteEndpoint(endpoint);
  82     if (ep == NULL)
  83     {
  84         OIC_LOG(DEBUG, TAG, "memory allocation failed !");
  85         return;
  86     }
  87
  88     CAResponseInfo_t* resInfo = (CAResponseInfo_t*) OICCalloc(1, sizeof(CAResponseInfo_t));
  89
  90     if (resInfo == NULL)
  91     {
  92         OIC_LOG(DEBUG, TAG, "memory allocation failed !");
  93         CADestroyRemoteEndpointInternal(ep);
  94         return;
  95     }
  96
  97     CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
  98     uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
  99
 100     resInfo->result = CA_RETRANSMIT_TIMEOUT;
 101     resInfo->info.type = type;
 102     resInfo->info.messageId = messageId;
 103
 104     CAData_t *cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 105     if (cadata == NULL)
 106     {
 107         OIC_LOG(DEBUG, TAG, "memory allocation failed !");
 108         CADestroyRemoteEndpointInternal(ep);
 109         OICFree(resInfo);
 110         return;
 111     }
 112
 113     cadata->type = SEND_TYPE_UNICAST;
 114     cadata->remoteEndpoint = ep;
 115     cadata->requestInfo = NULL;
 116     cadata->responseInfo = resInfo;
 117
 118     CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 119 }
 120
 121 static void CADataDestroyer(void *data, uint32_t size)
 122 {
 123     CAData_t *cadata = (CAData_t *) data;
 124
 125     if (cadata == NULL)
 126     {
 127         return;
 128     }
 129
 130     if (cadata->remoteEndpoint != NULL)
 131     {
 132         CADestroyRemoteEndpointInternal((CARemoteEndpoint_t *)cadata->remoteEndpoint);
 133     }
 134
 135     if (cadata->requestInfo != NULL)
 136     {
 137         CADestroyRequestInfoInternal((CARequestInfo_t *)cadata->requestInfo);
 138     }
 139
 140     if (cadata->responseInfo != NULL)
 141     {
 142         CADestroyResponseInfoInternal((CAResponseInfo_t *)cadata->responseInfo);
 143     }
 144
 145     if (cadata->options != NULL)
 146     {
 147         OICFree(cadata->options);
 148     }
 149
 150     OICFree(cadata);
 151 }
 152
 153 static void CAReceiveThreadProcess(void *threadData)
 154 {
 155     // Currently not supported
 156     // This will be enabled when RI supports multi threading
 157 #ifndef SINGLE_HANDLE
 158     CAData_t *data = (CAData_t *) threadData;
 159
 160     if (data == NULL)
 161     {
 162         OIC_LOG(ERROR, TAG, "thread data error!!");
 163         return;
 164     }
 165
 166     // parse the data and call the callbacks.
 167     // #1 parse the data
 168     // #2 get endpoint
 169     CARemoteEndpoint_t *rep = (CARemoteEndpoint_t *)(data->remoteEndpoint);
 170
 171     if (rep == NULL)
 172         return;
 173
 174     if (data->requestInfo != NULL)
 175     {
 176         if (g_requestHandler)
 177         {
 178             g_requestHandler(rep, data->requestInfo);
 179         }
 180     }
 181
 182     if (data->responseInfo != NULL)
 183     {
 184         if (g_responseHandler)
 185         {
 186             g_responseHandler(rep, data->responseInfo);
 187         }
 188     }
 189 #endif
 190 }
 191
 192 static void CASendThreadProcess(void *threadData)
 193 {
 194     CAData_t *data = (CAData_t *) threadData;
 195
 196     if (data == NULL)
 197     {
 198         OIC_LOG(ERROR, TAG, "thread data error!!");
 199         return;
 200     }
 201
 202     if (NULL == data->remoteEndpoint)
 203     {
 204         OIC_LOG(DEBUG, TAG, "remoteEndpoint is null");
 205         return;
 206     }
 207
 208     CAResult_t res = CA_STATUS_FAILED;
 209
 210     CASendDataType_t type = data->type;
 211
 212     if (type == SEND_TYPE_UNICAST)
 213     {
 214         coap_pdu_t *pdu = NULL;
 215
 216         if (data->requestInfo != NULL)
 217         {
 218             OIC_LOG(DEBUG, TAG, "requestInfo is available..");
 219
 220             pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri,
 221                                                data->requestInfo->method,
 222                                                data->requestInfo->info);
 223         }
 224         else if (data->responseInfo != NULL)
 225         {
 226             OIC_LOG(DEBUG, TAG, "responseInfo is available..");
 227
 228             pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri,
 229                                                data->responseInfo->result,
 230                                                data->responseInfo->info);
 231         }
 232         else
 233         {
 234             OIC_LOG(DEBUG, TAG, "request info, response info is empty");
 235         }
 236
 237         // interface controller function call.
 238         if (NULL != pdu)
 239         {
 240             OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
 241
 242             OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
 243
 244             OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
 245
 246             OIC_LOG_V(DEBUG, TAG, "PDU Maker - id : %d", ntohs(pdu->hdr->id));
 247
 248             OIC_LOG(DEBUG, TAG, "PDU Maker - token :");
 249
 250             OIC_LOG_BUFFER(DEBUG, TAG, pdu->hdr->token, pdu->hdr->token_length);
 251
 252             res = CASendUnicastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 253
 254             // for retransmission
 255             CARetransmissionSentData(&g_retransmissionContext, data->remoteEndpoint, pdu->hdr,
 256                                      pdu->length);
 257
 258             coap_delete_pdu(pdu);
 259         }
 260     }
 261     else if (type == SEND_TYPE_MULTICAST)
 262     {
 263         coap_pdu_t *pdu = NULL;
 264         CAInfo_t info = {};
 265
 266         info.options = data->options;
 267         info.numOptions = data->numOptions;
 268         info.token = data->requestInfo->info.token;
 269         info.tokenLength = data->requestInfo->info.tokenLength;
 270         info.type = data->requestInfo->info.type;
 271
 272         pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri, CA_GET, info);
 273
 274         if (NULL != pdu)
 275         {
 276             OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
 277
 278             OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
 279
 280             OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
 281
 282             OIC_LOG_V(DEBUG, TAG, "PDU Maker - id : %d", ntohs(pdu->hdr->id));
 283
 284             OIC_LOG(DEBUG, TAG, "PDU Maker - token");
 285
 286             OIC_LOG_BUFFER(DEBUG, TAG, pdu->hdr->token, pdu->hdr->token_length);
 287
 288             res = CASendMulticastData(pdu->hdr, pdu->length);
 289             coap_delete_pdu(pdu);
 290         }
 291     }
 292
 293     OIC_LOG_V(DEBUG, TAG, " Result :%d", res);
 294 }
 295
 296 static void CAReceivedPacketCallback(CARemoteEndpoint_t *endpoint, void *data,
 297                                      uint32_t dataLen)
 298 {
 299     OIC_LOG(DEBUG, TAG, "receivedPacketCallback in message handler!!");
 300
 301     if (NULL == data)
 302     {
 303         OIC_LOG(DEBUG, TAG, "received data is null");
 304         return;
 305     }
 306
 307     coap_pdu_t *pdu;
 308     uint32_t code = CA_NOT_FOUND;
 309
 310     OIC_LOG_V(DEBUG, TAG, "data : %s", data);
 311     pdu = (coap_pdu_t *) CAParsePDU((const char *) data, dataLen, &code);
 312     OICFree(data);
 313
 314     if(NULL == pdu)
 315     {
 316         OIC_LOG(DEBUG, TAG, "pdu is null");
 317         return;
 318     }
 319
 320     char uri[CA_MAX_URI_LENGTH] = { 0, };
 321     uint32_t bufLen = CA_MAX_URI_LENGTH;
 322
 323     if (code == CA_GET || code == CA_POST || code == CA_PUT || code == CA_DELETE)
 324     {
 325         CARequestInfo_t *ReqInfo;
 326         ReqInfo = (CARequestInfo_t *) OICCalloc(1, sizeof(CARequestInfo_t));
 327         if (ReqInfo == NULL)
 328         {
 329             OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 330             coap_delete_pdu(pdu);
 331             CAAdapterFreeRemoteEndpoint(endpoint);
 332             return;
 333         }
 334         CAGetRequestInfoFromPdu(pdu, ReqInfo, uri, bufLen);
 335
 336         if (NULL != ReqInfo->info.options)
 337         {
 338             uint32_t i;
 339             for (i = 0; i < ReqInfo->info.numOptions; i++)
 340             {
 341                 OIC_LOG_V(DEBUG, TAG, "Request- optionID: %d", ReqInfo->info.options[i].optionID);
 342
 343                 OIC_LOG_V(DEBUG, TAG, "Request- list: %s", ReqInfo->info.options[i].optionData);
 344             }
 345         }
 346
 347         if (NULL != ReqInfo->info.payload)
 348         {
 349             OIC_LOG_V(DEBUG, TAG, "Request- payload: %s", ReqInfo->info.payload);
 350         }
 351         OIC_LOG_V(DEBUG, TAG, "Request- code: %d", ReqInfo->method);
 352         if (NULL != ReqInfo->info.token)
 353         {
 354             OIC_LOG(DEBUG, TAG, "Request- token:");
 355             OIC_LOG_BUFFER(DEBUG, TAG, (const uint8_t *) ReqInfo->info.token,
 356                            ReqInfo->info.tokenLength);
 357         }
 358
 359         if (NULL != endpoint)
 360         {
 361             endpoint->resourceUri = (char *) OICCalloc(bufLen + 1, sizeof(char));
 362             if (endpoint->resourceUri == NULL)
 363             {
 364                 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 365                 OICFree(ReqInfo);
 366                 coap_delete_pdu(pdu);
 367                 CAAdapterFreeRemoteEndpoint(endpoint);
 368                 return;
 369             }
 370             memcpy(endpoint->resourceUri, uri, bufLen);
 371             OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
 372         }
 373         // store the data at queue.
 374         CAData_t *cadata = NULL;
 375         cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 376         if (cadata == NULL)
 377         {
 378             OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 379             if (endpoint != NULL && endpoint->resourceUri != NULL)
 380                 OICFree(endpoint->resourceUri);
 381             OICFree(ReqInfo);
 382             coap_delete_pdu(pdu);
 383             CAAdapterFreeRemoteEndpoint(endpoint);
 384             return;
 385         }
 386
 387         cadata->type = SEND_TYPE_UNICAST;
 388         cadata->remoteEndpoint = endpoint;
 389         cadata->requestInfo = ReqInfo;
 390         cadata->responseInfo = NULL;
 391         CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 392     }
 393     else
 394     {
 395         CAResponseInfo_t *ResInfo;
 396         ResInfo = (CAResponseInfo_t *) OICCalloc(1, sizeof(CAResponseInfo_t));
 397         if (ResInfo == NULL)
 398         {
 399             OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 400             coap_delete_pdu(pdu);
 401             CAAdapterFreeRemoteEndpoint(endpoint);
 402             return;
 403         }
 404         CAGetResponseInfoFromPdu(pdu, ResInfo, uri, bufLen);
 405
 406         if (NULL != ResInfo->info.options)
 407         {
 408             uint32_t i;
 409             for (i = 0; i < ResInfo->info.numOptions; i++)
 410             {
 411                 OIC_LOG_V(DEBUG, TAG, "Response- optionID: %d", ResInfo->info.options[i].optionID);
 412
 413                 OIC_LOG_V(DEBUG, TAG, "Response- list: %s", ResInfo->info.options[i].optionData);
 414             }
 415             if (NULL != ResInfo->info.payload)
 416             {
 417                 OIC_LOG_V(DEBUG, TAG, "Response- payload: %s", ResInfo->info.payload);
 418             }
 419             OIC_LOG_V(DEBUG, TAG, "Response- code: %d", ResInfo->result);
 420         }
 421
 422         if (NULL != endpoint)
 423         {
 424             endpoint->resourceUri = (char *) OICCalloc(bufLen + 1, sizeof(char));
 425             if (endpoint->resourceUri == NULL)
 426             {
 427                 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 428                 OICFree(ResInfo);
 429                 coap_delete_pdu(pdu);
 430                 CAAdapterFreeRemoteEndpoint(endpoint);
 431                 return;
 432             }
 433             memcpy(endpoint->resourceUri, uri, bufLen);
 434             OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
 435         }
 436
 437         // store the data at queue.
 438         CAData_t *cadata = NULL;
 439         cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 440         if (cadata == NULL)
 441         {
 442             OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
 443             if (endpoint != NULL && endpoint->resourceUri != NULL)
 444                 OICFree(endpoint->resourceUri);
 445             OICFree(ResInfo);
 446             coap_delete_pdu(pdu);
 447             CAAdapterFreeRemoteEndpoint(endpoint);
 448             return;
 449         }
 450
 451         cadata->type = SEND_TYPE_UNICAST;
 452         cadata->remoteEndpoint = endpoint;
 453         cadata->requestInfo = NULL;
 454         cadata->responseInfo = ResInfo;
 455
 456         CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 457
 458         // for retransmission
 459         CARetransmissionReceivedData(&g_retransmissionContext, endpoint, pdu->hdr, pdu->length);
 460     }
 461
 462     if(pdu)
 463     {
 464         coap_delete_pdu(pdu);
 465     }
 466 }
 467
 468 static void CANetworkChangedCallback(CALocalConnectivity_t *info, CANetworkStatus_t status)
 469 {
 470     OIC_LOG(DEBUG, TAG, "networkChangeCallback in message handler!!");
 471
 472     OIC_LOG_V(DEBUG, TAG, "Changed Network Status: %d", status);
 473 }
 474
 475 void CAHandleRequestResponseCallbacks()
 476 {
 477     OIC_LOG(DEBUG, TAG, "CAHandleRequestResponseCallbacks IN");
 478
 479     // parse the data and call the callbacks.
 480     // #1 parse the data
 481     // #2 get endpoint
 482
 483     u_mutex_lock(g_receiveThread.threadMutex);
 484
 485     u_queue_message_t *item = u_queue_get_element(g_receiveThread.dataQueue);
 486
 487     u_mutex_unlock(g_receiveThread.threadMutex);
 488
 489     if (item == NULL)
 490         return;
 491
 492     // get values
 493     void *msg = item->msg;
 494
 495     if (msg == NULL)
 496         return;
 497
 498     // get endpoint
 499     CAData_t *td = (CAData_t *) msg;
 500     CARemoteEndpoint_t *rep = td->remoteEndpoint;
 501
 502     if (rep == NULL)
 503         return;
 504
 505     if (td->requestInfo != NULL)
 506     {
 507         if (g_requestHandler)
 508         {
 509             g_requestHandler(rep, td->requestInfo);
 510         }
 511
 512         OICFree(td->requestInfo->info.options);
 513         OICFree(td->requestInfo->info.payload);
 514         OICFree(td->requestInfo->info.token);
 515         OICFree(td->requestInfo);
 516     }
 517
 518     if (td->responseInfo != NULL)
 519     {
 520         if (g_responseHandler)
 521         {
 522             g_responseHandler(rep, td->responseInfo);
 523         }
 524
 525         OICFree(td->responseInfo->info.options);
 526         OICFree(td->responseInfo->info.payload);
 527         OICFree(td->responseInfo->info.token);
 528         OICFree(td->responseInfo);
 529     }
 530
 531     if (NULL != rep->resourceUri)
 532     {
 533         OICFree(rep->resourceUri);
 534     }
 535
 536     OICFree(rep);
 537     OIC_LOG(DEBUG, TAG, "CAHandleRequestResponseCallbacks OUT");
 538 }
 539
 540 CAResult_t CADetachRequestMessage(const CARemoteEndpoint_t *object,
 541                                   const CARequestInfo_t *request)
 542 {
 543     OIC_LOG(DEBUG, TAG, "CADetachRequestMessage");
 544
 545     if (object == NULL || request == NULL)
 546     {
 547         return CA_STATUS_FAILED;
 548     }
 549
 550     CARemoteEndpoint_t *remoteEndpoint = NULL;
 551     CARequestInfo_t *requestInfo = NULL;
 552
 553     CAData_t *data = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 554     CA_MEMORY_ALLOC_CHECK(data);
 555
 556     // clone remote endpoint
 557
 558     remoteEndpoint = CACloneRemoteEndpoint(object);
 559     CA_MEMORY_ALLOC_CHECK(remoteEndpoint);
 560
 561     // clone request info
 562     requestInfo = CACloneRequestInfo(request);
 563     CA_MEMORY_ALLOC_CHECK(requestInfo);
 564
 565     // save data
 566     data->type = SEND_TYPE_UNICAST;
 567     data->remoteEndpoint = remoteEndpoint;
 568     data->requestInfo = requestInfo;
 569     data->responseInfo = NULL;
 570
 571     // add thread
 572     CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 573
 574     return CA_STATUS_OK;
 575
 576     // memory error label.
 577 memory_error_exit:
 578
 579     CADestroyRemoteEndpointInternal(remoteEndpoint);
 580
 581     CADestroyRequestInfoInternal(requestInfo);
 582
 583     if (data != NULL)
 584     {
 585         OICFree(data);
 586     }
 587
 588     return CA_MEMORY_ALLOC_FAILED;
 589 }
 590
 591 CAResult_t CADetachRequestToAllMessage(const CAGroupEndpoint_t *object,
 592                                        const CARequestInfo_t *request)
 593 {
 594     // ToDo
 595     OIC_LOG(DEBUG, TAG, "CADetachRequestToAllMessage");
 596
 597
 598     if (object == NULL || request == NULL)
 599     {
 600         return CA_STATUS_FAILED;
 601     }
 602
 603     CARemoteEndpoint_t *remoteEndpoint = NULL;
 604     CARequestInfo_t *requestInfo = NULL;
 605
 606     CAData_t *data = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 607     CA_MEMORY_ALLOC_CHECK(data);
 608
 609     CAAddress_t addr;
 610     memset(&addr, 0, sizeof(CAAddress_t));
 611     remoteEndpoint = CACreateRemoteEndpointInternal(object->resourceUri, addr,
 612                                          object->connectivityType);
 613
 614     // clone request info
 615     requestInfo = CACloneRequestInfo(request);
 616     CA_MEMORY_ALLOC_CHECK(requestInfo);
 617
 618     // save data
 619     data->type = SEND_TYPE_MULTICAST;
 620     data->remoteEndpoint = remoteEndpoint;
 621     data->requestInfo = requestInfo;
 622     data->responseInfo = NULL;
 623
 624     // add thread
 625     CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 626
 627     return CA_STATUS_OK;
 628
 629     // memory error label.
 630 memory_error_exit:
 631
 632     CADestroyRemoteEndpointInternal(remoteEndpoint);
 633
 634     CADestroyRequestInfoInternal(requestInfo);
 635
 636     if (data != NULL)
 637     {
 638         OICFree(data);
 639     }
 640
 641     return CA_MEMORY_ALLOC_FAILED;
 642 }
 643
 644 CAResult_t CADetachResponseMessage(const CARemoteEndpoint_t *object,
 645                                    const CAResponseInfo_t *response)
 646 {
 647     OIC_LOG(DEBUG, TAG, "CADetachResponseMessage");
 648
 649     if (object == NULL || response == NULL)
 650     {
 651         return CA_STATUS_FAILED;
 652     }
 653
 654     CARemoteEndpoint_t *remoteEndpoint = NULL;
 655     CAResponseInfo_t *responseInfo = NULL;
 656
 657     CAData_t *data = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 658     CA_MEMORY_ALLOC_CHECK(data);
 659
 660     // clone remote endpoint
 661     remoteEndpoint = CACloneRemoteEndpoint(object);
 662     CA_MEMORY_ALLOC_CHECK(remoteEndpoint);
 663
 664     // clone response info
 665     responseInfo = CACloneResponseInfo(response);
 666     CA_MEMORY_ALLOC_CHECK(responseInfo);
 667
 668     // save data
 669     data->type = SEND_TYPE_UNICAST;
 670     data->remoteEndpoint = remoteEndpoint;
 671     data->requestInfo = NULL;
 672     data->responseInfo = responseInfo;
 673
 674     // add thread
 675     CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 676
 677     return CA_STATUS_OK;
 678
 679     // memory error label.
 680 memory_error_exit:
 681
 682     CADestroyRemoteEndpointInternal(remoteEndpoint);
 683
 684     CADestroyResponseInfoInternal(responseInfo);
 685
 686     if (data != NULL)
 687     {
 688         OICFree(data);
 689     }
 690
 691     return CA_MEMORY_ALLOC_FAILED;
 692 }
 693
 694 CAResult_t CADetachMessageResourceUri(const CAURI_t resourceUri, const CAToken_t token,
 695                                       uint8_t tokenLength, const CAHeaderOption_t *options,
 696                                       uint8_t numOptions)
 697 {
 698     OIC_LOG(DEBUG, TAG, "IN");
 699     VERIFY_NON_NULL(resourceUri, TAG, "resourceUri is NULL");
 700     VERIFY_NON_NULL(token, TAG, "Token is NULL");
 701
 702     CARemoteEndpoint_t *remoteEndpoint = NULL;
 703     CARequestInfo_t *reqInfo = NULL;
 704     char *tempToken = NULL;
 705
 706     // allocate & initialize
 707     CAData_t *data = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 708     CA_MEMORY_ALLOC_CHECK(data);
 709
 710     CAAddress_t addr = {};
 711     remoteEndpoint = CACreateRemoteEndpointInternal(resourceUri, addr,
 712                                                     CA_ETHERNET | CA_WIFI | CA_EDR | CA_LE);
 713
 714     // create request info
 715     reqInfo = (CARequestInfo_t *) OICCalloc(1, sizeof(CARequestInfo_t));
 716     CA_MEMORY_ALLOC_CHECK(reqInfo);
 717
 718     if (tokenLength)
 719     {
 720         // copy token value
 721         tempToken = (char *) OICMalloc(tokenLength);
 722         CA_MEMORY_ALLOC_CHECK(tempToken);
 723         memcpy(tempToken, token, tokenLength);
 724     }
 725
 726     // save request info data
 727     reqInfo->method = CA_GET;
 728     reqInfo->info.type = CA_MSG_NONCONFIRM;
 729
 730     reqInfo->info.token = tempToken;
 731     reqInfo->info.tokenLength = tokenLength;
 732
 733     // save data
 734     data->type = SEND_TYPE_MULTICAST;
 735     data->remoteEndpoint = remoteEndpoint;
 736     data->requestInfo = reqInfo;
 737
 738     data->responseInfo = NULL;
 739     data->options = NULL;
 740     data->numOptions = 0;
 741
 742     if (options != NULL && numOptions > 0)
 743     {
 744         // copy data
 745         CAHeaderOption_t *temp = (CAHeaderOption_t *) OICCalloc(numOptions,
 746                                      sizeof(CAHeaderOption_t));
 747         CA_MEMORY_ALLOC_CHECK(temp);
 748
 749         memcpy(temp, options, sizeof(CAHeaderOption_t) * numOptions);
 750
 751         data->options = temp;
 752         data->numOptions = numOptions;
 753     }
 754
 755     // add thread
 756     CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 757
 758     return CA_STATUS_OK;
 759
 760     // memory error label.
 761 memory_error_exit:
 762
 763     CADestroyRemoteEndpointInternal(remoteEndpoint);
 764
 765     OICFree(tempToken);
 766     OICFree(reqInfo);
 767     OICFree(data);
 768     OIC_LOG(DEBUG, TAG, "OUT");
 769     return CA_MEMORY_ALLOC_FAILED;
 770 }
 771
 772 void CASetRequestResponseCallbacks(CARequestCallback ReqHandler,
 773                                    CAResponseCallback RespHandler)
 774 {
 775     OIC_LOG(DEBUG, TAG, "set request, response handler callback.");
 776
 777     g_requestHandler = ReqHandler;
 778     g_responseHandler = RespHandler;
 779 }
 780
 781 CAResult_t CAInitializeMessageHandler()
 782 {
 783     OIC_LOG(DEBUG, TAG, "CAInitializeMessageHandler - Entry");
 784     CASetPacketReceivedCallback(CAReceivedPacketCallback);
 785
 786     CASetNetworkChangeCallback(CANetworkChangedCallback);
 787
 788     // create thread pool
 789     CAResult_t res;
 790     res = u_thread_pool_init(MAX_THREAD_POOL_SIZE, &g_threadPoolHandle);
 791
 792     if (res != CA_STATUS_OK)
 793     {
 794         OIC_LOG(ERROR, TAG, "thread pool initialize error.");
 795         return res;
 796     }
 797
 798     // send thread initialize
 799     CAQueueingThreadInitialize(&g_sendThread, g_threadPoolHandle, CASendThreadProcess,
 800             CADataDestroyer);
 801
 802     // start send thread
 803     res = CAQueueingThreadStart(&g_sendThread);
 804
 805     if (res != CA_STATUS_OK)
 806     {
 807         OIC_LOG(ERROR, TAG, "thread start error(send thread).");
 808         u_thread_pool_free(g_threadPoolHandle);
 809         g_threadPoolHandle = NULL;
 810         return res;
 811     }
 812
 813     // receive thread initialize
 814     CAQueueingThreadInitialize(&g_receiveThread, g_threadPoolHandle, CAReceiveThreadProcess,
 815             CADataDestroyer);
 816
 817 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
 818     // start receive thread
 819     res = CAQueueingThreadStart(&gReceiveThread);
 820
 821     if (res != CA_STATUS_OK)
 822     {
 823         OIC_LOG(ERROR, TAG, "thread start error(receive thread).");
 824         return res;
 825     }
 826 #endif
 827
 828     // retransmission initialize
 829     CARetransmissionInitialize(&g_retransmissionContext, g_threadPoolHandle, CASendUnicastData,
 830             CATimeoutCallback, NULL);
 831
 832     // start retransmission
 833     res = CARetransmissionStart(&g_retransmissionContext);
 834
 835     if (res != CA_STATUS_OK)
 836     {
 837         OIC_LOG(ERROR, TAG, "thread start error(retransmission thread).");
 838         return res;
 839     }
 840
 841     // initialize interface adapters by controller
 842     CAInitializeAdapters(g_threadPoolHandle);
 843
 844     return CA_STATUS_OK;
 845 }
 846
 847 void CATerminateMessageHandler()
 848 {
 849     uint8_t i = 0;
 850     CAConnectivityType_t connType;
 851     u_arraylist_t *list = CAGetSelectedNetworkList();
 852     uint32_t length = u_arraylist_length(list);
 853
 854     for (i = 0; i < length; i++)
 855     {
 856         void* ptrType = u_arraylist_get(list, i);
 857
 858         if (ptrType == NULL)
 859         {
 860             continue;
 861         }
 862
 863         connType = *(CAConnectivityType_t *) ptrType;
 864         CAStopAdapter(connType);
 865     }
 866
 867     // stop retransmission
 868     if (g_retransmissionContext.threadMutex != NULL)
 869     {
 870         CARetransmissionStop(&g_retransmissionContext);
 871     }
 872
 873     // stop thread
 874     // delete thread data
 875     if (g_sendThread.threadMutex != NULL)
 876     {
 877         CAQueueingThreadStop(&g_sendThread);
 878     }
 879
 880     // stop thread
 881     // delete thread data
 882     if (g_receiveThread.threadMutex != NULL)
 883     {
 884 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
 885         CAQueueingThreadStop(&gReceiveThread);
 886 #endif
 887     }
 888
 889     // destroy thread pool
 890     if (g_threadPoolHandle != NULL)
 891     {
 892         u_thread_pool_free(g_threadPoolHandle);
 893         g_threadPoolHandle = NULL;
 894     }
 895
 896     CARetransmissionDestroy(&g_retransmissionContext);
 897     CAQueueingThreadDestroy(&g_sendThread);
 898     CAQueueingThreadDestroy(&g_receiveThread);
 899
 900     // terminate interface adapters by controller
 901     CATerminateAdapters();
 902
 903     OIC_LOG(DEBUG, TAG, "message handler termination is complete!");
 904 }
 905