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