1 /******************************************************************
3 * Copyright 2014 Samsung Electronics All Rights Reserved.
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
11 * http://www.apache.org/licenses/LICENSE-2.0
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.
19 ******************************************************************/
26 #include "cainterface.h"
27 #include "camessagehandler.h"
28 #include "caremotehandler.h"
29 #include "cainterfacecontroller.h"
30 #include "caprotocolmessage.h"
33 #include "config.h" /* for coap protocol */
35 #include "uthreadpool.h" /* for thread pool */
36 #include "caqueueingthread.h"
37 #include "caretransmission.h"
39 #include "oic_malloc.h"
43 #define MEMORY_ALLOC_CHECK(arg) { if (arg == NULL) {OIC_LOG_V(DEBUG, TAG, "memory error"); \
44 goto memory_error_exit;} }
46 #define MAX_THREAD_POOL_SIZE 10
50 SEND_TYPE_MULTICAST = 0, SEND_TYPE_UNICAST
55 CASendDataType_t type;
56 CARemoteEndpoint_t *remoteEndpoint;
57 CARequestInfo_t *requestInfo;
58 CAResponseInfo_t *responseInfo;
59 CAHeaderOption_t *options;
64 static u_thread_pool_t gThreadPoolHandle = NULL;
66 // message handler main thread
67 static CAQueueingThread_t gSendThread;
68 static CAQueueingThread_t gReceiveThread;
70 static CARetransmission_t gRetransmissionContext;
73 static CARequestCallback gRequestHandler = NULL;
74 static CAResponseCallback gResponseHandler = NULL;
76 static void CATimeoutCallback(const CARemoteEndpoint_t *endpoint, void *pdu, uint32_t size)
78 CARemoteEndpoint_t* ep = CACloneRemoteEndpoint(endpoint);
81 OIC_LOG(DEBUG, TAG, "memory allocation failed !");
85 CAResponseInfo_t* resInfo = (CAResponseInfo_t*) OICMalloc(sizeof(CAResponseInfo_t));
89 OIC_LOG(DEBUG, TAG, "memory allocation failed !");
90 CADestroyRemoteEndpointInternal(ep);
93 memset(resInfo, 0, sizeof(CAResponseInfo_t));
95 CAMessageType_t type = CAGetMessageTypeFromPduBinaryData(pdu, size);
96 uint16_t messageId = CAGetMessageIdFromPduBinaryData(pdu, size);
98 resInfo->result = CA_RETRANSMIT_TIMEOUT;
99 resInfo->info.type = type;
100 resInfo->info.messageId = messageId;
102 CAData_t *cadata = (CAData_t *) OICMalloc(sizeof(CAData_t));
105 OIC_LOG(DEBUG, TAG, "memory allocation failed !");
106 CADestroyRemoteEndpointInternal(ep);
110 memset(cadata, 0, sizeof(CAData_t));
112 cadata->type = SEND_TYPE_UNICAST;
113 cadata->remoteEndpoint = ep;
114 cadata->requestInfo = NULL;
115 cadata->responseInfo = resInfo;
117 CAQueueingThreadAddData(&gReceiveThread, cadata, sizeof(CAData_t));
120 static void CAReceiveThreadProcess(void *threadData)
123 // currently not support.
124 // CAHandleRequestResponseCallbacks codes will move to this function.
127 static void CASendThreadProcess(void *threadData)
129 CAData_t *data = (CAData_t *) threadData;
133 OIC_LOG(DEBUG, TAG, "thread data error!!");
137 if (NULL == data->remoteEndpoint)
139 OIC_LOG(DEBUG, TAG, "remoteEndpoint is null");
143 CAResult_t res = CA_STATUS_FAILED;
145 CASendDataType_t type = data->type;
147 if (type == SEND_TYPE_UNICAST)
149 coap_pdu_t *pdu = NULL;
151 if (data->requestInfo != NULL)
153 OIC_LOG_V(DEBUG, TAG, "requestInfo is available..");
155 pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri,
156 data->requestInfo->method, data->requestInfo->info);
158 else if (data->responseInfo != NULL)
160 OIC_LOG_V(DEBUG, TAG, "responseInfo is available..");
162 pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri,
163 data->responseInfo->result,
164 data->responseInfo->info);
168 OIC_LOG(DEBUG, TAG, "request info, response info is empty");
171 // interface controller function call.
174 OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
176 OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
178 OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
180 OIC_LOG_V(DEBUG, TAG, "PDU Maker - id : %d", ntohs(pdu->hdr->id));
182 OIC_LOG_V(DEBUG, TAG, "PDU Maker - token : %s", pdu->hdr->token);
184 OIC_LOG_V(DEBUG, TAG, "PDU Maker - buffer data : %s", pdu->hdr);
186 res = CASendUnicastData(data->remoteEndpoint, pdu->hdr, pdu->length);
188 // for retransmission
189 CARetransmissionSentData(&gRetransmissionContext, data->remoteEndpoint, pdu->hdr,
192 //For Unicast , data will be deleted by adapters
193 CADestroyRemoteEndpointInternal(data->remoteEndpoint);
195 coap_delete_pdu(pdu);
197 else if (type == SEND_TYPE_MULTICAST)
199 OIC_LOG(DEBUG, TAG, "both requestInfo & responseInfo is not available");
201 coap_pdu_t *pdu = NULL;
203 memset(&info, 0, sizeof(CAInfo_t));
205 info.options = data->options;
206 info.numOptions = data->numOptions;
207 info.token = data->requestInfo->info.token;
208 info.type = data->requestInfo->info.type;
210 pdu = (coap_pdu_t *) CAGeneratePdu(data->remoteEndpoint->resourceUri, CA_GET, info);
214 OIC_LOG_V(DEBUG, TAG, "PDU Maker - payload : %s", pdu->data);
216 OIC_LOG_V(DEBUG, TAG, "PDU Maker - type : %d", pdu->hdr->type);
218 OIC_LOG_V(DEBUG, TAG, "PDU Maker - code : %d", pdu->hdr->code);
220 OIC_LOG_V(DEBUG, TAG, "PDU Maker - id : %d", ntohs(pdu->hdr->id));
222 OIC_LOG_V(DEBUG, TAG, "PDU Maker - token : %s", pdu->hdr->token);
224 OIC_LOG_V(DEBUG, TAG, "PDU Maker - buffer data : %s", pdu->hdr);
226 res = CASendMulticastData(pdu->hdr, pdu->length);
228 coap_delete_pdu(pdu);
231 OIC_LOG_V(DEBUG, TAG, " Result :%d", res);
234 static void CAReceivedPacketCallback(CARemoteEndpoint_t *endpoint, void *data,
237 OIC_LOG(DEBUG, TAG, "receivedPacketCallback in message handler!!");
241 OIC_LOG(DEBUG, TAG, "received data is null");
246 uint32_t code = CA_NOT_FOUND;
248 OIC_LOG_V(DEBUG, TAG, "data : %s", data);
249 pdu = (coap_pdu_t *) CAParsePDU((const char *) data, dataLen, &code);
253 char uri[CA_MAX_URI_LENGTH] =
256 if (code == CA_GET || code == CA_POST || code == CA_PUT || code == CA_DELETE)
258 CARequestInfo_t *ReqInfo;
259 ReqInfo = (CARequestInfo_t *) OICMalloc(sizeof(CARequestInfo_t));
262 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
263 coap_delete_pdu(pdu);
266 memset(ReqInfo, 0, sizeof(CARequestInfo_t));
267 CAGetRequestInfoFromPdu(pdu, ReqInfo, uri);
269 if (NULL != ReqInfo->info.options)
272 for (i = 0; i < ReqInfo->info.numOptions; i++)
274 OIC_LOG_V(DEBUG, TAG, "Request- optionID: %d", ReqInfo->info.options[i].optionID);
276 OIC_LOG_V(DEBUG, TAG, "Request- list: %s", ReqInfo->info.options[i].optionData);
280 if (NULL != ReqInfo->info.payload)
282 OIC_LOG_V(DEBUG, TAG, "Request- payload: %s", ReqInfo->info.payload);
284 OIC_LOG_V(DEBUG, TAG, "Request- code: %d", ReqInfo->method);
285 OIC_LOG_V(DEBUG, TAG, "Request- token : %s", ReqInfo->info.token);
287 if (NULL != endpoint)
289 endpoint->resourceUri = (char *) OICMalloc(strlen(uri) + 1);
290 if (endpoint->resourceUri == NULL)
292 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
294 coap_delete_pdu(pdu);
297 memset(endpoint->resourceUri, 0, strlen(uri) + 1);
298 memcpy(endpoint->resourceUri, uri, strlen(uri));
299 OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
301 // store the data at queue.
302 CAData_t *cadata = NULL;
303 cadata = (CAData_t *) OICMalloc(sizeof(CAData_t));
306 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
307 if (endpoint->resourceUri != NULL)
308 OICFree(endpoint->resourceUri);
310 coap_delete_pdu(pdu);
313 memset(cadata, 0, sizeof(CAData_t));
315 cadata->type = SEND_TYPE_UNICAST;
316 cadata->remoteEndpoint = endpoint;
317 cadata->requestInfo = ReqInfo;
318 cadata->responseInfo = NULL;
319 CAQueueingThreadAddData(&gReceiveThread, cadata, sizeof(CAData_t));
323 CAResponseInfo_t *ResInfo;
324 ResInfo = (CAResponseInfo_t *) OICMalloc(sizeof(CAResponseInfo_t));
327 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
328 coap_delete_pdu(pdu);
331 memset(ResInfo, 0, sizeof(CAResponseInfo_t));
332 CAGetResponseInfoFromPdu(pdu, ResInfo, uri);
334 if (NULL != ResInfo->info.options)
337 for (i = 0; i < ResInfo->info.numOptions; i++)
339 OIC_LOG_V(DEBUG, TAG, "Response- optionID: %d", ResInfo->info.options[i].optionID);
341 OIC_LOG_V(DEBUG, TAG, "Response- list: %s", ResInfo->info.options[i].optionData);
343 if (NULL != ResInfo->info.payload)
345 OIC_LOG_V(DEBUG, TAG, "Response- payload: %s", ResInfo->info.payload);
346 } OIC_LOG_V(DEBUG, TAG, "Response- code: %d", ResInfo->result);
349 if (NULL != endpoint)
351 endpoint->resourceUri = (char *) OICMalloc(strlen(uri) + 1);
352 if (endpoint->resourceUri == NULL)
354 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
356 coap_delete_pdu(pdu);
359 memset(endpoint->resourceUri, 0, strlen(uri) + 1);
360 memcpy(endpoint->resourceUri, uri, strlen(uri));
361 OIC_LOG_V(DEBUG, TAG, "added resource URI : %s", endpoint->resourceUri);
364 // store the data at queue.
365 CAData_t *cadata = NULL;
366 cadata = (CAData_t *) OICMalloc(sizeof(CAData_t));
369 OIC_LOG(DEBUG, TAG, "CAReceivedPacketCallback, Memory allocation failed !");
370 if (endpoint->resourceUri != NULL)
371 OICFree(endpoint->resourceUri);
373 coap_delete_pdu(pdu);
376 memset(cadata, 0, sizeof(CAData_t));
378 cadata->type = SEND_TYPE_UNICAST;
379 cadata->remoteEndpoint = endpoint;
380 cadata->requestInfo = NULL;
381 cadata->responseInfo = ResInfo;
383 CAQueueingThreadAddData(&gReceiveThread, cadata, sizeof(CAData_t));
385 // for retransmission
386 CARetransmissionReceivedData(&gRetransmissionContext, endpoint, pdu->hdr, pdu->length);
391 static void CANetworkChangedCallback(CALocalConnectivity_t *info, CANetworkStatus_t status)
393 OIC_LOG(DEBUG, TAG, "networkChangeCallback in message handler!!");
395 OIC_LOG_V(DEBUG, TAG, "Changed Network Status: %d", status);
398 void CAHandleRequestResponseCallbacks()
400 OIC_LOG_V(DEBUG, TAG, "CAHandleRequestResponseCallbacks");
402 // parse the data and call the callbacks.
406 u_mutex_lock(gReceiveThread.threadMutex);
408 u_queue_message_t *item = u_queue_get_element(gReceiveThread.dataQueue);
410 u_mutex_unlock(gReceiveThread.threadMutex);
416 void *msg = item->msg;
422 CAData_t *td = (CAData_t *) msg;
423 CARemoteEndpoint_t *rep = td->remoteEndpoint;
428 if (td->requestInfo != NULL)
432 gRequestHandler(rep, td->requestInfo);
435 if (NULL != td->requestInfo->info.options)
437 OICFree(td->requestInfo->info.options);
440 if (NULL != td->requestInfo->info.payload)
442 OICFree(td->requestInfo->info.payload);
445 if (NULL != td->requestInfo->info.token)
447 OICFree(td->requestInfo->info.token);
449 OICFree(td->requestInfo);
452 if (td->responseInfo != NULL)
454 if (gResponseHandler)
456 gResponseHandler(rep, td->responseInfo);
459 if (NULL != td->responseInfo->info.options)
461 OICFree(td->responseInfo->info.options);
464 if (NULL != td->responseInfo->info.payload)
466 OICFree(td->responseInfo->info.payload);
469 if (NULL != td->responseInfo->info.token)
471 OICFree(td->responseInfo->info.token);
473 OICFree(td->responseInfo);
476 if (NULL != rep->resourceUri)
478 OICFree(rep->resourceUri);
483 CAResult_t CADetachRequestMessage(const CARemoteEndpoint_t *object,
484 const CARequestInfo_t *request)
486 OIC_LOG_V(DEBUG, TAG, "CADetachRequestMessage");
488 if (object == NULL || request == NULL)
490 return CA_STATUS_FAILED;
493 CAData_t *data = (CAData_t *) OICMalloc(sizeof(CAData_t));
494 MEMORY_ALLOC_CHECK(data);
497 memset(data, 0, sizeof(CAData_t));
499 // clone remote endpoint
500 CARemoteEndpoint_t *remoteEndpoint = CACloneRemoteEndpoint(object);
501 MEMORY_ALLOC_CHECK(remoteEndpoint);
503 // clone request info
504 CARequestInfo_t *requestInfo = CACloneRequestInfo(request);
505 MEMORY_ALLOC_CHECK(requestInfo);
508 data->type = SEND_TYPE_UNICAST;
509 data->remoteEndpoint = remoteEndpoint;
510 data->requestInfo = requestInfo;
511 data->responseInfo = NULL;
514 CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
518 // memory error label.
521 CADestroyRemoteEndpointInternal(remoteEndpoint);
523 CADestroyRequestInfoInternal(requestInfo);
530 return CA_MEMORY_ALLOC_FAILED;
533 CAResult_t CADetachRequestToAllMessage(const CAGroupEndpoint_t *object,
534 const CARequestInfo_t *request)
537 OIC_LOG_V(DEBUG, TAG, "CADetachRequestToAllMessage");
540 if (object == NULL || request == NULL)
542 return CA_STATUS_FAILED;
545 CAData_t *data = (CAData_t *) OICMalloc(sizeof(CAData_t));
546 MEMORY_ALLOC_CHECK(data);
549 memset(data, 0, sizeof(CAData_t));
552 memset(&addr, 0, sizeof(CAAddress_t));
553 CARemoteEndpoint_t *remoteEndpoint = CACreateRemoteEndpointInternal(object->resourceUri, addr,
554 object->connectivityType);
556 // clone request info
557 CARequestInfo_t *requestInfo = CACloneRequestInfo(request);
558 MEMORY_ALLOC_CHECK(requestInfo);
561 data->type = SEND_TYPE_MULTICAST;
562 data->remoteEndpoint = remoteEndpoint;
563 data->requestInfo = requestInfo;
564 data->responseInfo = NULL;
567 CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
571 // memory error label.
574 CADestroyRemoteEndpointInternal(remoteEndpoint);
576 CADestroyRequestInfoInternal(requestInfo);
583 return CA_MEMORY_ALLOC_FAILED;
586 CAResult_t CADetachResponseMessage(const CARemoteEndpoint_t *object,
587 const CAResponseInfo_t *response)
589 OIC_LOG_V(DEBUG, TAG, "CADetachResponseMessage");
591 if (object == NULL || response == NULL)
593 return CA_STATUS_FAILED;
596 CAData_t *data = (CAData_t *) OICMalloc(sizeof(CAData_t));
597 MEMORY_ALLOC_CHECK(data);
600 memset(data, 0, sizeof(CAData_t));
602 // clone remote endpoint
603 CARemoteEndpoint_t *remoteEndpoint = CACloneRemoteEndpoint(object);
604 MEMORY_ALLOC_CHECK(remoteEndpoint);
606 // clone response info
607 CAResponseInfo_t *responseInfo = CACloneResponseInfo(response);
608 MEMORY_ALLOC_CHECK(responseInfo);
611 data->type = SEND_TYPE_UNICAST;
612 data->remoteEndpoint = remoteEndpoint;
613 data->requestInfo = NULL;
614 data->responseInfo = responseInfo;
617 CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
621 // memory error label.
624 CADestroyRemoteEndpointInternal(remoteEndpoint);
626 CADestroyResponseInfoInternal(responseInfo);
633 return CA_MEMORY_ALLOC_FAILED;
636 CAResult_t CADetachMessageResourceUri(const CAURI_t resourceUri, const CAToken_t token,
637 const CAHeaderOption_t *options, uint8_t numOptions)
639 if (resourceUri == NULL)
641 return CA_STATUS_FAILED;
643 CAData_t *data = (CAData_t *) OICMalloc(sizeof(CAData_t));
644 MEMORY_ALLOC_CHECK(data);
647 memset(data, 0, sizeof(CAData_t));
650 memset(&addr, 0, sizeof(CAAddress_t));
651 CARemoteEndpoint_t *remoteEndpoint = CACreateRemoteEndpointInternal(resourceUri, addr,
652 CA_ETHERNET | CA_WIFI | CA_EDR | CA_LE);
655 data->type = SEND_TYPE_MULTICAST;
656 data->remoteEndpoint = remoteEndpoint;
657 CARequestInfo_t *ReqInfo;
658 ReqInfo = (CARequestInfo_t *) OICMalloc(sizeof(CARequestInfo_t));
661 OIC_LOG(DEBUG, TAG, "CADetachMessageResourceUri, Memory allocation failed !");
663 return CA_MEMORY_ALLOC_FAILED;
665 memset(ReqInfo, 0, sizeof(CARequestInfo_t));
666 ReqInfo->method = CA_GET;
667 ReqInfo->info.token = token;
668 ReqInfo->info.type = CA_MSG_NONCONFIRM;
669 data->requestInfo = ReqInfo;
671 data->responseInfo = NULL;
672 data->options = NULL;
673 data->numOptions = 0;
675 if (options != NULL && numOptions > 0)
678 CAHeaderOption_t *temp = (CAHeaderOption_t *) OICMalloc(
679 sizeof(CAHeaderOption_t) * numOptions);
681 MEMORY_ALLOC_CHECK(temp);
683 memset(temp, 0, sizeof(CAHeaderOption_t) * numOptions);
684 memcpy(temp, options, sizeof(CAHeaderOption_t) * numOptions);
686 data->options = temp;
687 data->numOptions = numOptions;
691 CAQueueingThreadAddData(&gSendThread, data, sizeof(CAData_t));
695 // memory error label.
698 CADestroyRemoteEndpointInternal(remoteEndpoint);
705 return CA_MEMORY_ALLOC_FAILED;
708 void CASetRequestResponseCallbacks(CARequestCallback ReqHandler,
709 CAResponseCallback RespHandler)
711 OIC_LOG_V(DEBUG, TAG, "set request, response handler callback.");
713 gRequestHandler = ReqHandler;
714 gResponseHandler = RespHandler;
717 CAResult_t CAInitializeMessageHandler()
719 OIC_LOG(DEBUG, TAG, "CAInitializeMessageHandler - Entry");
720 CASetPacketReceivedCallback(CAReceivedPacketCallback);
722 CASetNetworkChangeCallback(CANetworkChangedCallback);
724 // create thread pool
726 res = u_thread_pool_init(MAX_THREAD_POOL_SIZE, &gThreadPoolHandle);
728 if (res != CA_STATUS_OK)
730 OIC_LOG_V(DEBUG, TAG, "thread pool initialize error.");
734 // send thread initialize
735 CAQueueingThreadInitialize(&gSendThread, gThreadPoolHandle, CASendThreadProcess);
738 res = CAQueueingThreadStart(&gSendThread);
740 if (res != CA_STATUS_OK)
742 OIC_LOG_V(DEBUG, TAG, "thread start error(send thread).");
746 // receive thread initialize
747 CAQueueingThreadInitialize(&gReceiveThread, gThreadPoolHandle, CAReceiveThreadProcess);
749 // start receive thread
751 // currently not support.
752 // res = CAThreadStart(gReceiveThread);
754 // if (res != CA_STATUS_OK)
756 // OIC_LOG_V(DEBUG, TAG, "thread start error(receive thread).");
760 // retransmission initialize
761 CARetransmissionInitialize(&gRetransmissionContext, gThreadPoolHandle, CASendUnicastData,
762 CATimeoutCallback, NULL);
764 // start retransmission
765 res = CARetransmissionStart(&gRetransmissionContext);
767 if (res != CA_STATUS_OK)
769 OIC_LOG_V(DEBUG, TAG, "thread start error(retransmission thread).");
773 // initialize interface adapters by controller
774 CAInitializeAdapters(gThreadPoolHandle);
779 void CATerminateMessageHandler()
781 // terminate interface adapters by controller
782 CATerminateAdapters();
784 // stop retransmission
785 CARetransmissionStop(&gRetransmissionContext);
786 CARetransmissionDestroy(&gRetransmissionContext);
789 // delete thread data
790 CAQueueingThreadStop(&gSendThread);
791 CAQueueingThreadDestroy(&gSendThread);
794 // delete thread data
795 CAQueueingThreadDestroy(&gReceiveThread);
797 // destroy thread pool
798 u_thread_pool_free(gThreadPoolHandle);
800 OIC_LOG_V(DEBUG, TAG, "message handler termination is complete!");