1 //******************************************************************
3 // Copyright 2014 Intel Mobile Communications GmbH All Rights Reserved.
5 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
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 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
23 #include "ocstackconfig.h"
24 #include "ocstackinternal.h"
25 #include "ocobserve.h"
26 #include "ocresourcehandler.h"
28 #include "oic_malloc.h"
29 #include "oic_string.h"
30 #include "ocpayload.h"
31 #include "ocserverrequest.h"
34 #include <coap/utlist.h>
39 #define MOD_NAME "ocobserve"
41 #define TAG "OIC_RI_OBSERVE"
43 #define VERIFY_NON_NULL(arg) { if (!arg) {OIC_LOG(FATAL, TAG, #arg " is NULL"); goto exit;} }
45 static struct ResourceObserver * g_serverObsList = NULL;
47 * Determine observe QOS based on the QOS of the request.
48 * The qos passed as a parameter overrides what the client requested.
49 * If we want the client preference taking high priority make:
50 * qos = resourceObserver->qos;
52 * @param method RESTful method.
53 * @param resourceObserver Observer.
54 * @param appQoS Quality of service.
55 * @return The quality of service of the observer.
57 static OCQualityOfService DetermineObserverQoS(OCMethod method,
58 ResourceObserver * resourceObserver, OCQualityOfService appQoS)
60 if (!resourceObserver)
62 OIC_LOG(ERROR, TAG, "DetermineObserverQoS called with invalid resourceObserver");
66 OCQualityOfService decidedQoS = appQoS;
67 if (appQoS == OC_NA_QOS)
69 decidedQoS = resourceObserver->qos;
72 if (appQoS != OC_HIGH_QOS)
74 OIC_LOG_V(INFO, TAG, "Current NON count for this observer is %d",
75 resourceObserver->lowQosCount);
77 if ((resourceObserver->forceHighQos \
78 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT) \
79 && method != OC_REST_PRESENCE)
81 if (resourceObserver->forceHighQos \
82 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT)
85 resourceObserver->lowQosCount = 0;
86 // at some point we have to to send CON to check on the
87 // availability of observer
88 OIC_LOG(INFO, TAG, "This time we are sending the notification as High qos");
89 decidedQoS = OC_HIGH_QOS;
93 (resourceObserver->lowQosCount)++;
100 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
101 OCPresenceTrigger trigger, OCResourceType *resourceType, OCQualityOfService qos)
103 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
104 OCQualityOfService qos)
107 OIC_LOG(INFO, TAG, "Entering SendObserverNotification");
110 return OC_STACK_INVALID_PARAM;
113 OCStackResult result = OC_STACK_ERROR;
114 ResourceObserver * resourceObserver = g_serverObsList;
116 OCServerRequest * request = NULL;
117 OCEntityHandlerRequest ehRequest = {0};
118 OCEntityHandlerResult ehResult = OC_EH_ERROR;
119 bool observeErrorFlag = false;
121 // Find clients that are observing this resource
122 while (resourceObserver)
124 if (resourceObserver->resource == resPtr)
128 if (method != OC_REST_PRESENCE)
131 qos = DetermineObserverQoS(method, resourceObserver, qos);
133 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
134 0, resPtr->sequenceNum, qos, resourceObserver->query,
136 resourceObserver->token, resourceObserver->tokenLength,
137 resourceObserver->resUri, 0, resourceObserver->acceptFormat,
138 &resourceObserver->devAddr);
142 request->observeResult = OC_STACK_OK;
143 if (result == OC_STACK_OK)
145 result = FormOCEntityHandlerRequest(
147 (OCRequestHandle) request,
150 (OCResourceHandle) resPtr,
152 PAYLOAD_TYPE_REPRESENTATION,
154 request->payloadSize,
155 request->numRcvdVendorSpecificHeaderOptions,
156 request->rcvdVendorSpecificHeaderOptions,
157 OC_OBSERVE_NO_OPTION,
160 if (result == OC_STACK_OK)
162 ehResult = resPtr->entityHandler(OC_REQUEST_FLAG, &ehRequest,
163 resPtr->entityHandlerCallbackParam);
164 if (ehResult == OC_EH_ERROR)
166 FindAndDeleteServerRequest(request);
169 OCPayloadDestroy(ehRequest.payload);
176 OCEntityHandlerResponse ehResponse = {0};
178 //This is effectively the implementation for the presence entity handler.
179 OIC_LOG(DEBUG, TAG, "This notification is for Presence");
180 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
181 0, resPtr->sequenceNum, qos, resourceObserver->query,
183 resourceObserver->token, resourceObserver->tokenLength,
184 resourceObserver->resUri, 0, resourceObserver->acceptFormat,
185 &resourceObserver->devAddr);
187 if (result == OC_STACK_OK)
189 OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
190 resPtr->sequenceNum, maxAge, trigger,
191 resourceType ? resourceType->resourcetypename : NULL);
195 return OC_STACK_NO_MEMORY;
198 if (result == OC_STACK_OK)
200 ehResponse.ehResult = OC_EH_OK;
201 ehResponse.payload = (OCPayload*)presenceResBuf;
202 ehResponse.persistentBufferFlag = 0;
203 ehResponse.requestHandle = (OCRequestHandle) request;
204 ehResponse.resourceHandle = (OCResourceHandle) resPtr;
205 OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
206 resourceObserver->resUri);
207 result = OCDoResponse(&ehResponse);
210 OCPresencePayloadDestroy(presenceResBuf);
215 // Since we are in a loop, set an error flag to indicate at least one error occurred.
216 if (result != OC_STACK_OK)
218 observeErrorFlag = true;
221 resourceObserver = resourceObserver->next;
226 OIC_LOG(INFO, TAG, "Resource has no observers");
227 result = OC_STACK_NO_OBSERVERS;
229 else if (observeErrorFlag)
231 OIC_LOG(ERROR, TAG, "Observer notification error");
232 result = OC_STACK_ERROR;
237 OCStackResult SendListObserverNotification (OCResource * resource,
238 OCObservationId *obsIdList, uint8_t numberOfIds,
239 const OCRepPayload *payload,
241 OCQualityOfService qos)
244 if (!resource || !obsIdList || !payload)
246 return OC_STACK_INVALID_PARAM;
249 uint8_t numIds = numberOfIds;
250 ResourceObserver *observer = NULL;
251 uint8_t numSentNotification = 0;
252 OCServerRequest * request = NULL;
253 OCStackResult result = OC_STACK_ERROR;
254 bool observeErrorFlag = false;
256 OIC_LOG(INFO, TAG, "Entering SendListObserverNotification");
259 observer = GetObserverUsingId (*obsIdList);
262 // Found observer - verify if it matches the resource handle
263 if (observer->resource == resource)
265 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
268 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
269 0, resource->sequenceNum, qos, observer->query,
270 NULL, NULL, observer->token, observer->tokenLength,
271 observer->resUri, 0, observer->acceptFormat,
276 request->observeResult = OC_STACK_OK;
277 if (result == OC_STACK_OK)
279 OCEntityHandlerResponse ehResponse = {0};
280 ehResponse.ehResult = OC_EH_OK;
281 ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
282 if (!ehResponse.payload)
284 FindAndDeleteServerRequest(request);
287 memcpy(ehResponse.payload, payload, sizeof(*payload));
288 ehResponse.persistentBufferFlag = 0;
289 ehResponse.requestHandle = (OCRequestHandle) request;
290 ehResponse.resourceHandle = (OCResourceHandle) resource;
291 result = OCDoResponse(&ehResponse);
292 if (result == OC_STACK_OK)
294 OIC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
296 // Increment only if OCDoResponse is successful
297 numSentNotification++;
299 OICFree(ehResponse.payload);
300 FindAndDeleteServerRequest(request);
304 OIC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
309 FindAndDeleteServerRequest(request);
312 // Since we are in a loop, set an error flag to indicate
313 // at least one error occurred.
314 if (result != OC_STACK_OK)
316 observeErrorFlag = true;
324 if (numSentNotification == numberOfIds && !observeErrorFlag)
328 else if (numSentNotification == 0)
330 return OC_STACK_NO_OBSERVERS;
334 OIC_LOG(ERROR, TAG, "Observer notification error");
335 return OC_STACK_ERROR;
339 OCStackResult GenerateObserverId (OCObservationId *observationId)
341 ResourceObserver *resObs = NULL;
343 OIC_LOG(INFO, TAG, "Entering GenerateObserverId");
344 VERIFY_NON_NULL (observationId);
348 *observationId = OCGetRandomByte();
349 // Check if observation Id already exists
350 resObs = GetObserverUsingId (*observationId);
351 } while (NULL != resObs);
353 OIC_LOG_V(INFO, TAG, "GeneratedObservation ID is %u", *observationId);
357 return OC_STACK_ERROR;
360 OCStackResult AddObserver (const char *resUri,
362 OCObservationId obsId,
365 OCResource *resHandle,
366 OCQualityOfService qos,
367 OCPayloadFormat acceptFormat,
368 const OCDevAddr *devAddr)
370 // Check if resource exists and is observable.
373 return OC_STACK_INVALID_PARAM;
375 if (!(resHandle->resourceProperties & OC_OBSERVABLE))
377 return OC_STACK_RESOURCE_ERROR;
380 if (!resUri || !token)
382 return OC_STACK_INVALID_PARAM;
385 ResourceObserver *obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
388 obsNode->observeId = obsId;
390 obsNode->resUri = OICStrdup(resUri);
391 VERIFY_NON_NULL (obsNode->resUri);
394 obsNode->acceptFormat = acceptFormat;
397 obsNode->query = OICStrdup(query);
398 VERIFY_NON_NULL (obsNode->query);
400 // If tokenLength is zero, the return value depends on the
401 // particular library implementation (it may or may not be a null pointer).
404 obsNode->token = (CAToken_t)OICMalloc(tokenLength);
405 VERIFY_NON_NULL (obsNode->token);
406 memcpy(obsNode->token, token, tokenLength);
408 obsNode->tokenLength = tokenLength;
410 obsNode->devAddr = *devAddr;
411 obsNode->resource = resHandle;
413 LL_APPEND (g_serverObsList, obsNode);
421 OICFree(obsNode->resUri);
422 OICFree(obsNode->query);
425 return OC_STACK_NO_MEMORY;
428 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
430 ResourceObserver *out = NULL;
434 LL_FOREACH (g_serverObsList, out)
436 if (out->observeId == observeId)
442 OIC_LOG(INFO, TAG, "Observer node not found!!");
446 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
448 ResourceObserver *out = NULL;
452 OIC_LOG(INFO, TAG, "Looking for token");
453 OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
454 OIC_LOG(INFO, TAG, "\tFound token:");
456 LL_FOREACH (g_serverObsList, out)
458 OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
459 if ((memcmp(out->token, token, tokenLength) == 0))
467 OIC_LOG(ERROR, TAG, "Passed in NULL token");
470 OIC_LOG(INFO, TAG, "Observer node not found!!");
474 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
478 return OC_STACK_INVALID_PARAM;
481 ResourceObserver *obsNode = GetObserverUsingToken (token, tokenLength);
484 OIC_LOG_V(INFO, TAG, "deleting observer id %u with token", obsNode->observeId);
485 OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
486 LL_DELETE (g_serverObsList, obsNode);
487 OICFree(obsNode->resUri);
488 OICFree(obsNode->query);
489 OICFree(obsNode->token);
492 // it is ok if we did not find the observer...
496 void DeleteObserverList()
498 ResourceObserver *out = NULL;
499 ResourceObserver *tmp = NULL;
500 LL_FOREACH_SAFE (g_serverObsList, out, tmp)
504 DeleteObserverUsingToken ((out->token), out->tokenLength);
507 g_serverObsList = NULL;
511 * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
512 * option, which breaks the protocol abstraction requirement between RI & CA, and
513 * has to be fixed in the future. The function below adds the header option for observe.
514 * It should be noted that the observe header option is assumed to be the first option
515 * in the list of user defined header options and hence it is inserted at the front
516 * of the header options list and number of options adjusted accordingly.
519 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
520 OCHeaderOption *ocHdrOpt,
526 return OC_STACK_INVALID_PARAM;
529 if (numOptions > 0 && !ocHdrOpt)
531 OIC_LOG (INFO, TAG, "options are NULL though number is non zero");
532 return OC_STACK_INVALID_PARAM;
535 CAHeaderOption_t *tmpHdrOpt = NULL;
537 tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
538 if (NULL == tmpHdrOpt)
540 return OC_STACK_NO_MEMORY;
542 tmpHdrOpt[0].protocolID = CA_COAP_ID;
543 tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
544 tmpHdrOpt[0].optionLength = sizeof(uint8_t);
545 tmpHdrOpt[0].optionData[0] = observeFlag;
546 for (uint8_t i = 0; i < numOptions; i++)
548 memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
551 *caHdrOpt = tmpHdrOpt;
556 * CA layer passes observe information to the RI layer as a header option, which
557 * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
558 * in the future. The function below removes the observe header option and processes it.
559 * It should be noted that the observe header option is always assumed to be the first
560 * option in the list of user defined header options and hence it is deleted from the
561 * front of the header options list and the number of options is adjusted accordingly.
564 GetObserveHeaderOption (uint32_t * observationOption,
565 CAHeaderOption_t *options,
566 uint8_t * numOptions)
568 if (!observationOption)
570 return OC_STACK_INVALID_PARAM;
573 if (!options || !numOptions)
575 OIC_LOG (INFO, TAG, "No options present");
579 for(uint8_t i = 0; i < *numOptions; i++)
581 if (options[i].protocolID == CA_COAP_ID &&
582 options[i].optionID == COAP_OPTION_OBSERVE)
584 *observationOption = options[i].optionData[0];
585 for(uint8_t c = i; c < *numOptions-1; c++)
587 options[i] = options[i+1];