f7b2d3c1843344aecfce203200fdafb2c88ea342
[platform/upstream/iotivity.git] / resource / csdk / stack / src / ocobserve.c
1 //******************************************************************
2 //
3 // Copyright 2014 Intel Mobile Communications GmbH 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 <string.h>
22 #include "ocstack.h"
23 #include "ocstackconfig.h"
24 #include "ocstackinternal.h"
25 #include "ocobserve.h"
26 #include "ocresourcehandler.h"
27 #include "ocrandom.h"
28 #include "oic_malloc.h"
29 #include "oic_string.h"
30 #include "ocpayload.h"
31 #include "ocserverrequest.h"
32 #include "logger.h"
33
34 #include <coap/utlist.h>
35 #include <coap/pdu.h>
36 #include <coap/coap.h>
37
38 // Module Name
39 #define MOD_NAME "ocobserve"
40
41 #define TAG  "OIC_RI_OBSERVE"
42
43 #define VERIFY_NON_NULL(arg) { if (!arg) {OIC_LOG(FATAL, TAG, #arg " is NULL"); goto exit;} }
44
45 static struct ResourceObserver * g_serverObsList = NULL;
46 /**
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;
51  *
52  * @param method RESTful method.
53  * @param resourceObserver Observer.
54  * @param appQoS Quality of service.
55  * @return The quality of service of the observer.
56  */
57 static OCQualityOfService DetermineObserverQoS(OCMethod method,
58         ResourceObserver * resourceObserver, OCQualityOfService appQoS)
59 {
60     if (!resourceObserver)
61     {
62         OIC_LOG(ERROR, TAG, "DetermineObserverQoS called with invalid resourceObserver");
63         return OC_NA_QOS;
64     }
65
66     OCQualityOfService decidedQoS = appQoS;
67     if (appQoS == OC_NA_QOS)
68     {
69         decidedQoS = resourceObserver->qos;
70     }
71
72     if (appQoS != OC_HIGH_QOS)
73     {
74         OIC_LOG_V(INFO, TAG, "Current NON count for this observer is %d",
75                 resourceObserver->lowQosCount);
76 #ifdef WITH_PRESENCE
77         if ((resourceObserver->forceHighQos \
78                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT) \
79                 && method != OC_REST_PRESENCE)
80 #else
81         if (resourceObserver->forceHighQos \
82                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT)
83 #endif
84         {
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;
90         }
91         else
92         {
93             (resourceObserver->lowQosCount)++;
94         }
95     }
96     return decidedQoS;
97 }
98
99 /**
100  * Create a get request and pass to entityhandler to notify specific observer.
101  *
102  * @param observer Observer that need to be notified.
103  * @param qos Quality of service of resource.
104  *
105  * @return ::OC_STACK_OK on success, some other value upon failure.
106  */
107 static OCStackResult SendObserveNotification(ResourceObserver *observer,
108                                              OCQualityOfService qos)
109 {
110     OCStackResult result = OC_STACK_ERROR;
111     OCServerRequest * request = NULL;
112     OCEntityHandlerRequest ehRequest = {0};
113     OCEntityHandlerResult ehResult = OC_EH_ERROR;
114
115     result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
116                               0, observer->resource->sequenceNum, qos,
117                               observer->query, NULL, NULL,
118                               observer->token, observer->tokenLength,
119                               observer->resUri, 0, observer->acceptFormat,
120                               &observer->devAddr);
121
122     if (request)
123     {
124         request->observeResult = OC_STACK_OK;
125         if (result == OC_STACK_OK)
126         {
127             result = FormOCEntityHandlerRequest(
128                         &ehRequest,
129                         (OCRequestHandle) request->requestId,
130                         request->method,
131                         &request->devAddr,
132                         (OCResourceHandle) observer->resource,
133                         request->query,
134                         PAYLOAD_TYPE_REPRESENTATION,
135                         request->payload,
136                         request->payloadSize,
137                         request->numRcvdVendorSpecificHeaderOptions,
138                         request->rcvdVendorSpecificHeaderOptions,
139                         OC_OBSERVE_NO_OPTION,
140                         0,
141                         request->coapID);
142             if (result == OC_STACK_OK)
143             {
144                 ehResult = observer->resource->entityHandler(OC_REQUEST_FLAG, &ehRequest,
145                                     observer->resource->entityHandlerCallbackParam);
146                 if (ehResult == OC_EH_ERROR)
147                 {
148                     FindAndDeleteServerRequest(request);
149                 }
150                 // Reset Observer TTL.
151                 observer->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
152             }
153             OCPayloadDestroy(ehRequest.payload);
154         }
155     }
156
157     return result;
158 }
159
160 #ifdef WITH_PRESENCE
161 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
162         OCPresenceTrigger trigger, OCResourceType *resourceType, OCQualityOfService qos)
163 #else
164 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
165         OCQualityOfService qos)
166 #endif
167 {
168     OIC_LOG(INFO, TAG, "Entering SendObserverNotification");
169     if (!resPtr)
170     {
171         return OC_STACK_INVALID_PARAM;
172     }
173
174     OCStackResult result = OC_STACK_ERROR;
175     ResourceObserver * resourceObserver = g_serverObsList;
176     uint8_t numObs = 0;
177     OCServerRequest * request = NULL;
178     OCEntityHandlerRequest ehRequest = {0};
179     OCEntityHandlerResult ehResult = OC_EH_ERROR;
180     bool observeErrorFlag = false;
181
182     // Find clients that are observing this resource
183     while (resourceObserver)
184     {
185         if (resourceObserver->resource == resPtr)
186         {
187             numObs++;
188 #ifdef WITH_PRESENCE
189             if (method != OC_REST_PRESENCE)
190             {
191 #endif
192                 qos = DetermineObserverQoS(method, resourceObserver, qos);
193                 result = SendObserveNotification(resourceObserver, qos);
194 #ifdef WITH_PRESENCE
195             }
196             else
197             {
198                 OCEntityHandlerResponse ehResponse = {0};
199
200                 //This is effectively the implementation for the presence entity handler.
201                 OIC_LOG(DEBUG, TAG, "This notification is for Presence");
202                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
203                         0, resPtr->sequenceNum, qos, resourceObserver->query,
204                         NULL, NULL,
205                         resourceObserver->token, resourceObserver->tokenLength,
206                         resourceObserver->resUri, 0, resourceObserver->acceptFormat,
207                         &resourceObserver->devAddr);
208
209                 if (result == OC_STACK_OK)
210                 {
211                     OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
212                             resPtr->sequenceNum, maxAge, trigger,
213                             resourceType ? resourceType->resourcetypename : NULL);
214
215                     if (!presenceResBuf)
216                     {
217                         return OC_STACK_NO_MEMORY;
218                     }
219
220                     if (result == OC_STACK_OK)
221                     {
222                         ehResponse.ehResult = OC_EH_OK;
223                         ehResponse.payload = (OCPayload*)presenceResBuf;
224                         ehResponse.persistentBufferFlag = 0;
225                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
226                         ehResponse.resourceHandle = (OCResourceHandle) resPtr;
227                         OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
228                                 resourceObserver->resUri);
229                         result = OCDoResponse(&ehResponse);
230                     }
231
232                     OCPresencePayloadDestroy(presenceResBuf);
233                 }
234             }
235 #endif
236
237             // Since we are in a loop, set an error flag to indicate at least one error occurred.
238             if (result != OC_STACK_OK)
239             {
240                 observeErrorFlag = true;
241             }
242         }
243         resourceObserver = resourceObserver->next;
244     }
245
246     if (numObs == 0)
247     {
248         OIC_LOG(INFO, TAG, "Resource has no observers");
249         result = OC_STACK_NO_OBSERVERS;
250     }
251     else if (observeErrorFlag)
252     {
253         OIC_LOG(ERROR, TAG, "Observer notification error");
254         result = OC_STACK_ERROR;
255     }
256     return result;
257 }
258
259 OCStackResult SendListObserverNotification (OCResource * resource,
260         OCObservationId  *obsIdList, uint8_t numberOfIds,
261         const OCRepPayload *payload,
262         uint32_t maxAge,
263         OCQualityOfService qos)
264 {
265     (void)maxAge;
266     if (!resource || !obsIdList || !payload)
267     {
268         return OC_STACK_INVALID_PARAM;
269     }
270
271     uint8_t numIds = numberOfIds;
272     ResourceObserver *observer = NULL;
273     uint8_t numSentNotification = 0;
274     OCServerRequest * request = NULL;
275     OCStackResult result = OC_STACK_ERROR;
276     bool observeErrorFlag = false;
277
278     OIC_LOG(INFO, TAG, "Entering SendListObserverNotification");
279     while(numIds)
280     {
281         observer = GetObserverUsingId (*obsIdList);
282         if (observer)
283         {
284             // Found observer - verify if it matches the resource handle
285             if (observer->resource == resource)
286             {
287                 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
288
289
290                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
291                         0, resource->sequenceNum, qos, observer->query,
292                         NULL, NULL, observer->token, observer->tokenLength,
293                         observer->resUri, 0, observer->acceptFormat,
294                         &observer->devAddr);
295
296                 if (request)
297                 {
298                     request->observeResult = OC_STACK_OK;
299                     if (result == OC_STACK_OK)
300                     {
301                         OCEntityHandlerResponse ehResponse = {0};
302                         ehResponse.ehResult = OC_EH_OK;
303                         ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
304                         if (!ehResponse.payload)
305                         {
306                             FindAndDeleteServerRequest(request);
307                             continue;
308                         }
309                         memcpy(ehResponse.payload, payload, sizeof(*payload));
310                         ehResponse.persistentBufferFlag = 0;
311                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
312                         ehResponse.resourceHandle = (OCResourceHandle) resource;
313                         result = OCDoResponse(&ehResponse);
314                         if (result == OC_STACK_OK)
315                         {
316                             OIC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
317
318                             // Increment only if OCDoResponse is successful
319                             numSentNotification++;
320
321                             OICFree(ehResponse.payload);
322                             FindAndDeleteServerRequest(request);
323                         }
324                         else
325                         {
326                             OIC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
327                         }
328                         // Reset Observer TTL.
329                         observer->TTL =
330                                 GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
331                     }
332                     else
333                     {
334                         FindAndDeleteServerRequest(request);
335                     }
336                 }
337                 // Since we are in a loop, set an error flag to indicate
338                 // at least one error occurred.
339                 if (result != OC_STACK_OK)
340                 {
341                     observeErrorFlag = true;
342                 }
343             }
344         }
345         obsIdList++;
346         numIds--;
347     }
348
349     if (numSentNotification == numberOfIds && !observeErrorFlag)
350     {
351         return OC_STACK_OK;
352     }
353     else if (numSentNotification == 0)
354     {
355         return OC_STACK_NO_OBSERVERS;
356     }
357     else
358     {
359         OIC_LOG(ERROR, TAG, "Observer notification error");
360         return OC_STACK_ERROR;
361     }
362 }
363
364 OCStackResult GenerateObserverId (OCObservationId *observationId)
365 {
366     ResourceObserver *resObs = NULL;
367
368     OIC_LOG(INFO, TAG, "Entering GenerateObserverId");
369     VERIFY_NON_NULL (observationId);
370
371     do
372     {
373         *observationId = OCGetRandomByte();
374         // Check if observation Id already exists
375         resObs = GetObserverUsingId (*observationId);
376     } while (NULL != resObs);
377
378     OIC_LOG_V(INFO, TAG, "GeneratedObservation ID is %u", *observationId);
379
380     return OC_STACK_OK;
381 exit:
382     return OC_STACK_ERROR;
383 }
384
385 OCStackResult AddObserver (const char         *resUri,
386                            const char         *query,
387                            OCObservationId    obsId,
388                            CAToken_t          token,
389                            uint8_t            tokenLength,
390                            OCResource         *resHandle,
391                            OCQualityOfService qos,
392                            OCPayloadFormat    acceptFormat,
393                            const OCDevAddr    *devAddr)
394 {
395     // Check if resource exists and is observable.
396     if (!resHandle)
397     {
398         return OC_STACK_INVALID_PARAM;
399     }
400     if (!(resHandle->resourceProperties & OC_OBSERVABLE))
401     {
402         return OC_STACK_RESOURCE_ERROR;
403     }
404
405     if (!resUri || !token)
406     {
407         return OC_STACK_INVALID_PARAM;
408     }
409
410     ResourceObserver *obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
411     if (obsNode)
412     {
413         obsNode->observeId = obsId;
414
415         obsNode->resUri = OICStrdup(resUri);
416         VERIFY_NON_NULL (obsNode->resUri);
417
418         obsNode->qos = qos;
419         obsNode->acceptFormat = acceptFormat;
420         if (query)
421         {
422             obsNode->query = OICStrdup(query);
423             VERIFY_NON_NULL (obsNode->query);
424         }
425         // If tokenLength is zero, the return value depends on the
426         // particular library implementation (it may or may not be a null pointer).
427         if (tokenLength)
428         {
429             obsNode->token = (CAToken_t)OICMalloc(tokenLength);
430             VERIFY_NON_NULL (obsNode->token);
431             memcpy(obsNode->token, token, tokenLength);
432         }
433         obsNode->tokenLength = tokenLength;
434
435         obsNode->devAddr = *devAddr;
436         obsNode->resource = resHandle;
437
438         if ((strcmp(resUri, OC_RSRVD_PRESENCE_URI) == 0))
439         {
440             obsNode->TTL = 0;
441         }
442         else
443         {
444             obsNode->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
445         }
446
447         LL_APPEND (g_serverObsList, obsNode);
448
449         return OC_STACK_OK;
450     }
451
452 exit:
453     if (obsNode)
454     {
455         OICFree(obsNode->resUri);
456         OICFree(obsNode->query);
457         OICFree(obsNode);
458     }
459     return OC_STACK_NO_MEMORY;
460 }
461
462 /*
463  * This function checks if the node is past its time to live and
464  * deletes it if timed-out. Calling this function with a  presence callback
465  * with ttl set to 0 will not delete anything as presence nodes have
466  * their own mechanisms for timeouts. A null argument will cause the function to
467  * silently return.
468  */
469 static void CheckTimedOutObserver(ResourceObserver* observer)
470 {
471     if (!observer || observer->TTL == 0)
472     {
473         return;
474     }
475
476     coap_tick_t now;
477     coap_ticks(&now);
478
479     if (observer->TTL < now)
480     {
481         // Send confirmable notification message to observer.
482         OIC_LOG(INFO, TAG, "Sending High-QoS notification to observer");
483         SendObserveNotification(observer, OC_HIGH_QOS);
484     }
485 }
486
487 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
488 {
489     ResourceObserver *out = NULL;
490
491     if (observeId)
492     {
493         LL_FOREACH (g_serverObsList, out)
494         {
495             if (out->observeId == observeId)
496             {
497                 return out;
498             }
499             CheckTimedOutObserver(out);
500         }
501     }
502     OIC_LOG(INFO, TAG, "Observer node not found!!");
503     return NULL;
504 }
505
506 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
507 {
508     if (token)
509     {
510         OIC_LOG(INFO, TAG, "Looking for token");
511         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
512
513         ResourceObserver *out = NULL;
514         LL_FOREACH (g_serverObsList, out)
515         {
516             /* de-annotate below line if want to see all token in cbList */
517             //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
518             if ((memcmp(out->token, token, tokenLength) == 0))
519             {
520                 OIC_LOG(INFO, TAG, "Found in observer list");
521                 return out;
522             }
523             CheckTimedOutObserver(out);
524         }
525     }
526     else
527     {
528         OIC_LOG(ERROR, TAG, "Passed in NULL token");
529     }
530
531     OIC_LOG(INFO, TAG, "Observer node not found!!");
532     return NULL;
533 }
534
535 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
536 {
537     if (!token)
538     {
539         return OC_STACK_INVALID_PARAM;
540     }
541
542     ResourceObserver *obsNode = GetObserverUsingToken (token, tokenLength);
543     if (obsNode)
544     {
545         OIC_LOG_V(INFO, TAG, "deleting observer id  %u with token", obsNode->observeId);
546         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
547         LL_DELETE (g_serverObsList, obsNode);
548         OICFree(obsNode->resUri);
549         OICFree(obsNode->query);
550         OICFree(obsNode->token);
551         OICFree(obsNode);
552     }
553     // it is ok if we did not find the observer...
554     return OC_STACK_OK;
555 }
556
557 OCStackResult DeleteObserverUsingDevAddr(const OCDevAddr *devAddr)
558 {
559     if (!devAddr)
560     {
561         return OC_STACK_INVALID_PARAM;
562     }
563
564     ResourceObserver *out = NULL;
565     ResourceObserver *tmp = NULL;
566     LL_FOREACH_SAFE(g_serverObsList, out, tmp)
567     {
568         if (out)
569         {
570             if ((strcmp(out->devAddr.addr, devAddr->addr) == 0)
571                     && out->devAddr.port == devAddr->port)
572             {
573                 OIC_LOG_V(INFO, TAG, "deleting observer id  %u with %s:%u",
574                           out->observeId, out->devAddr.addr, out->devAddr.port);
575                 OCStackFeedBack(out->token, out->tokenLength, OC_OBSERVER_NOT_INTERESTED);
576             }
577         }
578     }
579
580     return OC_STACK_OK;
581 }
582
583 void DeleteObserverList()
584 {
585     ResourceObserver *out = NULL;
586     ResourceObserver *tmp = NULL;
587     LL_FOREACH_SAFE (g_serverObsList, out, tmp)
588     {
589         if (out)
590         {
591             DeleteObserverUsingToken ((out->token), out->tokenLength);
592         }
593     }
594     g_serverObsList = NULL;
595 }
596
597 /*
598  * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
599  * option, which breaks the protocol abstraction requirement between RI & CA, and
600  * has to be fixed in the future. The function below adds the header option for observe.
601  * It should be noted that the observe header option is assumed to be the first option
602  * in the list of user defined header options and hence it is inserted at the front
603  * of the header options list and number of options adjusted accordingly.
604  */
605 OCStackResult
606 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
607                            OCHeaderOption *ocHdrOpt,
608                            uint8_t numOptions,
609                            uint8_t observeFlag)
610 {
611     if (!caHdrOpt)
612     {
613         return OC_STACK_INVALID_PARAM;
614     }
615
616     if (numOptions > 0 && !ocHdrOpt)
617     {
618         OIC_LOG (INFO, TAG, "options are NULL though number is non zero");
619         return OC_STACK_INVALID_PARAM;
620     }
621
622     CAHeaderOption_t *tmpHdrOpt = NULL;
623
624     tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
625     if (NULL == tmpHdrOpt)
626     {
627         return OC_STACK_NO_MEMORY;
628     }
629     tmpHdrOpt[0].protocolID = CA_COAP_ID;
630     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
631     tmpHdrOpt[0].optionLength = sizeof(uint8_t);
632     tmpHdrOpt[0].optionData[0] = observeFlag;
633     for (uint8_t i = 0; i < numOptions; i++)
634     {
635         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
636     }
637
638     *caHdrOpt = tmpHdrOpt;
639     return OC_STACK_OK;
640 }
641
642 /*
643  * CA layer passes observe information to the RI layer as a header option, which
644  * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
645  * in the future. The function below removes the observe header option and processes it.
646  * It should be noted that the observe header option is always assumed to be the first
647  * option in the list of user defined header options and hence it is deleted from the
648  * front of the header options list and the number of options is adjusted accordingly.
649  */
650 OCStackResult
651 GetObserveHeaderOption (uint32_t * observationOption,
652                         CAHeaderOption_t *options,
653                         uint8_t * numOptions)
654 {
655     if (!observationOption)
656     {
657         return OC_STACK_INVALID_PARAM;
658     }
659
660     if (!options || !numOptions)
661     {
662         OIC_LOG (INFO, TAG, "No options present");
663         return OC_STACK_OK;
664     }
665
666     for(uint8_t i = 0; i < *numOptions; i++)
667     {
668         if (options[i].protocolID == CA_COAP_ID &&
669                 options[i].optionID == COAP_OPTION_OBSERVE)
670         {
671             *observationOption = options[i].optionData[0];
672             for(uint8_t c = i; c < *numOptions-1; c++)
673             {
674                 options[i] = options[i+1];
675             }
676             (*numOptions)--;
677             return OC_STACK_OK;
678         }
679     }
680     return OC_STACK_OK;
681 }
682