resource/csdk/connectivity/src/camessagehandler.c

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