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