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 "caprotocolmessage.h"
  30 #include "logger.h"
  31 #include "config.h" /* for coap protocol */
  32 #include "oic_malloc.h"
  33 #include "canetworkconfigurator.h"
  34 #include "caadapterutils.h"
  35 #include "cainterfacecontroller.h"
  36 #include "caretransmission.h"
  37
  38 #ifdef WITH_BWT
  39 #include "cablockwisetransfer.h"
  40 #endif
  41
  42 #ifndef  SINGLE_THREAD
  43 #include "uqueue.h"
  44 #include "cathreadpool.h" /* for thread pool */
  45 #include "caqueueingthread.h"
  46
  47 #define SINGLE_HANDLE
  48 #define MAX_THREAD_POOL_SIZE    20
  49
  50 // thread pool handle
  51 static ca_thread_pool_t g_threadPoolHandle = NULL;
  52
  53 // message handler main thread
  54 static CAQueueingThread_t g_sendThread;
  55 static CAQueueingThread_t g_receiveThread;
  56
  57 #else
  58 #define CA_MAX_RT_ARRAY_SIZE    3
  59 #endif  /* SINGLE_THREAD */
  60
  61 #define TAG "CA_MSG_HNDLR"
  62
  63 static CARetransmission_t g_retransmissionContext;
  64
  65 // handler field
  66 static CARequestCallback g_requestHandler = NULL;
  67 static CAResponseCallback g_responseHandler = NULL;
  68 static CAErrorCallback g_errorHandler = NULL;
  69
  70 static void CAErrorHandler(const CAEndpoint_t *endpoint,
  71                            const void *data, uint32_t dataLen,
  72                            CAResult_t result);
  73
  74 static CAData_t* CAGenerateHandlerData(const CAEndpoint_t *endpoint, const void *data,
  75                                        CADataType_t dataType);
  76
  77 #ifdef SINGLE_THREAD
  78 static void CAProcessReceivedData(CAData_t *data);
  79 #endif
  80 static void CADestroyData(void *data, uint32_t size);
  81 static void CALogPayloadInfo(CAInfo_t *info);
  82 static bool CADropSecondRequest(const CAEndpoint_t *endpoint, uint16_t messageId);
  83
  84 #ifdef WITH_BWT
  85 void CAAddDataToSendThread(CAData_t *data)
  86 {
  87     OIC_LOG(DEBUG, TAG, "IN");
  88     VERIFY_NON_NULL_VOID(data, TAG, "data");
  89
  90     // add thread
  91     CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
  92
  93     OIC_LOG(DEBUG, TAG, "OUT");
  94 }
  95
  96 void CAAddDataToReceiveThread(CAData_t *data)
  97 {
  98     OIC_LOG(DEBUG, TAG, "IN - CAAddDataToReceiveThread");
  99     VERIFY_NON_NULL_VOID(data, TAG, "data");
 100
 101     // add thread
 102     CAQueueingThreadAddData(&g_receiveThread, data, sizeof(CAData_t));
 103
 104     OIC_LOG(DEBUG, TAG, "OUT - CAAddDataToReceiveThread");
 105 }
 106 #endif
 107
 108 static bool CAIsSelectedNetworkAvailable()
 109 {
 110     u_arraylist_t *list = CAGetSelectedNetworkList();
 111     if (!list || list->length == 0)
 112     {
 113         OIC_LOG(ERROR, TAG, "No selected network");
 114         return false;
 115     }
 116
 117     return true;
 118 }
 119
 120 static CAData_t* CAGenerateHandlerData(const CAEndpoint_t *endpoint, const void *data, CADataType_t dataType)
 121 {
 122     OIC_LOG(DEBUG, TAG, "CAGenerateHandlerData IN");
 123     CAInfo_t *info = NULL;
 124     CAData_t *cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 125     if (!cadata)
 126     {
 127         OIC_LOG(ERROR, TAG, "memory allocation failed");
 128         return NULL;
 129     }
 130
 131     CAEndpoint_t* ep = CACloneEndpoint(endpoint);
 132     if (!ep)
 133     {
 134         OIC_LOG(ERROR, TAG, "endpoint clone failed");
 135         OICFree(cadata);
 136         return NULL;
 137     }
 138
 139     OIC_LOG_V(DEBUG, TAG, "address : %s", ep->addr);
 140     CAResult_t result;
 141
 142     if(CA_RESPONSE_DATA == dataType)
 143     {
 144         CAResponseInfo_t* resInfo = (CAResponseInfo_t*)OICCalloc(1, sizeof(CAResponseInfo_t));
 145         if (!resInfo)
 146         {
 147             OIC_LOG(ERROR, TAG, "memory allocation failed");
 148             OICFree(cadata);
 149             CAFreeEndpoint(ep);
 150             return NULL;
 151         }
 152
 153         result = CAGetResponseInfoFromPDU(data, resInfo);
 154         if (CA_STATUS_OK != result)
 155         {
 156             OIC_LOG(ERROR, TAG, "CAGetResponseInfoFromPDU Failed");
 157             CAFreeEndpoint(ep);
 158             CADestroyResponseInfoInternal(resInfo);
 159             OICFree(cadata);
 160             return NULL;
 161         }
 162         cadata->responseInfo = resInfo;
 163         info = &resInfo->info;
 164         OIC_LOG(DEBUG, TAG, "Response Info :");
 165         CALogPayloadInfo(info);
 166     }
 167     else if (CA_REQUEST_DATA == dataType)
 168     {
 169         CARequestInfo_t* reqInfo = (CARequestInfo_t*)OICCalloc(1, sizeof(CARequestInfo_t));
 170         if (!reqInfo)
 171         {
 172             OIC_LOG(ERROR, TAG, "memory allocation failed");
 173             OICFree(cadata);
 174             CAFreeEndpoint(ep);
 175             return NULL;
 176         }
 177
 178         result = CAGetRequestInfoFromPDU(data, reqInfo);
 179         if (CA_STATUS_OK != result)
 180         {
 181             OIC_LOG(ERROR, TAG, "CAGetRequestInfoFromPDU failed");
 182             CAFreeEndpoint(ep);
 183             CADestroyRequestInfoInternal(reqInfo);
 184             OICFree(cadata);
 185             return NULL;
 186         }
 187
 188         if (CADropSecondRequest(endpoint, reqInfo->info.messageId))
 189         {
 190             OIC_LOG(ERROR, TAG, "Second Request with same Token, Drop it");
 191             CAFreeEndpoint(ep);
 192             CADestroyRequestInfoInternal(reqInfo);
 193             OICFree(cadata);
 194             return NULL;
 195         }
 196
 197         cadata->requestInfo = reqInfo;
 198         info = &reqInfo->info;
 199         OIC_LOG(DEBUG, TAG, "Request Info :");
 200         CALogPayloadInfo(info);
 201    }
 202     else if (CA_ERROR_DATA == dataType)
 203     {
 204         CAErrorInfo_t *errorInfo = (CAErrorInfo_t *)OICCalloc(1, sizeof (CAErrorInfo_t));
 205         if (!errorInfo)
 206         {
 207             OIC_LOG(ERROR, TAG, "Memory allocation failed!");
 208             OICFree(cadata);
 209             CAFreeEndpoint(ep);
 210             return NULL;
 211         }
 212
 213         CAResult_t result = CAGetErrorInfoFromPDU(data, errorInfo);
 214         if (CA_STATUS_OK != result)
 215         {
 216             OIC_LOG(ERROR, TAG, "CAGetErrorInfoFromPDU failed");
 217             CAFreeEndpoint(ep);
 218             OICFree(errorInfo);
 219             OICFree(cadata);
 220             return NULL;
 221         }
 222
 223         cadata->errorInfo = errorInfo;
 224         info = &errorInfo->info;
 225         OIC_LOG(DEBUG, TAG, "error Info :");
 226         CALogPayloadInfo(info);
 227     }
 228
 229     cadata->remoteEndpoint = ep;
 230     cadata->dataType = dataType;
 231
 232     return cadata;
 233
 234     OIC_LOG(DEBUG, TAG, "CAGenerateHandlerData OUT");
 235 }
 236
 237 static void CATimeoutCallback(const CAEndpoint_t *endpoint, const void *pdu, uint32_t size)
 238 {
 239     OIC_LOG(DEBUG, TAG, "IN");
 240     VERIFY_NON_NULL_VOID(endpoint, TAG, "endpoint");
 241     VERIFY_NON_NULL_VOID(pdu, TAG, "pdu");
 242
 243     CAEndpoint_t* ep = CACloneEndpoint(endpoint);
 244     if (!ep)
 245     {
 246         OIC_LOG(ERROR, TAG, "clone failed");
 247         return;
 248     }
 249
 250     CAResponseInfo_t* resInfo = (CAResponseInfo_t*)OICCalloc(1, sizeof(CAResponseInfo_t));
 251
 252     if (!resInfo)
 253     {
 254         OIC_LOG(ERROR, TAG, "calloc failed");
 255         CAFreeEndpoint(ep);
 256         return;
 257     }
 258
 259     resInfo->result = CA_RETRANSMIT_TIMEOUT;
 260     resInfo->info.type = CAGetMessageTypeFromPduBinaryData(pdu, size);
 261     resInfo->info.messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
 262
 263     CAResult_t res = CAGetTokenFromPDU((const coap_hdr_t *) pdu, &(resInfo->info));
 264     if (CA_STATUS_OK != res)
 265     {
 266         OIC_LOG(ERROR, TAG, "fail to get Token from retransmission list");
 267         CADestroyResponseInfoInternal(resInfo);
 268         CAFreeEndpoint(ep);
 269         return;
 270     }
 271
 272     CAData_t *cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 273     if (NULL == cadata)
 274     {
 275         OIC_LOG(ERROR, TAG, "memory allocation failed !");
 276         CAFreeEndpoint(ep);
 277         CADestroyResponseInfoInternal(resInfo);
 278         return;
 279     }
 280
 281     cadata->type = SEND_TYPE_UNICAST;
 282     cadata->remoteEndpoint = ep;
 283     cadata->requestInfo = NULL;
 284     cadata->responseInfo = resInfo;
 285
 286 #ifdef SINGLE_THREAD
 287     CAProcessReceivedData(cadata);
 288 #else
 289     CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 290 #endif
 291
 292     OIC_LOG(DEBUG, TAG, "OUT");
 293 }
 294
 295 static void CADestroyData(void *data, uint32_t size)
 296 {
 297     OIC_LOG(DEBUG, TAG, "CADestroyData IN");
 298     CAData_t *cadata = (CAData_t *) data;
 299
 300     if (NULL == cadata)
 301     {
 302         OIC_LOG(ERROR, TAG, "cadata is NULL");
 303         return;
 304     }
 305
 306     if (NULL != cadata->remoteEndpoint)
 307     {
 308         CAFreeEndpoint(cadata->remoteEndpoint);
 309     }
 310
 311     if (NULL != cadata->requestInfo)
 312     {
 313         CADestroyRequestInfoInternal((CARequestInfo_t *) cadata->requestInfo);
 314     }
 315
 316     if (NULL != cadata->responseInfo)
 317     {
 318         CADestroyResponseInfoInternal((CAResponseInfo_t *) cadata->responseInfo);
 319     }
 320
 321     if (NULL != cadata->errorInfo)
 322     {
 323         CADestroyErrorInfoInternal(cadata->errorInfo);
 324     }
 325
 326     OICFree(cadata->options);
 327     OICFree(cadata);
 328     OIC_LOG(DEBUG, TAG, "CADestroyData OUT");
 329 }
 330
 331 #ifdef SINGLE_THREAD
 332 static void CAProcessReceivedData(CAData_t *data)
 333 {
 334     OIC_LOG(DEBUG, TAG, "CAProcessReceivedData IN");
 335     if (!data)
 336     {
 337         OIC_LOG(ERROR, TAG, "thread data error!!");
 338         return;
 339     }
 340
 341     // parse the data and call the callbacks.
 342     // #1 parse the data
 343     // #2 get endpoint
 344     CAEndpoint_t *rep = (CAEndpoint_t *)(data->remoteEndpoint);
 345     if (!rep)
 346     {
 347         OIC_LOG(ERROR, TAG, "remoteEndpoint error!!");
 348         return;
 349     }
 350
 351     if (data->requestInfo && g_requestHandler)
 352     {
 353         g_requestHandler(rep, data->requestInfo);
 354     }
 355     else if (data->responseInfo && g_responseHandler)
 356     {
 357         g_responseHandler(rep, data->responseInfo);
 358     }
 359     else if (data->errorInfo && g_errorHandler)
 360     {
 361         g_errorHandler(rep, data->errorInfo);
 362     }
 363
 364 #ifdef SINGLE_THREAD
 365     CADestroyData(data, sizeof(CAData_t));
 366 #endif
 367
 368     OIC_LOG(DEBUG, TAG, "CAProcessReceivedData OUT");
 369 }
 370 #endif
 371
 372 #ifndef SINGLE_THREAD
 373
 374 static void CAReceiveThreadProcess(void *threadData)
 375 {
 376     OIC_LOG(DEBUG, TAG, "IN");
 377 #ifndef SINGLE_HANDLE
 378     CAData_t *data = (CAData_t *) threadData;
 379     CAProcessReceivedData(data);
 380 #endif
 381     OIC_LOG(DEBUG, TAG, "OUT");
 382 }
 383 #endif
 384
 385 static void CAProcessSendData(const CAData_t *data)
 386 {
 387     OIC_LOG(DEBUG, TAG, "IN");
 388     VERIFY_NON_NULL_VOID(data, TAG, "data");
 389     VERIFY_NON_NULL_VOID(data->remoteEndpoint, TAG, "remoteEndpoint");
 390
 391     CAResult_t res = CA_STATUS_FAILED;
 392
 393     CASendDataType_t type = data->type;
 394
 395     coap_pdu_t *pdu = NULL;
 396
 397     if (SEND_TYPE_UNICAST == type)
 398     {
 399         CAInfo_t *info = NULL;
 400
 401         OIC_LOG(DEBUG,TAG,"Unicast message");
 402         if (NULL != data->requestInfo)
 403         {
 404             OIC_LOG(DEBUG, TAG, "requestInfo is available..");
 405
 406             info = &data->requestInfo->info;
 407             pdu = CAGeneratePDU(data->requestInfo->method, info);
 408         }
 409         else if (NULL != data->responseInfo)
 410         {
 411             OIC_LOG(DEBUG, TAG, "responseInfo is available..");
 412
 413             info = &data->responseInfo->info;
 414             pdu = CAGeneratePDU(data->responseInfo->result, info);
 415         }
 416         else
 417         {
 418             OIC_LOG(DEBUG, TAG, "request info, response info is empty");
 419             return;
 420         }
 421
 422         // interface controller function call.
 423         if (NULL != pdu)
 424         {
 425 #ifdef WITH_BWT
 426             if (CA_ADAPTER_GATT_BTLE != data->remoteEndpoint->adapter)
 427             {
 428                 // Blockwise transfer
 429                 if (NULL != info)
 430                 {
 431                     CAResult_t res = CAAddBlockOption(&pdu, *info);
 432                     if (CA_STATUS_OK != res)
 433                     {
 434                         OIC_LOG(INFO, TAG, "to write block option has failed");
 435                         CAErrorHandler(data->remoteEndpoint, pdu->hdr, pdu->length, res);
 436                         coap_delete_pdu(pdu);
 437                         return;
 438                     }
 439                 }
 440             }
 441 #endif
 442             CALogPDUInfo(pdu);
 443
 444             res = CASendUnicastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 445             if (CA_STATUS_OK != res)
 446             {
 447                 OIC_LOG_V(ERROR, TAG, "send failed:%d", res);
 448                 coap_delete_pdu(pdu);
 449                 return;
 450             }
 451             // for retransmission
 452             res = CARetransmissionSentData(&g_retransmissionContext, data->remoteEndpoint, pdu->hdr,
 453                                            pdu->length);
 454             if (CA_STATUS_OK != res)
 455             {
 456                 OIC_LOG_V(INFO, TAG, "retransmission is not enabled due to error, res : %d", res);
 457                 coap_delete_pdu(pdu);
 458                 return;
 459             }
 460
 461             coap_delete_pdu(pdu);
 462         }
 463         else
 464         {
 465             OIC_LOG(ERROR,TAG,"Failed to generate unicast PDU");
 466             return;
 467         }
 468     }
 469     else if (SEND_TYPE_MULTICAST == type)
 470     {
 471         OIC_LOG(DEBUG,TAG,"Multicast message");
 472         if (NULL != data->requestInfo)
 473         {
 474             OIC_LOG(DEBUG, TAG, "requestInfo is available..");
 475             CAInfo_t *info = &data->requestInfo->info;
 476
 477             pdu = CAGeneratePDU(CA_GET, info);
 478             if (NULL != pdu)
 479             {
 480 #ifdef WITH_BWT
 481                 if (CA_ADAPTER_GATT_BTLE != data->remoteEndpoint->adapter)
 482                 {
 483                     // Blockwise transfer
 484                     CAResult_t res = CAAddBlockOption(&pdu,
 485                                                       data->requestInfo->info);
 486                     if (CA_STATUS_OK != res)
 487                     {
 488                         OIC_LOG(DEBUG, TAG, "CAAddBlockOption has failed");
 489                         CAErrorHandler(data->remoteEndpoint, pdu->hdr, pdu->length, res);
 490                         coap_delete_pdu(pdu);
 491                         return;
 492                     }
 493                 }
 494 #endif
 495                 CALogPDUInfo(pdu);
 496
 497                 res = CASendMulticastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 498                 if (CA_STATUS_OK != res)
 499                 {
 500                     OIC_LOG_V(ERROR, TAG, "send failed:%d", res);
 501                     coap_delete_pdu(pdu);
 502                     return;
 503                 }
 504
 505                 coap_delete_pdu(pdu);
 506             }
 507             else
 508             {
 509                 OIC_LOG(ERROR,TAG,"Failed to generate multicast PDU");
 510             }
 511         }
 512         else
 513         {
 514             OIC_LOG(ERROR, TAG, "request info is empty");
 515         }
 516     }
 517
 518     OIC_LOG(DEBUG, TAG, "OUT");
 519 }
 520
 521 #ifndef SINGLE_THREAD
 522 static void CASendThreadProcess(void *threadData)
 523 {
 524     CAData_t *data = (CAData_t *) threadData;
 525     CAProcessSendData(data);
 526 }
 527
 528 #endif
 529
 530 /*
 531  * If a second message arrives with the same token and the other address
 532  * family, drop it.  Typically, IPv6 beats IPv4, so the IPv4 message is dropped.
 533  * This can be made more robust (for instance, another message could arrive
 534  * in between), but it is good enough for now.
 535  */
 536 static bool CADropSecondRequest(const CAEndpoint_t *endpoint, uint16_t messageId)
 537 {
 538     if (!endpoint)
 539     {
 540         return true;
 541     }
 542     if (endpoint->adapter != CA_ADAPTER_IP)
 543     {
 544         return false;
 545     }
 546
 547     bool ret = false;
 548     CATransportFlags_t familyFlags = endpoint->flags & CA_IPFAMILY_MASK;
 549
 550     if (messageId == caglobals.ca.previousRequestMessageId)
 551     {
 552         if ((familyFlags ^ caglobals.ca.previousRequestFlags) == CA_IPFAMILY_MASK)
 553         {
 554             if (familyFlags & CA_IPV6)
 555             {
 556                 OIC_LOG(INFO, TAG, "IPv6 duplicate response ignored");
 557             }
 558             else
 559             {
 560                 OIC_LOG(INFO, TAG, "IPv4 duplicate response ignored");
 561             }
 562             ret = true;
 563         }
 564     }
 565     caglobals.ca.previousRequestFlags = familyFlags;
 566     caglobals.ca.previousRequestMessageId = messageId;
 567     return ret;
 568 }
 569
 570 static void CAReceivedPacketCallback(const CAEndpoint_t *remoteEndpoint, const void *data, uint32_t dataLen)
 571 {
 572     OIC_LOG(DEBUG, TAG, "IN");
 573     VERIFY_NON_NULL_VOID(remoteEndpoint, TAG, "remoteEndpoint");
 574     VERIFY_NON_NULL_VOID(data, TAG, "data");
 575
 576     uint32_t code = CA_NOT_FOUND;
 577     CAData_t *cadata = NULL;
 578
 579     coap_pdu_t *pdu = (coap_pdu_t *) CAParsePDU((const char *) data, dataLen, &code);
 580     if (NULL == pdu)
 581     {
 582         OIC_LOG(ERROR, TAG, "Parse PDU failed");
 583         return;
 584     }
 585
 586     OIC_LOG_V(DEBUG, TAG, "code = %d", code);
 587     if (CA_GET == code || CA_POST == code || CA_PUT == code || CA_DELETE == code)
 588     {
 589         cadata = CAGenerateHandlerData(remoteEndpoint, pdu, CA_REQUEST_DATA);
 590         if (!cadata)
 591         {
 592             OIC_LOG(ERROR, TAG, "CAReceivedPacketCallback, CAGenerateHandlerData failed!");
 593             coap_delete_pdu(pdu);
 594             return;
 595         }
 596     }
 597     else
 598     {
 599         cadata = CAGenerateHandlerData(remoteEndpoint, pdu, CA_RESPONSE_DATA);
 600         if (!cadata)
 601         {
 602             OIC_LOG(ERROR, TAG, "CAReceivedPacketCallback, CAGenerateHandlerData failed!");
 603             coap_delete_pdu(pdu);
 604             return;
 605         }
 606
 607         // for retransmission
 608         void *retransmissionPdu = NULL;
 609         CARetransmissionReceivedData(&g_retransmissionContext, cadata->remoteEndpoint, pdu->hdr,
 610                                      pdu->length, &retransmissionPdu);
 611
 612         // get token from saved data in retransmission list
 613         if (retransmissionPdu && CA_EMPTY == code)
 614         {
 615             if (cadata->responseInfo)
 616             {
 617                 CAInfo_t *info = &cadata->responseInfo->info;
 618                 CAResult_t res = CAGetTokenFromPDU((const coap_hdr_t *)retransmissionPdu,
 619                                                    info);
 620                 if (CA_STATUS_OK != res)
 621                 {
 622                     OIC_LOG(ERROR, TAG, "fail to get Token from retransmission list");
 623                     OICFree(info->token);
 624                     info->tokenLength = 0;
 625                 }
 626             }
 627         }
 628         OICFree(retransmissionPdu);
 629     }
 630
 631     cadata->type = SEND_TYPE_UNICAST;
 632
 633 #ifdef SINGLE_THREAD
 634     CAProcessReceivedData(cadata);
 635 #else
 636 #ifdef WITH_BWT
 637         if (CA_ADAPTER_GATT_BTLE != remoteEndpoint->adapter)
 638         {
 639             CAResult_t res = CAReceiveBlockWiseData(pdu, remoteEndpoint, cadata, dataLen);
 640             if (CA_NOT_SUPPORTED == res)
 641             {
 642                 OIC_LOG(ERROR, TAG, "this message does not have block option");
 643                 CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 644             }
 645             else
 646             {
 647                 CADestroyData(cadata, sizeof(CAData_t));
 648             }
 649         }
 650         else
 651 #endif
 652         {
 653             CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 654         }
 655 #endif
 656
 657     coap_delete_pdu(pdu);
 658
 659     OIC_LOG(DEBUG, TAG, "OUT");
 660 }
 661
 662 static void CANetworkChangedCallback(const CAEndpoint_t *info, CANetworkStatus_t status)
 663 {
 664     OIC_LOG(DEBUG, TAG, "IN");
 665
 666     OIC_LOG(DEBUG, TAG, "OUT");
 667 }
 668
 669 void CAHandleRequestResponseCallbacks()
 670 {
 671 #ifdef SINGLE_THREAD
 672     CAReadData();
 673     CARetransmissionBaseRoutine((void *)&g_retransmissionContext);
 674 #else
 675 #ifdef SINGLE_HANDLE
 676     // parse the data and call the callbacks.
 677     // #1 parse the data
 678     // #2 get endpoint
 679
 680     ca_mutex_lock(g_receiveThread.threadMutex);
 681
 682     u_queue_message_t *item = u_queue_get_element(g_receiveThread.dataQueue);
 683
 684     ca_mutex_unlock(g_receiveThread.threadMutex);
 685
 686     if (NULL == item)
 687     {
 688         return;
 689     }
 690
 691     // get values
 692     void *msg = item->msg;
 693
 694     if (NULL == msg)
 695     {
 696         return;
 697     }
 698
 699     // get endpoint
 700     CAData_t *td = (CAData_t *) msg;
 701
 702     if (td->requestInfo && g_requestHandler)
 703     {
 704         OIC_LOG_V(DEBUG, TAG, "request callback : %d", td->requestInfo->info.numOptions);
 705         g_requestHandler(td->remoteEndpoint, td->requestInfo);
 706     }
 707     else if (td->responseInfo && g_responseHandler)
 708     {
 709         OIC_LOG_V(DEBUG, TAG, "response callback : %d", td->responseInfo->info.numOptions);
 710         g_responseHandler(td->remoteEndpoint, td->responseInfo);
 711     }
 712     else if (td->errorInfo && g_errorHandler)
 713     {
 714         OIC_LOG_V(DEBUG, TAG, "error callback error: %d", td->errorInfo->result);
 715         g_errorHandler(td->remoteEndpoint, td->errorInfo);
 716     }
 717
 718     CADestroyData(msg, sizeof(CAData_t));
 719     OICFree(item);
 720
 721 #endif /* SINGLE_HANDLE */
 722 #endif
 723 }
 724
 725 static CAData_t* CAPrepareSendData(const CAEndpoint_t *endpoint, const void *sendData,
 726                                    CADataType_t dataType)
 727 {
 728     OIC_LOG(DEBUG, TAG, "CAPrepareSendData IN");
 729     CAInfo_t *info = NULL;
 730
 731     CAData_t *cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 732     if (!cadata)
 733     {
 734         OIC_LOG(ERROR, TAG, "memory allocation failed");
 735         return NULL;
 736     }
 737
 738     if(CA_REQUEST_DATA == dataType)
 739     {
 740         // clone request info
 741         CARequestInfo_t *request = CACloneRequestInfo((CARequestInfo_t *)sendData);
 742
 743         if(!request)
 744         {
 745             OIC_LOG(ERROR, TAG, "CACloneRequestInfo failed");
 746             OICFree(cadata);
 747             return NULL;
 748         }
 749
 750         cadata->type = request->isMulticast ? SEND_TYPE_MULTICAST : SEND_TYPE_UNICAST;
 751         info = &request->info;
 752         cadata->requestInfo =  request;
 753     }
 754     else if(CA_RESPONSE_DATA == dataType)
 755     {
 756         // clone response info
 757         CAResponseInfo_t *response = CACloneResponseInfo((CAResponseInfo_t *)sendData);
 758
 759         if(!response)
 760         {
 761             OIC_LOG(ERROR, TAG, "CACloneResponseInfo failed");
 762             OICFree(cadata);
 763             return NULL;
 764         }
 765
 766         cadata->type = SEND_TYPE_UNICAST;
 767         info = &response->info;
 768         cadata->responseInfo = response;
 769     }
 770
 771     if (NULL != info->options && 0 < info->numOptions)
 772     {
 773         uint8_t numOptions = info->numOptions;
 774         // copy data
 775         CAHeaderOption_t *headerOption = (CAHeaderOption_t *) OICMalloc(sizeof(CAHeaderOption_t)
 776                                                                         * numOptions);
 777         if(!headerOption)
 778         {
 779             OIC_LOG(ERROR, TAG, "memory allocation failed");
 780             CADestroyData(cadata, sizeof(CAData_t));
 781             return NULL;
 782         }
 783
 784         memcpy(headerOption, info->options, sizeof(CAHeaderOption_t) * numOptions);
 785
 786         cadata->options = headerOption;
 787         cadata->numOptions = numOptions;
 788     }
 789
 790     CAEndpoint_t* ep = CACloneEndpoint(endpoint);
 791     if (!ep)
 792     {
 793         OIC_LOG(ERROR, TAG, "endpoint clone failed");
 794         CADestroyData(cadata, sizeof(CAData_t));
 795         return NULL;
 796     }
 797
 798     cadata->remoteEndpoint = ep;
 799     return cadata;
 800 }
 801
 802 CAResult_t CADetachRequestMessage(const CAEndpoint_t *object, const CARequestInfo_t *request)
 803 {
 804     OIC_LOG(DEBUG, TAG, "IN");
 805
 806     VERIFY_NON_NULL(object, TAG, "object");
 807     VERIFY_NON_NULL(request, TAG, "request");
 808
 809     if (false == CAIsSelectedNetworkAvailable())
 810     {
 811         return CA_STATUS_FAILED;
 812     }
 813
 814 #ifdef ARDUINO
 815     // If max retransmission queue is reached, then don't handle new request
 816     if (CA_MAX_RT_ARRAY_SIZE == u_arraylist_length(g_retransmissionContext.dataList))
 817     {
 818         OIC_LOG(ERROR, TAG, "max RT queue size reached!");
 819         return CA_SEND_FAILED;
 820     }
 821 #endif /* ARDUINO */
 822
 823     CAData_t *data = CAPrepareSendData(object, request, CA_REQUEST_DATA);
 824     if(!data)
 825     {
 826         OIC_LOG(ERROR, TAG, "CAPrepareSendData failed");
 827         return CA_MEMORY_ALLOC_FAILED;
 828     }
 829
 830 #ifdef SINGLE_THREAD
 831     CAProcessSendData(data);
 832     CADestroyData(data, sizeof(CAData_t));
 833 #else
 834 #ifdef WITH_BWT
 835     if (CA_ADAPTER_GATT_BTLE != object->adapter)
 836     {
 837         // send block data
 838         CAResult_t res = CASendBlockWiseData(data);
 839         if(CA_NOT_SUPPORTED == res)
 840         {
 841             OIC_LOG(DEBUG, TAG, "normal msg will be sent");
 842             CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 843             return CA_STATUS_OK;
 844         }
 845         else
 846         {
 847             CADestroyData(data, sizeof(CAData_t));
 848         }
 849         return res;
 850     }
 851     else
 852 #endif
 853     {
 854         CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 855     }
 856 #endif
 857
 858     OIC_LOG(DEBUG, TAG, "OUT");
 859     return CA_STATUS_OK;
 860 }
 861
 862 CAResult_t CADetachResponseMessage(const CAEndpoint_t *object,
 863                                    const CAResponseInfo_t *response)
 864 {
 865     OIC_LOG(DEBUG, TAG, "IN");
 866     VERIFY_NON_NULL(object, TAG, "object");
 867     VERIFY_NON_NULL(response, TAG, "response");
 868
 869     if (false == CAIsSelectedNetworkAvailable())
 870     {
 871         return CA_STATUS_FAILED;
 872     }
 873
 874     CAData_t *data = CAPrepareSendData(object, response, CA_RESPONSE_DATA);
 875     if(!data)
 876     {
 877         OIC_LOG(ERROR, TAG, "CAPrepareSendData failed");
 878         return CA_MEMORY_ALLOC_FAILED;
 879     }
 880
 881 #ifdef SINGLE_THREAD
 882     CAProcessSendData(data);
 883     CADestroyData(data, sizeof(CAData_t));
 884 #else
 885 #ifdef WITH_BWT
 886     if (CA_ADAPTER_GATT_BTLE != object->adapter)
 887     {
 888         // send block data
 889         CAResult_t res = CASendBlockWiseData(data);
 890         if(CA_NOT_SUPPORTED == res)
 891         {
 892             OIC_LOG(DEBUG, TAG, "normal msg will be sent");
 893             CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 894             return CA_STATUS_OK;
 895         }
 896         else
 897         {
 898             CADestroyData(data, sizeof(CAData_t));
 899         }
 900         return res;
 901     }
 902     else
 903 #endif
 904     {
 905         CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 906     }
 907 #endif
 908
 909     OIC_LOG(DEBUG, TAG, "OUT");
 910     return CA_STATUS_OK;
 911 }
 912
 913 CAResult_t CADetachMessageResourceUri(const CAURI_t resourceUri, const CAToken_t token,
 914                                       uint8_t tokenLength, const CAHeaderOption_t *options,
 915                                       uint8_t numOptions)
 916 {
 917     return CA_NOT_SUPPORTED;
 918 }
 919
 920 void CASetInterfaceCallbacks(CARequestCallback ReqHandler, CAResponseCallback RespHandler,
 921                              CAErrorCallback errorHandler)
 922 {
 923     OIC_LOG(DEBUG, TAG, "IN");
 924     g_requestHandler = ReqHandler;
 925     g_responseHandler = RespHandler;
 926     g_errorHandler = errorHandler;
 927     OIC_LOG(DEBUG, TAG, "OUT");
 928 }
 929
 930 CAResult_t CAInitializeMessageHandler()
 931 {
 932     OIC_LOG(DEBUG, TAG, "IN");
 933     CASetPacketReceivedCallback(CAReceivedPacketCallback);
 934
 935     CASetNetworkChangeCallback(CANetworkChangedCallback);
 936     CASetErrorHandleCallback(CAErrorHandler);
 937
 938 #ifndef SINGLE_THREAD
 939     // create thread pool
 940     CAResult_t res = ca_thread_pool_init(MAX_THREAD_POOL_SIZE, &g_threadPoolHandle);
 941
 942     if (CA_STATUS_OK != res)
 943     {
 944         OIC_LOG(ERROR, TAG, "thread pool initialize error.");
 945         return res;
 946     }
 947
 948     // send thread initialize
 949     if (CA_STATUS_OK != CAQueueingThreadInitialize(&g_sendThread, g_threadPoolHandle,
 950                                                    CASendThreadProcess, CADestroyData))
 951     {
 952         OIC_LOG(ERROR, TAG, "Failed to Initialize send queue thread");
 953         return CA_STATUS_FAILED;
 954     }
 955
 956     // start send thread
 957     res = CAQueueingThreadStart(&g_sendThread);
 958
 959     if (CA_STATUS_OK != res)
 960     {
 961         OIC_LOG(ERROR, TAG, "thread start error(send thread).");
 962         ca_thread_pool_free(g_threadPoolHandle);
 963         g_threadPoolHandle = NULL;
 964         return res;
 965     }
 966
 967     // receive thread initialize
 968     if (CA_STATUS_OK != CAQueueingThreadInitialize(&g_receiveThread, g_threadPoolHandle,
 969                                                    CAReceiveThreadProcess, CADestroyData))
 970     {
 971         OIC_LOG(ERROR, TAG, "Failed to Initialize receive queue thread");
 972         return CA_STATUS_FAILED;
 973     }
 974
 975 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
 976     // start receive thread
 977     res = CAQueueingThreadStart(&gReceiveThread);
 978
 979     if (res != CA_STATUS_OK)
 980     {
 981         OIC_LOG(ERROR, TAG, "thread start error(receive thread).");
 982         return res;
 983     }
 984 #endif /* SINGLE_HANDLE */
 985
 986     // retransmission initialize
 987     CARetransmissionInitialize(&g_retransmissionContext, g_threadPoolHandle, CASendUnicastData,
 988                                CATimeoutCallback, NULL);
 989
 990 #ifdef WITH_BWT
 991     // block-wise transfer initialize
 992     CAInitializeBlockWiseTransfer(CAAddDataToSendThread, CAAddDataToReceiveThread);
 993 #endif
 994
 995     // start retransmission
 996     res = CARetransmissionStart(&g_retransmissionContext);
 997
 998     if (CA_STATUS_OK != res)
 999     {
1000         OIC_LOG(ERROR, TAG, "thread start error(retransmission thread).");
1001         return res;
1002     }
1003
1004     // initialize interface adapters by controller
1005     CAInitializeAdapters(g_threadPoolHandle);
1006 #else
1007     // retransmission initialize
1008     CARetransmissionInitialize(&g_retransmissionContext, NULL, CASendUnicastData,
1009                                CATimeoutCallback, NULL);
1010     CAInitializeAdapters();
1011 #endif
1012
1013     OIC_LOG(DEBUG, TAG, "OUT");
1014     return CA_STATUS_OK;
1015 }
1016
1017 void CATerminateMessageHandler()
1018 {
1019     OIC_LOG(DEBUG, TAG, "IN");
1020 #ifndef SINGLE_THREAD
1021     CATransportAdapter_t connType;
1022     u_arraylist_t *list = CAGetSelectedNetworkList();
1023     uint32_t length = u_arraylist_length(list);
1024
1025     uint32_t i = 0;
1026     for (i = 0; i < length; i++)
1027     {
1028         void* ptrType = u_arraylist_get(list, i);
1029
1030         if (NULL == ptrType)
1031         {
1032             continue;
1033         }
1034
1035         connType = *(CATransportAdapter_t *)ptrType;
1036         CAStopAdapter(connType);
1037     }
1038
1039     // stop retransmission
1040     if (NULL != g_retransmissionContext.threadMutex)
1041     {
1042         CARetransmissionStop(&g_retransmissionContext);
1043     }
1044
1045     // stop thread
1046     // delete thread data
1047     if (NULL != g_sendThread.threadMutex)
1048     {
1049         CAQueueingThreadStop(&g_sendThread);
1050     }
1051
1052     // stop thread
1053     // delete thread data
1054     if (NULL != g_receiveThread.threadMutex)
1055     {
1056 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
1057         CAQueueingThreadStop(&gReceiveThread);
1058 #endif /* SINGLE_HANDLE */
1059     }
1060
1061     // destroy thread pool
1062     if (NULL != g_threadPoolHandle)
1063     {
1064         ca_thread_pool_free(g_threadPoolHandle);
1065         g_threadPoolHandle = NULL;
1066     }
1067
1068 #ifdef WITH_BWT
1069     CATerminateBlockWiseTransfer();
1070 #endif
1071     CARetransmissionDestroy(&g_retransmissionContext);
1072     CAQueueingThreadDestroy(&g_sendThread);
1073     CAQueueingThreadDestroy(&g_receiveThread);
1074
1075     // terminate interface adapters by controller
1076     CATerminateAdapters();
1077 #else
1078     // terminate interface adapters by controller
1079     CATerminateAdapters();
1080
1081     // stop retransmission
1082     CARetransmissionStop(&g_retransmissionContext);
1083     CARetransmissionDestroy(&g_retransmissionContext);
1084 #endif
1085
1086     OIC_LOG(DEBUG, TAG, "OUT");
1087 }
1088
1089 void CALogPDUInfo(coap_pdu_t *pdu)
1090 {
1091     VERIFY_NON_NULL_VOID(pdu, TAG, "pdu");
1092
1093     OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
1094
1095     OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
1096
1097     OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
1098
1099     OIC_LOG(DEBUG, TAG, "PDU Maker - token :");
1100
1101     OIC_LOG_BUFFER(DEBUG, TAG, pdu->hdr->token, pdu->hdr->token_length);
1102 }
1103
1104 static void CALogPayloadInfo(CAInfo_t *info)
1105 {
1106     if(info)
1107     {
1108         if (!info->options)
1109         {
1110             for (uint32_t i = 0; i < info->numOptions; i++)
1111             {
1112                 OIC_LOG_V(DEBUG, TAG, "optionID: %d", info->options[i].optionID);
1113
1114                 OIC_LOG_V(DEBUG, TAG, "list: %s", info->options[i].optionData);
1115             }
1116         }
1117
1118         if (!info->payload)
1119         {
1120             OIC_LOG_V(DEBUG, TAG, "payload: %p(%u)", info->payload,
1121                       info->payloadSize);
1122         }
1123
1124         if (!info->token)
1125         {
1126             OIC_LOG(DEBUG, TAG, "token:");
1127             OIC_LOG_BUFFER(DEBUG, TAG, (const uint8_t *) info->token,
1128                            info->tokenLength);
1129         }
1130         OIC_LOG_V(DEBUG, TAG, "msgID: %d", info->messageId);
1131     }
1132     else
1133     {
1134         OIC_LOG(DEBUG, TAG, "info is NULL, cannot output log data");
1135     }
1136 }
1137
1138 void CAErrorHandler(const CAEndpoint_t *endpoint,
1139                     const void *data, uint32_t dataLen,
1140                     CAResult_t result)
1141 {
1142     OIC_LOG(DEBUG, TAG, "IN");
1143
1144 #ifndef SINGLE_THREAD
1145
1146     VERIFY_NON_NULL_VOID(endpoint, TAG, "remoteEndpoint");
1147     VERIFY_NON_NULL_VOID(data, TAG, "data");
1148
1149     uint32_t code = CA_NOT_FOUND;
1150     //Do not free remoteEndpoint and data. Currently they will be freed in data thread
1151     //Get PDU data
1152     coap_pdu_t *pdu = (coap_pdu_t *)CAParsePDU((const char *)data, dataLen, &code);
1153     if (NULL == pdu)
1154     {
1155         OIC_LOG(ERROR, TAG, "Parse PDU failed");
1156         return;
1157     }
1158
1159     CAData_t *cadata = CAGenerateHandlerData(endpoint, pdu, CA_ERROR_DATA);
1160     if(!cadata)
1161     {
1162         OIC_LOG(ERROR, TAG, "CAErrorHandler, CAGenerateHandlerData failed!");
1163         coap_delete_pdu(pdu);
1164         return;
1165     }
1166
1167     cadata->errorInfo->result = result;
1168
1169     CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
1170     coap_delete_pdu(pdu);
1171 #endif
1172
1173     OIC_LOG(DEBUG, TAG, "OUT");
1174     return;
1175 }