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