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