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
  33 #include "utlist.h"
  34 #include "pdu.h"
  35
  36
  37 // Module Name
  38 #define MOD_NAME PCF("ocobserve")
  39
  40 #define TAG  PCF("OCStackObserve")
  41
  42 #define VERIFY_NON_NULL(arg) { if (!arg) {OC_LOG(FATAL, TAG, #arg " is NULL"); goto exit;} }
  43
  44 static struct ResourceObserver * serverObsList = NULL;
  45 /**
  46  * Determine observe QOS based on the QOS of the request.
  47  * The qos passed as a parameter overrides what the client requested.
  48  * If we want the client preference taking high priority make:
  49  *     qos = resourceObserver->qos;
  50  *
  51  * @param method RESTful method.
  52  * @param resourceObserver Observer.
  53  * @param appQoS Quality of service.
  54  * @return The quality of service of the observer.
  55  */
  56 static OCQualityOfService DetermineObserverQoS(OCMethod method,
  57         ResourceObserver * resourceObserver, OCQualityOfService appQoS)
  58 {
  59     if(!resourceObserver)
  60     {
  61         OC_LOG(ERROR, TAG, "DetermineObserverQoS called with invalid resourceObserver");
  62         return OC_NA_QOS;
  63     }
  64
  65     OCQualityOfService decidedQoS = appQoS;
  66     if(appQoS == OC_NA_QOS)
  67     {
  68         decidedQoS = resourceObserver->qos;
  69     }
  70
  71     if(appQoS != OC_HIGH_QOS)
  72     {
  73         OC_LOG_V(INFO, TAG, "Current NON count for this observer is %d",
  74                 resourceObserver->lowQosCount);
  75         #ifdef WITH_PRESENCE
  76         if((resourceObserver->forceHighQos \
  77                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT) \
  78                 && method != OC_REST_PRESENCE)
  79         #else
  80         if(resourceObserver->forceHighQos \
  81                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT)
  82         #endif
  83         {
  84             resourceObserver->lowQosCount = 0;
  85             // at some point we have to to send CON to check on the
  86             // availability of observer
  87             OC_LOG(INFO, TAG, PCF("This time we are sending the  notification as High qos"));
  88             decidedQoS = OC_HIGH_QOS;
  89         }
  90         else
  91         {
  92             (resourceObserver->lowQosCount)++;
  93         }
  94     }
  95     return decidedQoS;
  96 }
  97
  98 #ifdef WITH_PRESENCE
  99 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 100         OCPresenceTrigger trigger, OCResourceType *resourceType, OCQualityOfService qos)
 101 #else
 102 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 103         OCQualityOfService qos)
 104 #endif
 105 {
 106     OC_LOG(INFO, TAG, PCF("Entering SendObserverNotification"));
 107     if(!resPtr)
 108     {
 109         return OC_STACK_INVALID_PARAM;
 110     }
 111
 112     OCStackResult result = OC_STACK_ERROR;
 113     ResourceObserver * resourceObserver = serverObsList;
 114     uint8_t numObs = 0;
 115     OCServerRequest * request = NULL;
 116     OCEntityHandlerRequest ehRequest = {};
 117     OCEntityHandlerResult ehResult = OC_EH_ERROR;
 118     bool observeErrorFlag = false;
 119
 120     // Find clients that are observing this resource
 121     while (resourceObserver)
 122     {
 123         if (resourceObserver->resource == resPtr)
 124         {
 125             numObs++;
 126             #ifdef WITH_PRESENCE
 127             if(method != OC_REST_PRESENCE)
 128             {
 129             #endif
 130                 qos = DetermineObserverQoS(method, resourceObserver, qos);
 131
 132                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 133                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 134                         NULL, NULL,
 135                         resourceObserver->token, resourceObserver->tokenLength,
 136                         resourceObserver->resUri, 0,
 137                         &resourceObserver->devAddr);
 138
 139                 if(request)
 140                 {
 141                     request->observeResult = OC_STACK_OK;
 142                     if(result == OC_STACK_OK)
 143                     {
 144                         result = FormOCEntityHandlerRequest(&ehRequest, (OCRequestHandle) request,
 145                                     request->method, (OCResourceHandle) resPtr, request->query,
 146                                     request->payload, request->payloadSize,
 147                                     request->numRcvdVendorSpecificHeaderOptions,
 148                                     request->rcvdVendorSpecificHeaderOptions,
 149                                     OC_OBSERVE_NO_OPTION, 0);
 150                         if(result == OC_STACK_OK)
 151                         {
 152                             ehResult = resPtr->entityHandler(OC_REQUEST_FLAG, &ehRequest,
 153                                                 resPtr->entityHandlerCallbackParam);
 154                             if(ehResult == OC_EH_ERROR)
 155                             {
 156                                 FindAndDeleteServerRequest(request);
 157                             }
 158                         }
 159                     }
 160                 }
 161             #ifdef WITH_PRESENCE
 162             }
 163             else
 164             {
 165                 OCEntityHandlerResponse ehResponse = {};
 166
 167                 //This is effectively the implementation for the presence entity handler.
 168                 OC_LOG(DEBUG, TAG, PCF("This notification is for Presence"));
 169                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 170                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 171                         NULL, NULL,
 172                         resourceObserver->token, resourceObserver->tokenLength,
 173                         resourceObserver->resUri, 0,
 174                         &resourceObserver->devAddr);
 175
 176                 if(result == OC_STACK_OK)
 177                 {
 178                     OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
 179                             resPtr->sequenceNum, maxAge, trigger,
 180                             resourceType ? resourceType->resourcetypename : NULL);
 181
 182                     if(!presenceResBuf)
 183                     {
 184                         return OC_STACK_NO_MEMORY;
 185                     }
 186
 187                     if(result == OC_STACK_OK)
 188                     {
 189                         ehResponse.ehResult = OC_EH_OK;
 190                         ehResponse.payload = (OCPayload*)presenceResBuf;
 191                         ehResponse.persistentBufferFlag = 0;
 192                         ehResponse.requestHandle = (OCRequestHandle) request;
 193                         ehResponse.resourceHandle = (OCResourceHandle) resPtr;
 194                         OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
 195                                 resourceObserver->resUri);
 196                         result = OCDoResponse(&ehResponse);
 197                     }
 198
 199                     OCPresencePayloadDestroy(presenceResBuf);
 200                 }
 201             }
 202             #endif
 203
 204             // Since we are in a loop, set an error flag to indicate at least one error occurred.
 205             if (result != OC_STACK_OK)
 206             {
 207                 observeErrorFlag = true;
 208             }
 209         }
 210         resourceObserver = resourceObserver->next;
 211     }
 212
 213     if (numObs == 0)
 214     {
 215         OC_LOG(INFO, TAG, PCF("Resource has no observers"));
 216         result = OC_STACK_NO_OBSERVERS;
 217     }
 218     else if (observeErrorFlag)
 219     {
 220         OC_LOG(ERROR, TAG, PCF("Observer notification error"));
 221         result = OC_STACK_ERROR;
 222     }
 223     return result;
 224 }
 225
 226 OCStackResult SendListObserverNotification (OCResource * resource,
 227         OCObservationId  *obsIdList, uint8_t numberOfIds,
 228         const OCRepPayload *payload,
 229         uint32_t maxAge,
 230         OCQualityOfService qos)
 231 {
 232     if(!resource || !obsIdList || !payload)
 233     {
 234         return OC_STACK_INVALID_PARAM;
 235     }
 236
 237     uint8_t numIds = numberOfIds;
 238     ResourceObserver *observer = NULL;
 239     uint8_t numSentNotification = 0;
 240     OCServerRequest * request = NULL;
 241     OCStackResult result = OC_STACK_ERROR;
 242     bool observeErrorFlag = false;
 243
 244     OC_LOG(INFO, TAG, PCF("Entering SendListObserverNotification"));
 245     while(numIds)
 246     {
 247         observer = GetObserverUsingId (*obsIdList);
 248         if(observer)
 249         {
 250             // Found observer - verify if it matches the resource handle
 251             if (observer->resource == resource)
 252             {
 253                 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
 254
 255
 256                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 257                         0, resource->sequenceNum, qos, observer->query,
 258                         NULL, NULL, observer->token, observer->tokenLength,
 259                         observer->resUri, 0,
 260                         &observer->devAddr);
 261
 262                 if(request)
 263                 {
 264                     request->observeResult = OC_STACK_OK;
 265                     if(result == OC_STACK_OK)
 266                     {
 267                         OCEntityHandlerResponse ehResponse = {};
 268                         ehResponse.ehResult = OC_EH_OK;
 269                         ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
 270                         if(!ehResponse.payload)
 271                         {
 272                             FindAndDeleteServerRequest(request);
 273                             continue;
 274                         }
 275                         memcpy(ehResponse.payload, payload, sizeof(*payload));
 276                         ehResponse.persistentBufferFlag = 0;
 277                         ehResponse.requestHandle = (OCRequestHandle) request;
 278                         ehResponse.resourceHandle = (OCResourceHandle) resource;
 279                         result = OCDoResponse(&ehResponse);
 280                         if(result == OC_STACK_OK)
 281                         {
 282                             OC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
 283
 284                             // Increment only if OCDoResponse is successful
 285                             numSentNotification++;
 286
 287                             OICFree(ehResponse.payload);
 288                             FindAndDeleteServerRequest(request);
 289                         }
 290                         else
 291                         {
 292                             OC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
 293                         }
 294                     }
 295                     else
 296                     {
 297                         FindAndDeleteServerRequest(request);
 298                     }
 299                 }
 300                 // Since we are in a loop, set an error flag to indicate
 301                 // at least one error occurred.
 302                 if (result != OC_STACK_OK)
 303                 {
 304                     observeErrorFlag = true;
 305                 }
 306             }
 307         }
 308         obsIdList++;
 309         numIds--;
 310     }
 311
 312     if(numSentNotification == numberOfIds && !observeErrorFlag)
 313     {
 314         return OC_STACK_OK;
 315     }
 316     else if(numSentNotification == 0)
 317     {
 318         return OC_STACK_NO_OBSERVERS;
 319     }
 320     else
 321     {
 322         OC_LOG(ERROR, TAG, PCF("Observer notification error"));
 323         return OC_STACK_ERROR;
 324     }
 325 }
 326
 327 OCStackResult GenerateObserverId (OCObservationId *observationId)
 328 {
 329     ResourceObserver *resObs = NULL;
 330
 331     OC_LOG(INFO, TAG, PCF("Entering GenerateObserverId"));
 332     VERIFY_NON_NULL (observationId);
 333
 334     do
 335     {
 336         *observationId = OCGetRandomByte();
 337         // Check if observation Id already exists
 338         resObs = GetObserverUsingId (*observationId);
 339     } while (NULL != resObs);
 340
 341     OC_LOG_V(INFO, TAG, "Generated bservation ID is %u", *observationId);
 342
 343     return OC_STACK_OK;
 344 exit:
 345     return OC_STACK_ERROR;
 346 }
 347
 348 OCStackResult AddObserver (const char         *resUri,
 349                            const char         *query,
 350                            OCObservationId    obsId,
 351                            CAToken_t          token,
 352                            uint8_t            tokenLength,
 353                            OCResource         *resHandle,
 354                            OCQualityOfService qos,
 355                            const OCDevAddr    *devAddr)
 356 {
 357     // Check if resource exists and is observable.
 358     if (!resHandle)
 359     {
 360         return OC_STACK_INVALID_PARAM;
 361     }
 362     if (!(resHandle->resourceProperties & OC_OBSERVABLE))
 363     {
 364         return OC_STACK_RESOURCE_ERROR;
 365     }
 366     ResourceObserver *obsNode = NULL;
 367
 368     if(!resUri || !token || !*token)
 369     {
 370         return OC_STACK_INVALID_PARAM;
 371     }
 372
 373     obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
 374     if (obsNode)
 375     {
 376         obsNode->observeId = obsId;
 377
 378         obsNode->resUri = OICStrdup(resUri);
 379         VERIFY_NON_NULL (obsNode->resUri);
 380
 381         obsNode->qos = qos;
 382         if(query)
 383         {
 384             obsNode->query = OICStrdup(query);
 385             VERIFY_NON_NULL (obsNode->query);
 386         }
 387         // If tokenLength is zero, the return value depends on the
 388         // particular library implementation (it may or may not be a null pointer).
 389         if(tokenLength)
 390         {
 391             obsNode->token = (CAToken_t)OICMalloc(tokenLength);
 392             VERIFY_NON_NULL (obsNode->token);
 393             memcpy(obsNode->token, token, tokenLength);
 394         }
 395         obsNode->tokenLength = tokenLength;
 396
 397         obsNode->devAddr = *devAddr;
 398         obsNode->resource = resHandle;
 399
 400         LL_APPEND (serverObsList, obsNode);
 401
 402         return OC_STACK_OK;
 403     }
 404
 405 exit:
 406     if (obsNode)
 407     {
 408         OICFree(obsNode->resUri);
 409         OICFree(obsNode->query);
 410         OICFree(obsNode);
 411     }
 412     return OC_STACK_NO_MEMORY;
 413 }
 414
 415 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
 416 {
 417     ResourceObserver *out = NULL;
 418
 419     if (observeId)
 420     {
 421         LL_FOREACH (serverObsList, out)
 422         {
 423             if (out->observeId == observeId)
 424             {
 425                 return out;
 426             }
 427         }
 428     }
 429     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 430     return NULL;
 431 }
 432
 433 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
 434 {
 435     ResourceObserver *out = NULL;
 436
 437     if(token && *token)
 438     {
 439         OC_LOG(INFO, TAG,PCF("Looking for token"));
 440         OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 441         OC_LOG(INFO, TAG,PCF("\tFound token:"));
 442
 443         LL_FOREACH (serverObsList, out)
 444         {
 445             OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 446             if((memcmp(out->token, token, tokenLength) == 0))
 447             {
 448                 return out;
 449             }
 450         }
 451     }
 452     else
 453     {
 454         OC_LOG(ERROR, TAG,PCF("Passed in NULL token"));
 455     }
 456
 457     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 458     return NULL;
 459 }
 460
 461 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
 462 {
 463     if(!token || !*token)
 464     {
 465         return OC_STACK_INVALID_PARAM;
 466     }
 467
 468     ResourceObserver *obsNode = NULL;
 469
 470     obsNode = GetObserverUsingToken (token, tokenLength);
 471     if (obsNode)
 472     {
 473         OC_LOG_V(INFO, TAG, "deleting observer id  %u with token", obsNode->observeId);
 474         OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
 475         LL_DELETE (serverObsList, obsNode);
 476         OICFree(obsNode->resUri);
 477         OICFree(obsNode->query);
 478         OICFree(obsNode->token);
 479         OICFree(obsNode);
 480     }
 481     // it is ok if we did not find the observer...
 482     return OC_STACK_OK;
 483 }
 484
 485 void DeleteObserverList()
 486 {
 487     ResourceObserver *out = NULL;
 488     ResourceObserver *tmp = NULL;
 489     LL_FOREACH_SAFE (serverObsList, out, tmp)
 490     {
 491         if(out)
 492         {
 493             DeleteObserverUsingToken ((out->token), out->tokenLength);
 494         }
 495     }
 496     serverObsList = NULL;
 497 }
 498
 499 /*
 500  * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
 501  * option, which breaks the protocol abstraction requirement between RI & CA, and
 502  * has to be fixed in the future. The function below adds the header option for observe.
 503  * It should be noted that the observe header option is assumed to be the first option
 504  * in the list of user defined header options and hence it is inserted at the front
 505  * of the header options list and number of options adjusted accordingly.
 506  */
 507 OCStackResult
 508 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
 509                            OCHeaderOption *ocHdrOpt,
 510                            uint8_t numOptions,
 511                            uint8_t observeFlag)
 512 {
 513     if(!caHdrOpt)
 514     {
 515         return OC_STACK_INVALID_PARAM;
 516     }
 517
 518     CAHeaderOption_t *tmpHdrOpt = NULL;
 519
 520     tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
 521     if (NULL == tmpHdrOpt)
 522     {
 523         return OC_STACK_NO_MEMORY;
 524     }
 525     tmpHdrOpt[0].protocolID = CA_COAP_ID;
 526     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
 527     tmpHdrOpt[0].optionLength = sizeof(uint32_t);
 528     tmpHdrOpt[0].optionData[0] = observeFlag;
 529     for (uint8_t i = 0; i < numOptions; i++)
 530     {
 531         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
 532     }
 533
 534     *caHdrOpt = tmpHdrOpt;
 535     return OC_STACK_OK;
 536 }
 537
 538 /*
 539  * CA layer passes observe information to the RI layer as a header option, which
 540  * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
 541  * in the future. The function below removes the observe header option and processes it.
 542  * It should be noted that the observe header option is always assumed to be the first
 543  * option in the list of user defined header options and hence it is deleted from the
 544  * front of the header options list and the number of options is adjusted accordingly.
 545  */
 546 OCStackResult
 547 GetObserveHeaderOption (uint32_t * observationOption,
 548                         CAHeaderOption_t *options,
 549                         uint8_t * numOptions)
 550 {
 551     if(!observationOption)
 552     {
 553         return OC_STACK_INVALID_PARAM;
 554     }
 555     *observationOption = OC_OBSERVE_NO_OPTION;
 556
 557     if(!options || !numOptions)
 558     {
 559         return OC_STACK_INVALID_PARAM;
 560     }
 561
 562     for(uint8_t i = 0; i < *numOptions; i++)
 563     {
 564         if(options[i].protocolID == CA_COAP_ID &&
 565                 options[i].optionID == COAP_OPTION_OBSERVE)
 566         {
 567             *observationOption = options[i].optionData[0];
 568             for(uint8_t c = i; c < *numOptions-1; c++)
 569             {
 570                 options[i] = options[i+1];
 571             }
 572             (*numOptions)--;
 573             return OC_STACK_OK;
 574         }
 575     }
 576     return OC_STACK_OK;
 577 }
 578