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(
 145                                     &ehRequest,
 146                                     (OCRequestHandle) request,
 147                                     request->method,
 148                                     &request->devAddr,
 149                                     (OCResourceHandle) resPtr,
 150                                     request->query,
 151                                     request->payload,
 152                                     request->payloadSize,
 153                                     request->numRcvdVendorSpecificHeaderOptions,
 154                                     request->rcvdVendorSpecificHeaderOptions,
 155                                     OC_OBSERVE_NO_OPTION,
 156                                     0);
 157                         if(result == OC_STACK_OK)
 158                         {
 159                             ehResult = resPtr->entityHandler(OC_REQUEST_FLAG, &ehRequest,
 160                                                 resPtr->entityHandlerCallbackParam);
 161                             if(ehResult == OC_EH_ERROR)
 162                             {
 163                                 FindAndDeleteServerRequest(request);
 164                             }
 165                         }
 166                     }
 167                 }
 168             #ifdef WITH_PRESENCE
 169             }
 170             else
 171             {
 172                 OCEntityHandlerResponse ehResponse = {};
 173
 174                 //This is effectively the implementation for the presence entity handler.
 175                 OC_LOG(DEBUG, TAG, PCF("This notification is for Presence"));
 176                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 177                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 178                         NULL, NULL,
 179                         resourceObserver->token, resourceObserver->tokenLength,
 180                         resourceObserver->resUri, 0,
 181                         &resourceObserver->devAddr);
 182
 183                 if(result == OC_STACK_OK)
 184                 {
 185                     OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
 186                             resPtr->sequenceNum, maxAge, trigger,
 187                             resourceType ? resourceType->resourcetypename : NULL);
 188
 189                     if(!presenceResBuf)
 190                     {
 191                         return OC_STACK_NO_MEMORY;
 192                     }
 193
 194                     if(result == OC_STACK_OK)
 195                     {
 196                         ehResponse.ehResult = OC_EH_OK;
 197                         ehResponse.payload = (OCPayload*)presenceResBuf;
 198                         ehResponse.persistentBufferFlag = 0;
 199                         ehResponse.requestHandle = (OCRequestHandle) request;
 200                         ehResponse.resourceHandle = (OCResourceHandle) resPtr;
 201                         OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
 202                                 resourceObserver->resUri);
 203                         result = OCDoResponse(&ehResponse);
 204                     }
 205
 206                     OCPresencePayloadDestroy(presenceResBuf);
 207                 }
 208             }
 209             #endif
 210
 211             // Since we are in a loop, set an error flag to indicate at least one error occurred.
 212             if (result != OC_STACK_OK)
 213             {
 214                 observeErrorFlag = true;
 215             }
 216         }
 217         resourceObserver = resourceObserver->next;
 218     }
 219
 220     if (numObs == 0)
 221     {
 222         OC_LOG(INFO, TAG, PCF("Resource has no observers"));
 223         result = OC_STACK_NO_OBSERVERS;
 224     }
 225     else if (observeErrorFlag)
 226     {
 227         OC_LOG(ERROR, TAG, PCF("Observer notification error"));
 228         result = OC_STACK_ERROR;
 229     }
 230     return result;
 231 }
 232
 233 OCStackResult SendListObserverNotification (OCResource * resource,
 234         OCObservationId  *obsIdList, uint8_t numberOfIds,
 235         const OCRepPayload *payload,
 236         uint32_t maxAge,
 237         OCQualityOfService qos)
 238 {
 239     if(!resource || !obsIdList || !payload)
 240     {
 241         return OC_STACK_INVALID_PARAM;
 242     }
 243
 244     uint8_t numIds = numberOfIds;
 245     ResourceObserver *observer = NULL;
 246     uint8_t numSentNotification = 0;
 247     OCServerRequest * request = NULL;
 248     OCStackResult result = OC_STACK_ERROR;
 249     bool observeErrorFlag = false;
 250
 251     OC_LOG(INFO, TAG, PCF("Entering SendListObserverNotification"));
 252     while(numIds)
 253     {
 254         observer = GetObserverUsingId (*obsIdList);
 255         if(observer)
 256         {
 257             // Found observer - verify if it matches the resource handle
 258             if (observer->resource == resource)
 259             {
 260                 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
 261
 262
 263                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 264                         0, resource->sequenceNum, qos, observer->query,
 265                         NULL, NULL, observer->token, observer->tokenLength,
 266                         observer->resUri, 0,
 267                         &observer->devAddr);
 268
 269                 if(request)
 270                 {
 271                     request->observeResult = OC_STACK_OK;
 272                     if(result == OC_STACK_OK)
 273                     {
 274                         OCEntityHandlerResponse ehResponse = {};
 275                         ehResponse.ehResult = OC_EH_OK;
 276                         ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
 277                         if(!ehResponse.payload)
 278                         {
 279                             FindAndDeleteServerRequest(request);
 280                             continue;
 281                         }
 282                         memcpy(ehResponse.payload, payload, sizeof(*payload));
 283                         ehResponse.persistentBufferFlag = 0;
 284                         ehResponse.requestHandle = (OCRequestHandle) request;
 285                         ehResponse.resourceHandle = (OCResourceHandle) resource;
 286                         result = OCDoResponse(&ehResponse);
 287                         if(result == OC_STACK_OK)
 288                         {
 289                             OC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
 290
 291                             // Increment only if OCDoResponse is successful
 292                             numSentNotification++;
 293
 294                             OICFree(ehResponse.payload);
 295                             FindAndDeleteServerRequest(request);
 296                         }
 297                         else
 298                         {
 299                             OC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
 300                         }
 301                     }
 302                     else
 303                     {
 304                         FindAndDeleteServerRequest(request);
 305                     }
 306                 }
 307                 // Since we are in a loop, set an error flag to indicate
 308                 // at least one error occurred.
 309                 if (result != OC_STACK_OK)
 310                 {
 311                     observeErrorFlag = true;
 312                 }
 313             }
 314         }
 315         obsIdList++;
 316         numIds--;
 317     }
 318
 319     if(numSentNotification == numberOfIds && !observeErrorFlag)
 320     {
 321         return OC_STACK_OK;
 322     }
 323     else if(numSentNotification == 0)
 324     {
 325         return OC_STACK_NO_OBSERVERS;
 326     }
 327     else
 328     {
 329         OC_LOG(ERROR, TAG, PCF("Observer notification error"));
 330         return OC_STACK_ERROR;
 331     }
 332 }
 333
 334 OCStackResult GenerateObserverId (OCObservationId *observationId)
 335 {
 336     ResourceObserver *resObs = NULL;
 337
 338     OC_LOG(INFO, TAG, PCF("Entering GenerateObserverId"));
 339     VERIFY_NON_NULL (observationId);
 340
 341     do
 342     {
 343         *observationId = OCGetRandomByte();
 344         // Check if observation Id already exists
 345         resObs = GetObserverUsingId (*observationId);
 346     } while (NULL != resObs);
 347
 348     OC_LOG_V(INFO, TAG, "Generated bservation ID is %u", *observationId);
 349
 350     return OC_STACK_OK;
 351 exit:
 352     return OC_STACK_ERROR;
 353 }
 354
 355 OCStackResult AddObserver (const char         *resUri,
 356                            const char         *query,
 357                            OCObservationId    obsId,
 358                            CAToken_t          token,
 359                            uint8_t            tokenLength,
 360                            OCResource         *resHandle,
 361                            OCQualityOfService qos,
 362                            const OCDevAddr    *devAddr)
 363 {
 364     // Check if resource exists and is observable.
 365     if (!resHandle)
 366     {
 367         return OC_STACK_INVALID_PARAM;
 368     }
 369     if (!(resHandle->resourceProperties & OC_OBSERVABLE))
 370     {
 371         return OC_STACK_RESOURCE_ERROR;
 372     }
 373     ResourceObserver *obsNode = NULL;
 374
 375     if(!resUri || !token || !*token)
 376     {
 377         return OC_STACK_INVALID_PARAM;
 378     }
 379
 380     obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
 381     if (obsNode)
 382     {
 383         obsNode->observeId = obsId;
 384
 385         obsNode->resUri = OICStrdup(resUri);
 386         VERIFY_NON_NULL (obsNode->resUri);
 387
 388         obsNode->qos = qos;
 389         if(query)
 390         {
 391             obsNode->query = OICStrdup(query);
 392             VERIFY_NON_NULL (obsNode->query);
 393         }
 394         // If tokenLength is zero, the return value depends on the
 395         // particular library implementation (it may or may not be a null pointer).
 396         if(tokenLength)
 397         {
 398             obsNode->token = (CAToken_t)OICMalloc(tokenLength);
 399             VERIFY_NON_NULL (obsNode->token);
 400             memcpy(obsNode->token, token, tokenLength);
 401         }
 402         obsNode->tokenLength = tokenLength;
 403
 404         obsNode->devAddr = *devAddr;
 405         obsNode->resource = resHandle;
 406
 407         LL_APPEND (serverObsList, obsNode);
 408
 409         return OC_STACK_OK;
 410     }
 411
 412 exit:
 413     if (obsNode)
 414     {
 415         OICFree(obsNode->resUri);
 416         OICFree(obsNode->query);
 417         OICFree(obsNode);
 418     }
 419     return OC_STACK_NO_MEMORY;
 420 }
 421
 422 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
 423 {
 424     ResourceObserver *out = NULL;
 425
 426     if (observeId)
 427     {
 428         LL_FOREACH (serverObsList, out)
 429         {
 430             if (out->observeId == observeId)
 431             {
 432                 return out;
 433             }
 434         }
 435     }
 436     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 437     return NULL;
 438 }
 439
 440 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
 441 {
 442     ResourceObserver *out = NULL;
 443
 444     if(token && *token)
 445     {
 446         OC_LOG(INFO, TAG,PCF("Looking for token"));
 447         OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 448         OC_LOG(INFO, TAG,PCF("\tFound token:"));
 449
 450         LL_FOREACH (serverObsList, out)
 451         {
 452             OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 453             if((memcmp(out->token, token, tokenLength) == 0))
 454             {
 455                 return out;
 456             }
 457         }
 458     }
 459     else
 460     {
 461         OC_LOG(ERROR, TAG,PCF("Passed in NULL token"));
 462     }
 463
 464     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 465     return NULL;
 466 }
 467
 468 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
 469 {
 470     if(!token || !*token)
 471     {
 472         return OC_STACK_INVALID_PARAM;
 473     }
 474
 475     ResourceObserver *obsNode = NULL;
 476
 477     obsNode = GetObserverUsingToken (token, tokenLength);
 478     if (obsNode)
 479     {
 480         OC_LOG_V(INFO, TAG, "deleting observer id  %u with token", obsNode->observeId);
 481         OC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
 482         LL_DELETE (serverObsList, obsNode);
 483         OICFree(obsNode->resUri);
 484         OICFree(obsNode->query);
 485         OICFree(obsNode->token);
 486         OICFree(obsNode);
 487     }
 488     // it is ok if we did not find the observer...
 489     return OC_STACK_OK;
 490 }
 491
 492 void DeleteObserverList()
 493 {
 494     ResourceObserver *out = NULL;
 495     ResourceObserver *tmp = NULL;
 496     LL_FOREACH_SAFE (serverObsList, out, tmp)
 497     {
 498         if(out)
 499         {
 500             DeleteObserverUsingToken ((out->token), out->tokenLength);
 501         }
 502     }
 503     serverObsList = NULL;
 504 }
 505
 506 /*
 507  * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
 508  * option, which breaks the protocol abstraction requirement between RI & CA, and
 509  * has to be fixed in the future. The function below adds the header option for observe.
 510  * It should be noted that the observe header option is assumed to be the first option
 511  * in the list of user defined header options and hence it is inserted at the front
 512  * of the header options list and number of options adjusted accordingly.
 513  */
 514 OCStackResult
 515 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
 516                            OCHeaderOption *ocHdrOpt,
 517                            uint8_t numOptions,
 518                            uint8_t observeFlag)
 519 {
 520     if(!caHdrOpt)
 521     {
 522         return OC_STACK_INVALID_PARAM;
 523     }
 524
 525     CAHeaderOption_t *tmpHdrOpt = NULL;
 526
 527     tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
 528     if (NULL == tmpHdrOpt)
 529     {
 530         return OC_STACK_NO_MEMORY;
 531     }
 532     tmpHdrOpt[0].protocolID = CA_COAP_ID;
 533     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
 534     tmpHdrOpt[0].optionLength = sizeof(uint32_t);
 535     tmpHdrOpt[0].optionData[0] = observeFlag;
 536     for (uint8_t i = 0; i < numOptions; i++)
 537     {
 538         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
 539     }
 540
 541     *caHdrOpt = tmpHdrOpt;
 542     return OC_STACK_OK;
 543 }
 544
 545 /*
 546  * CA layer passes observe information to the RI layer as a header option, which
 547  * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
 548  * in the future. The function below removes the observe header option and processes it.
 549  * It should be noted that the observe header option is always assumed to be the first
 550  * option in the list of user defined header options and hence it is deleted from the
 551  * front of the header options list and the number of options is adjusted accordingly.
 552  */
 553 OCStackResult
 554 GetObserveHeaderOption (uint32_t * observationOption,
 555                         CAHeaderOption_t *options,
 556                         uint8_t * numOptions)
 557 {
 558     if(!observationOption)
 559     {
 560         return OC_STACK_INVALID_PARAM;
 561     }
 562     *observationOption = OC_OBSERVE_NO_OPTION;
 563
 564     if(!options || !numOptions)
 565     {
 566         return OC_STACK_INVALID_PARAM;
 567     }
 568
 569     for(uint8_t i = 0; i < *numOptions; i++)
 570     {
 571         if(options[i].protocolID == CA_COAP_ID &&
 572                 options[i].optionID == COAP_OPTION_OBSERVE)
 573         {
 574             *observationOption = options[i].optionData[0];
 575             for(uint8_t c = i; c < *numOptions-1; c++)
 576             {
 577                 options[i] = options[i+1];
 578             }
 579             (*numOptions)--;
 580             return OC_STACK_OK;
 581         }
 582     }
 583     return OC_STACK_OK;
 584 }
 585