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