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
 400         OIC_LOG(DEBUG,TAG,"Unicast message");
 401         if (NULL != data->requestInfo)
 402         {
 403             OIC_LOG(DEBUG, TAG, "requestInfo is available..");
 404
 405             pdu = CAGeneratePDU(data->requestInfo->method, &data->requestInfo->info);
 406
 407 #ifdef WITH_BWT
 408             if (CA_ADAPTER_GATT_BTLE != data->remoteEndpoint->adapter)
 409             {
 410                 // Blockwise transfer
 411                 CAResult_t res = CAAddBlockOption(&pdu,
 412                                                   data->requestInfo->info);
 413                 if (CA_STATUS_OK != res)
 414                 {
 415                     OIC_LOG(INFO, TAG, "to write block option has failed");
 416                     CAErrorHandler(data->remoteEndpoint, pdu->hdr, pdu->length, res);
 417                     coap_delete_pdu(pdu);
 418                     return;
 419                 }
 420             }
 421 #endif
 422         }
 423         else if (NULL != data->responseInfo)
 424         {
 425             OIC_LOG(DEBUG, TAG, "responseInfo is available..");
 426
 427             pdu = CAGeneratePDU(data->responseInfo->result, &data->responseInfo->info);
 428
 429 #ifdef WITH_BWT
 430             if (CA_ADAPTER_GATT_BTLE != data->remoteEndpoint->adapter)
 431             {
 432                 // Blockwise transfer
 433                 CAResult_t res = CAAddBlockOption(&pdu,
 434                                                   data->responseInfo->info);
 435                 if (CA_STATUS_OK != res)
 436                 {
 437                     OIC_LOG(INFO, TAG, "to write block option has failed");
 438                     CAErrorHandler(data->remoteEndpoint, pdu->hdr, pdu->length, res);
 439                     coap_delete_pdu(pdu);
 440                     return;
 441                 }
 442             }
 443 #endif
 444         }
 445         else
 446         {
 447             OIC_LOG(DEBUG, TAG, "request info, response info is empty");
 448             return;
 449         }
 450
 451         // interface controller function call.
 452         if (NULL != pdu)
 453         {
 454             CALogPDUInfo(pdu);
 455
 456             res = CASendUnicastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 457             if (CA_STATUS_OK != res)
 458             {
 459                 OIC_LOG_V(ERROR, TAG, "send failed:%d", res);
 460                 coap_delete_pdu(pdu);
 461                 return;
 462             }
 463             // for retransmission
 464             res = CARetransmissionSentData(&g_retransmissionContext, data->remoteEndpoint, pdu->hdr,
 465                                            pdu->length);
 466             if (CA_STATUS_OK != res)
 467             {
 468                 OIC_LOG_V(INFO, TAG, "retransmission is not enabled due to error, res : %d", res);
 469                 coap_delete_pdu(pdu);
 470                 return;
 471             }
 472
 473             coap_delete_pdu(pdu);
 474         }
 475         else
 476         {
 477             OIC_LOG(ERROR,TAG,"Failed to generate unicast PDU");
 478             return;
 479         }
 480     }
 481     else if (SEND_TYPE_MULTICAST == type)
 482     {
 483         OIC_LOG(DEBUG,TAG,"Multicast message");
 484         if (NULL != data->requestInfo)
 485         {
 486             OIC_LOG(DEBUG, TAG, "requestInfo is available..");
 487             CAInfo_t *info = &data->requestInfo->info;
 488
 489             pdu = CAGeneratePDU(CA_GET, info);
 490             if (NULL != pdu)
 491             {
 492 #ifdef WITH_BWT
 493                 if (CA_ADAPTER_GATT_BTLE != data->remoteEndpoint->adapter)
 494                 {
 495                     // Blockwise transfer
 496                     CAResult_t res = CAAddBlockOption(&pdu,
 497                                                       data->requestInfo->info);
 498                     if (CA_STATUS_OK != res)
 499                     {
 500                         OIC_LOG(DEBUG, TAG, "CAAddBlockOption has failed");
 501                         CAErrorHandler(data->remoteEndpoint, pdu->hdr, pdu->length, res);
 502                         coap_delete_pdu(pdu);
 503                         return;
 504                     }
 505                 }
 506 #endif
 507                 CALogPDUInfo(pdu);
 508
 509                 res = CASendMulticastData(data->remoteEndpoint, pdu->hdr, pdu->length);
 510                 if (CA_STATUS_OK != res)
 511                 {
 512                     OIC_LOG_V(ERROR, TAG, "send failed:%d", res);
 513                     coap_delete_pdu(pdu);
 514                     return;
 515                 }
 516
 517                 coap_delete_pdu(pdu);
 518             }
 519             else
 520             {
 521                 OIC_LOG(ERROR,TAG,"Failed to generate multicast PDU");
 522             }
 523         }
 524         else
 525         {
 526             OIC_LOG(ERROR, TAG, "request info is empty");
 527         }
 528     }
 529
 530     OIC_LOG(DEBUG, TAG, "OUT");
 531 }
 532
 533 #ifndef SINGLE_THREAD
 534 static void CASendThreadProcess(void *threadData)
 535 {
 536     CAData_t *data = (CAData_t *) threadData;
 537     CAProcessSendData(data);
 538 }
 539
 540 #endif
 541
 542 /*
 543  * If a second message arrives with the same token and the other address
 544  * family, drop it.  Typically, IPv6 beats IPv4, so the IPv4 message is dropped.
 545  * This can be made more robust (for instance, another message could arrive
 546  * in between), but it is good enough for now.
 547  */
 548 static bool CADropSecondRequest(const CAEndpoint_t *endpoint, uint16_t messageId)
 549 {
 550     if (!endpoint)
 551     {
 552         return true;
 553     }
 554     if (endpoint->adapter != CA_ADAPTER_IP)
 555     {
 556         return false;
 557     }
 558
 559     bool ret = false;
 560     CATransportFlags_t familyFlags = endpoint->flags & CA_IPFAMILY_MASK;
 561
 562     if (messageId == caglobals.ca.previousRequestMessageId)
 563     {
 564         if ((familyFlags ^ caglobals.ca.previousRequestFlags) == CA_IPFAMILY_MASK)
 565         {
 566             if (familyFlags & CA_IPV6)
 567             {
 568                 OIC_LOG(INFO, TAG, "IPv6 duplicate response ignored");
 569             }
 570             else
 571             {
 572                 OIC_LOG(INFO, TAG, "IPv4 duplicate response ignored");
 573             }
 574             ret = true;
 575         }
 576     }
 577     caglobals.ca.previousRequestFlags = familyFlags;
 578     caglobals.ca.previousRequestMessageId = messageId;
 579     return ret;
 580 }
 581
 582 static void CAReceivedPacketCallback(const CAEndpoint_t *remoteEndpoint, const void *data, uint32_t dataLen)
 583 {
 584     OIC_LOG(DEBUG, TAG, "IN");
 585     VERIFY_NON_NULL_VOID(remoteEndpoint, TAG, "remoteEndpoint");
 586     VERIFY_NON_NULL_VOID(data, TAG, "data");
 587
 588     uint32_t code = CA_NOT_FOUND;
 589     CAData_t *cadata = NULL;
 590
 591     coap_pdu_t *pdu = (coap_pdu_t *) CAParsePDU((const char *) data, dataLen, &code);
 592     if (NULL == pdu)
 593     {
 594         OIC_LOG(ERROR, TAG, "Parse PDU failed");
 595         return;
 596     }
 597
 598     OIC_LOG_V(DEBUG, TAG, "code = %d", code);
 599     if (CA_GET == code || CA_POST == code || CA_PUT == code || CA_DELETE == code)
 600     {
 601         cadata = CAGenerateHandlerData(remoteEndpoint, pdu, CA_REQUEST_DATA);
 602         if (!cadata)
 603         {
 604             OIC_LOG(ERROR, TAG, "CAReceivedPacketCallback, CAGenerateHandlerData failed!");
 605             coap_delete_pdu(pdu);
 606             return;
 607         }
 608     }
 609     else
 610     {
 611         cadata = CAGenerateHandlerData(remoteEndpoint, pdu, CA_RESPONSE_DATA);
 612         if (!cadata)
 613         {
 614             OIC_LOG(ERROR, TAG, "CAReceivedPacketCallback, CAGenerateHandlerData failed!");
 615             coap_delete_pdu(pdu);
 616             return;
 617         }
 618
 619         // for retransmission
 620         void *retransmissionPdu = NULL;
 621         CARetransmissionReceivedData(&g_retransmissionContext, cadata->remoteEndpoint, pdu->hdr,
 622                                      pdu->length, &retransmissionPdu);
 623
 624         // get token from saved data in retransmission list
 625         if (retransmissionPdu && CA_EMPTY == code)
 626         {
 627             if (cadata->responseInfo)
 628             {
 629                 CAInfo_t *info = &cadata->responseInfo->info;
 630                 CAResult_t res = CAGetTokenFromPDU((const coap_hdr_t *)retransmissionPdu,
 631                                                    info);
 632                 if (CA_STATUS_OK != res)
 633                 {
 634                     OIC_LOG(ERROR, TAG, "fail to get Token from retransmission list");
 635                     OICFree(info->token);
 636                     info->tokenLength = 0;
 637                 }
 638             }
 639         }
 640         OICFree(retransmissionPdu);
 641     }
 642
 643     cadata->type = SEND_TYPE_UNICAST;
 644
 645 #ifdef SINGLE_THREAD
 646     CAProcessReceivedData(cadata);
 647 #else
 648 #ifdef WITH_BWT
 649         if (CA_ADAPTER_GATT_BTLE != remoteEndpoint->adapter)
 650         {
 651             CAResult_t res = CAReceiveBlockWiseData(pdu, remoteEndpoint, cadata, dataLen);
 652             if (CA_NOT_SUPPORTED == res)
 653             {
 654                 OIC_LOG(ERROR, TAG, "this message does not have block option");
 655                 CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 656             }
 657             else
 658             {
 659                 CADestroyData(cadata, sizeof(CAData_t));
 660             }
 661         }
 662         else
 663 #endif
 664         {
 665             CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
 666         }
 667 #endif
 668
 669     coap_delete_pdu(pdu);
 670
 671     OIC_LOG(DEBUG, TAG, "OUT");
 672 }
 673
 674 static void CANetworkChangedCallback(const CAEndpoint_t *info, CANetworkStatus_t status)
 675 {
 676     OIC_LOG(DEBUG, TAG, "IN");
 677
 678     OIC_LOG(DEBUG, TAG, "OUT");
 679 }
 680
 681 void CAHandleRequestResponseCallbacks()
 682 {
 683 #ifdef SINGLE_THREAD
 684     CAReadData();
 685     CARetransmissionBaseRoutine((void *)&g_retransmissionContext);
 686 #else
 687 #ifdef SINGLE_HANDLE
 688     // parse the data and call the callbacks.
 689     // #1 parse the data
 690     // #2 get endpoint
 691
 692     ca_mutex_lock(g_receiveThread.threadMutex);
 693
 694     u_queue_message_t *item = u_queue_get_element(g_receiveThread.dataQueue);
 695
 696     ca_mutex_unlock(g_receiveThread.threadMutex);
 697
 698     if (NULL == item)
 699     {
 700         return;
 701     }
 702
 703     // get values
 704     void *msg = item->msg;
 705
 706     if (NULL == msg)
 707     {
 708         return;
 709     }
 710
 711     // get endpoint
 712     CAData_t *td = (CAData_t *) msg;
 713
 714     if (td->requestInfo && g_requestHandler)
 715     {
 716         OIC_LOG_V(DEBUG, TAG, "request callback : %d", td->requestInfo->info.numOptions);
 717         g_requestHandler(td->remoteEndpoint, td->requestInfo);
 718     }
 719     else if (td->responseInfo && g_responseHandler)
 720     {
 721         OIC_LOG_V(DEBUG, TAG, "response callback : %d", td->responseInfo->info.numOptions);
 722         g_responseHandler(td->remoteEndpoint, td->responseInfo);
 723     }
 724     else if (td->errorInfo && g_errorHandler)
 725     {
 726         OIC_LOG_V(DEBUG, TAG, "error callback error: %d", td->errorInfo->result);
 727         g_errorHandler(td->remoteEndpoint, td->errorInfo);
 728     }
 729
 730     CADestroyData(msg, sizeof(CAData_t));
 731     OICFree(item);
 732
 733 #endif /* SINGLE_HANDLE */
 734 #endif
 735 }
 736
 737 static CAData_t* CAPrepareSendData(const CAEndpoint_t *endpoint, const void *sendData,
 738                                    CADataType_t dataType)
 739 {
 740     OIC_LOG(DEBUG, TAG, "CAPrepareSendData IN");
 741     CAInfo_t *info = NULL;
 742
 743     CAData_t *cadata = (CAData_t *) OICCalloc(1, sizeof(CAData_t));
 744     if (!cadata)
 745     {
 746         OIC_LOG(ERROR, TAG, "memory allocation failed");
 747         return NULL;
 748     }
 749
 750     if(CA_REQUEST_DATA == dataType)
 751     {
 752         // clone request info
 753         CARequestInfo_t *request = CACloneRequestInfo((CARequestInfo_t *)sendData);
 754
 755         if(!request)
 756         {
 757             OIC_LOG(ERROR, TAG, "CACloneRequestInfo failed");
 758             OICFree(cadata);
 759             return NULL;
 760         }
 761
 762         cadata->type = request->isMulticast ? SEND_TYPE_MULTICAST : SEND_TYPE_UNICAST;
 763         info = &request->info;
 764         cadata->requestInfo =  request;
 765     }
 766     else if(CA_RESPONSE_DATA == dataType)
 767     {
 768         // clone response info
 769         CAResponseInfo_t *response = CACloneResponseInfo((CAResponseInfo_t *)sendData);
 770
 771         if(!response)
 772         {
 773             OIC_LOG(ERROR, TAG, "CACloneResponseInfo failed");
 774             OICFree(cadata);
 775             return NULL;
 776         }
 777
 778         cadata->type = SEND_TYPE_UNICAST;
 779         info = &response->info;
 780         cadata->responseInfo = response;
 781     }
 782
 783     if (NULL != info->options && 0 < info->numOptions)
 784     {
 785         uint8_t numOptions = info->numOptions;
 786         // copy data
 787         CAHeaderOption_t *headerOption = (CAHeaderOption_t *) OICMalloc(sizeof(CAHeaderOption_t)
 788                                                                         * numOptions);
 789         if(!headerOption)
 790         {
 791             OIC_LOG(ERROR, TAG, "memory allocation failed");
 792             CADestroyData(cadata, sizeof(CAData_t));
 793             return NULL;
 794         }
 795
 796         memcpy(headerOption, info->options, sizeof(CAHeaderOption_t) * numOptions);
 797
 798         cadata->options = headerOption;
 799         cadata->numOptions = numOptions;
 800     }
 801
 802     CAEndpoint_t* ep = CACloneEndpoint(endpoint);
 803     if (!ep)
 804     {
 805         OIC_LOG(ERROR, TAG, "endpoint clone failed");
 806         CADestroyData(cadata, sizeof(CAData_t));
 807         return NULL;
 808     }
 809
 810     cadata->remoteEndpoint = ep;
 811     return cadata;
 812 }
 813
 814 CAResult_t CADetachRequestMessage(const CAEndpoint_t *object, const CARequestInfo_t *request)
 815 {
 816     OIC_LOG(DEBUG, TAG, "IN");
 817
 818     VERIFY_NON_NULL(object, TAG, "object");
 819     VERIFY_NON_NULL(request, TAG, "request");
 820
 821     if (false == CAIsSelectedNetworkAvailable())
 822     {
 823         return CA_STATUS_FAILED;
 824     }
 825
 826 #ifdef ARDUINO
 827     // If max retransmission queue is reached, then don't handle new request
 828     if (CA_MAX_RT_ARRAY_SIZE == u_arraylist_length(g_retransmissionContext.dataList))
 829     {
 830         OIC_LOG(ERROR, TAG, "max RT queue size reached!");
 831         return CA_SEND_FAILED;
 832     }
 833 #endif /* ARDUINO */
 834
 835     CAData_t *data = CAPrepareSendData(object, request, CA_REQUEST_DATA);
 836     if(!data)
 837     {
 838         OIC_LOG(ERROR, TAG, "CAPrepareSendData failed");
 839         return CA_MEMORY_ALLOC_FAILED;
 840     }
 841
 842 #ifdef SINGLE_THREAD
 843     CAProcessSendData(data);
 844     CADestroyData(data, sizeof(CAData_t));
 845 #else
 846 #ifdef WITH_BWT
 847     if (CA_ADAPTER_GATT_BTLE != object->adapter)
 848     {
 849         // send block data
 850         CAResult_t res = CASendBlockWiseData(data);
 851         if(CA_NOT_SUPPORTED == res)
 852         {
 853             OIC_LOG(DEBUG, TAG, "normal msg will be sent");
 854             CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 855             return CA_STATUS_OK;
 856         }
 857         else
 858         {
 859             CADestroyData(data, sizeof(CAData_t));
 860         }
 861         return res;
 862     }
 863     else
 864 #endif
 865     {
 866         CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 867     }
 868 #endif
 869
 870     OIC_LOG(DEBUG, TAG, "OUT");
 871     return CA_STATUS_OK;
 872 }
 873
 874 CAResult_t CADetachResponseMessage(const CAEndpoint_t *object,
 875                                    const CAResponseInfo_t *response)
 876 {
 877     OIC_LOG(DEBUG, TAG, "IN");
 878     VERIFY_NON_NULL(object, TAG, "object");
 879     VERIFY_NON_NULL(response, TAG, "response");
 880
 881     if (false == CAIsSelectedNetworkAvailable())
 882     {
 883         return CA_STATUS_FAILED;
 884     }
 885
 886     CAData_t *data = CAPrepareSendData(object, response, CA_RESPONSE_DATA);
 887     if(!data)
 888     {
 889         OIC_LOG(ERROR, TAG, "CAPrepareSendData failed");
 890         return CA_MEMORY_ALLOC_FAILED;
 891     }
 892
 893 #ifdef SINGLE_THREAD
 894     CAProcessSendData(data);
 895     CADestroyData(data, sizeof(CAData_t));
 896 #else
 897 #ifdef WITH_BWT
 898     if (CA_ADAPTER_GATT_BTLE != object->adapter)
 899     {
 900         // send block data
 901         CAResult_t res = CASendBlockWiseData(data);
 902         if(CA_NOT_SUPPORTED == res)
 903         {
 904             OIC_LOG(DEBUG, TAG, "normal msg will be sent");
 905             CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 906             return CA_STATUS_OK;
 907         }
 908         else
 909         {
 910             CADestroyData(data, sizeof(CAData_t));
 911         }
 912         return res;
 913     }
 914     else
 915 #endif
 916     {
 917         CAQueueingThreadAddData(&g_sendThread, data, sizeof(CAData_t));
 918     }
 919 #endif
 920
 921     OIC_LOG(DEBUG, TAG, "OUT");
 922     return CA_STATUS_OK;
 923 }
 924
 925 CAResult_t CADetachMessageResourceUri(const CAURI_t resourceUri, const CAToken_t token,
 926                                       uint8_t tokenLength, const CAHeaderOption_t *options,
 927                                       uint8_t numOptions)
 928 {
 929     return CA_NOT_SUPPORTED;
 930 }
 931
 932 void CASetInterfaceCallbacks(CARequestCallback ReqHandler, CAResponseCallback RespHandler,
 933                              CAErrorCallback errorHandler)
 934 {
 935     OIC_LOG(DEBUG, TAG, "IN");
 936     g_requestHandler = ReqHandler;
 937     g_responseHandler = RespHandler;
 938     g_errorHandler = errorHandler;
 939     OIC_LOG(DEBUG, TAG, "OUT");
 940 }
 941
 942 CAResult_t CAInitializeMessageHandler()
 943 {
 944     OIC_LOG(DEBUG, TAG, "IN");
 945     CASetPacketReceivedCallback(CAReceivedPacketCallback);
 946
 947     CASetNetworkChangeCallback(CANetworkChangedCallback);
 948     CASetErrorHandleCallback(CAErrorHandler);
 949
 950 #ifndef SINGLE_THREAD
 951     // create thread pool
 952     CAResult_t res = ca_thread_pool_init(MAX_THREAD_POOL_SIZE, &g_threadPoolHandle);
 953
 954     if (CA_STATUS_OK != res)
 955     {
 956         OIC_LOG(ERROR, TAG, "thread pool initialize error.");
 957         return res;
 958     }
 959
 960     // send thread initialize
 961     if (CA_STATUS_OK != CAQueueingThreadInitialize(&g_sendThread, g_threadPoolHandle,
 962                                                    CASendThreadProcess, CADestroyData))
 963     {
 964         OIC_LOG(ERROR, TAG, "Failed to Initialize send queue thread");
 965         return CA_STATUS_FAILED;
 966     }
 967
 968     // start send thread
 969     res = CAQueueingThreadStart(&g_sendThread);
 970
 971     if (CA_STATUS_OK != res)
 972     {
 973         OIC_LOG(ERROR, TAG, "thread start error(send thread).");
 974         ca_thread_pool_free(g_threadPoolHandle);
 975         g_threadPoolHandle = NULL;
 976         return res;
 977     }
 978
 979     // receive thread initialize
 980     if (CA_STATUS_OK != CAQueueingThreadInitialize(&g_receiveThread, g_threadPoolHandle,
 981                                                    CAReceiveThreadProcess, CADestroyData))
 982     {
 983         OIC_LOG(ERROR, TAG, "Failed to Initialize receive queue thread");
 984         return CA_STATUS_FAILED;
 985     }
 986
 987 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
 988     // start receive thread
 989     res = CAQueueingThreadStart(&gReceiveThread);
 990
 991     if (res != CA_STATUS_OK)
 992     {
 993         OIC_LOG(ERROR, TAG, "thread start error(receive thread).");
 994         return res;
 995     }
 996 #endif /* SINGLE_HANDLE */
 997
 998     // retransmission initialize
 999     CARetransmissionInitialize(&g_retransmissionContext, g_threadPoolHandle, CASendUnicastData,
1000                                CATimeoutCallback, NULL);
1001
1002 #ifdef WITH_BWT
1003     // block-wise transfer initialize
1004     CAInitializeBlockWiseTransfer(CAAddDataToSendThread, CAAddDataToReceiveThread);
1005 #endif
1006
1007     // start retransmission
1008     res = CARetransmissionStart(&g_retransmissionContext);
1009
1010     if (CA_STATUS_OK != res)
1011     {
1012         OIC_LOG(ERROR, TAG, "thread start error(retransmission thread).");
1013         return res;
1014     }
1015
1016     // initialize interface adapters by controller
1017     CAInitializeAdapters(g_threadPoolHandle);
1018 #else
1019     // retransmission initialize
1020     CARetransmissionInitialize(&g_retransmissionContext, NULL, CASendUnicastData,
1021                                CATimeoutCallback, NULL);
1022     CAInitializeAdapters();
1023 #endif
1024
1025     OIC_LOG(DEBUG, TAG, "OUT");
1026     return CA_STATUS_OK;
1027 }
1028
1029 void CATerminateMessageHandler()
1030 {
1031     OIC_LOG(DEBUG, TAG, "IN");
1032 #ifndef SINGLE_THREAD
1033     CATransportAdapter_t connType;
1034     u_arraylist_t *list = CAGetSelectedNetworkList();
1035     uint32_t length = u_arraylist_length(list);
1036
1037     uint32_t i = 0;
1038     for (i = 0; i < length; i++)
1039     {
1040         void* ptrType = u_arraylist_get(list, i);
1041
1042         if (NULL == ptrType)
1043         {
1044             continue;
1045         }
1046
1047         connType = *(CATransportAdapter_t *)ptrType;
1048         CAStopAdapter(connType);
1049     }
1050
1051     // stop retransmission
1052     if (NULL != g_retransmissionContext.threadMutex)
1053     {
1054         CARetransmissionStop(&g_retransmissionContext);
1055     }
1056
1057     // stop thread
1058     // delete thread data
1059     if (NULL != g_sendThread.threadMutex)
1060     {
1061         CAQueueingThreadStop(&g_sendThread);
1062     }
1063
1064     // stop thread
1065     // delete thread data
1066     if (NULL != g_receiveThread.threadMutex)
1067     {
1068 #ifndef SINGLE_HANDLE // This will be enabled when RI supports multi threading
1069         CAQueueingThreadStop(&gReceiveThread);
1070 #endif /* SINGLE_HANDLE */
1071     }
1072
1073     // destroy thread pool
1074     if (NULL != g_threadPoolHandle)
1075     {
1076         ca_thread_pool_free(g_threadPoolHandle);
1077         g_threadPoolHandle = NULL;
1078     }
1079
1080 #ifdef WITH_BWT
1081     CATerminateBlockWiseTransfer();
1082 #endif
1083     CARetransmissionDestroy(&g_retransmissionContext);
1084     CAQueueingThreadDestroy(&g_sendThread);
1085     CAQueueingThreadDestroy(&g_receiveThread);
1086
1087     // terminate interface adapters by controller
1088     CATerminateAdapters();
1089 #else
1090     // terminate interface adapters by controller
1091     CATerminateAdapters();
1092
1093     // stop retransmission
1094     CARetransmissionStop(&g_retransmissionContext);
1095     CARetransmissionDestroy(&g_retransmissionContext);
1096 #endif
1097
1098     OIC_LOG(DEBUG, TAG, "OUT");
1099 }
1100
1101 void CALogPDUInfo(coap_pdu_t *pdu)
1102 {
1103     VERIFY_NON_NULL_VOID(pdu, TAG, "pdu");
1104
1105     OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
1106
1107     OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
1108
1109     OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
1110
1111     OIC_LOG(DEBUG, TAG, "PDU Maker - token :");
1112
1113     OIC_LOG_BUFFER(DEBUG, TAG, pdu->hdr->token, pdu->hdr->token_length);
1114 }
1115
1116 static void CALogPayloadInfo(CAInfo_t *info)
1117 {
1118     if(info)
1119     {
1120         if (!info->options)
1121         {
1122             for (uint32_t i = 0; i < info->numOptions; i++)
1123             {
1124                 OIC_LOG_V(DEBUG, TAG, "optionID: %d", info->options[i].optionID);
1125
1126                 OIC_LOG_V(DEBUG, TAG, "list: %s", info->options[i].optionData);
1127             }
1128         }
1129
1130         if (!info->payload)
1131         {
1132             OIC_LOG_V(DEBUG, TAG, "payload: %p(%u)", info->payload,
1133                       info->payloadSize);
1134         }
1135
1136         if (!info->token)
1137         {
1138             OIC_LOG(DEBUG, TAG, "token:");
1139             OIC_LOG_BUFFER(DEBUG, TAG, (const uint8_t *) info->token,
1140                            info->tokenLength);
1141         }
1142         OIC_LOG_V(DEBUG, TAG, "msgID: %d", info->messageId);
1143     }
1144     else
1145     {
1146         OIC_LOG(DEBUG, TAG, "info is NULL, cannot output log data");
1147     }
1148 }
1149
1150 void CAErrorHandler(const CAEndpoint_t *endpoint,
1151                     const void *data, uint32_t dataLen,
1152                     CAResult_t result)
1153 {
1154     OIC_LOG(DEBUG, TAG, "IN");
1155
1156 #ifndef SINGLE_THREAD
1157
1158     VERIFY_NON_NULL_VOID(endpoint, TAG, "remoteEndpoint");
1159     VERIFY_NON_NULL_VOID(data, TAG, "data");
1160
1161     uint32_t code = CA_NOT_FOUND;
1162     //Do not free remoteEndpoint and data. Currently they will be freed in data thread
1163     //Get PDU data
1164     coap_pdu_t *pdu = (coap_pdu_t *)CAParsePDU((const char *)data, dataLen, &code);
1165     if (NULL == pdu)
1166     {
1167         OIC_LOG(ERROR, TAG, "Parse PDU failed");
1168         return;
1169     }
1170
1171     CAData_t *cadata = CAGenerateHandlerData(endpoint, pdu, CA_ERROR_DATA);
1172     if(!cadata)
1173     {
1174         OIC_LOG(ERROR, TAG, "CAErrorHandler, CAGenerateHandlerData failed!");
1175         coap_delete_pdu(pdu);
1176         return;
1177     }
1178
1179     cadata->errorInfo->result = result;
1180
1181     CAQueueingThreadAddData(&g_receiveThread, cadata, sizeof(CAData_t));
1182     coap_delete_pdu(pdu);
1183 #endif
1184
1185     OIC_LOG(DEBUG, TAG, "OUT");
1186     return;
1187 }