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