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 "occoap.h"
  28 #include "utlist.h"
  29 #include "debug.h"
  30 #include "ocrandom.h"
  31 #include "ocmalloc.h"
  32 #include "ocserverrequest.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     OCQualityOfService decidedQoS = appQoS;
  51     if(appQoS == OC_NA_QOS)
  52     {
  53         decidedQoS = resourceObserver->qos;
  54     }
  55
  56     if(appQoS != OC_HIGH_QOS)
  57     {
  58         OC_LOG_V(INFO, TAG, "Current NON count for this observer is %d",
  59                 resourceObserver->lowQosCount);
  60         #ifdef WITH_PRESENCE
  61         if((resourceObserver->forceHighQos \
  62                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT) \
  63                 && method != OC_REST_PRESENCE)
  64         #else
  65         if(resourceObserver->forceHighQos \
  66                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT)
  67         #endif
  68             {
  69             resourceObserver->lowQosCount = 0;
  70             // at some point we have to to send CON to check on the
  71             // availability of observer
  72             OC_LOG(INFO, TAG, PCF("This time we are sending the  notification as High qos"));
  73             decidedQoS = OC_HIGH_QOS;
  74             }
  75         else
  76         {
  77             (resourceObserver->lowQosCount)++;
  78         }
  79     }
  80     return decidedQoS;
  81 }
  82
  83 #ifdef WITH_PRESENCE
  84 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
  85         OCResourceType *resourceType, OCQualityOfService qos)
  86 #else
  87 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
  88         OCQualityOfService qos)
  89 #endif
  90 {
  91     OC_LOG(INFO, TAG, PCF("Entering SendObserverNotification"));
  92     OCStackResult result = OC_STACK_ERROR;
  93     ResourceObserver * resourceObserver = serverObsList;
  94     uint8_t numObs = 0;
  95     OCServerRequest * request = NULL;
  96     OCEntityHandlerRequest ehRequest = {0};
  97     OCEntityHandlerResult ehResult = OC_EH_ERROR;
  98
  99     // Find clients that are observing this resource
 100     while (resourceObserver)
 101     {
 102         if (resourceObserver->resource == resPtr)
 103         {
 104             numObs++;
 105             #ifdef WITH_PRESENCE
 106             if(method != OC_REST_PRESENCE)
 107             {
 108             #endif
 109                 qos = DetermineObserverQoS(method, resourceObserver, qos);
 110
 111 #ifdef CA_INT
 112                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 113                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 114                         NULL, NULL,
 115                         &resourceObserver->token, resourceObserver->addr,
 116                         resourceObserver->resUri, 0,
 117                         &(resourceObserver->addressInfo), resourceObserver->connectivityType);
 118 #else
 119                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 120                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 121                         NULL, NULL,
 122                         &resourceObserver->token, resourceObserver->addr,
 123                         resourceObserver->resUri, 0);
 124 #endif // CA_INT
 125
 126                 request->observeResult = OC_STACK_OK;
 127                 if(request && result == OC_STACK_OK)
 128                 {
 129                     result = FormOCEntityHandlerRequest(&ehRequest, (OCRequestHandle) request,
 130                                 request->method, (OCResourceHandle) resPtr, request->query,
 131                                 request->reqJSONPayload, request->numRcvdVendorSpecificHeaderOptions,
 132                                 request->rcvdVendorSpecificHeaderOptions, OC_OBSERVE_NO_OPTION, 0);
 133                     if(result == OC_STACK_OK)
 134                     {
 135                         ehResult = resPtr->entityHandler(OC_REQUEST_FLAG, &ehRequest);
 136                         if(ehResult == OC_EH_ERROR)
 137                         {
 138                             FindAndDeleteServerRequest(request);
 139                         }
 140                     }
 141                 }
 142             #ifdef WITH_PRESENCE
 143             }
 144             else
 145             {
 146                 OCEntityHandlerResponse ehResponse = {0};
 147                 unsigned char presenceResBuf[MAX_RESPONSE_LENGTH] = {0};
 148                 //This is effectively the implementation for the presence entity handler.
 149                 OC_LOG(DEBUG, TAG, PCF("This notification is for Presence"));
 150 #ifdef CA_INT
 151                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 152                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 153                         NULL, NULL,
 154                         &resourceObserver->token, resourceObserver->addr,
 155                         resourceObserver->resUri, 0,
 156                         &(resourceObserver->addressInfo), resourceObserver->connectivityType);
 157
 158 #else
 159                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 160                         0, OC_OBSERVE_NO_OPTION, OC_LOW_QOS,
 161                         NULL, NULL, NULL, &resourceObserver->token,
 162                         resourceObserver->addr, resourceObserver->resUri, 0);
 163 #endif
 164                 if(result == OC_STACK_OK)
 165                 {
 166                     // we create the payload here
 167                     if(resourceType && resourceType->resourcetypename)
 168                     {
 169                         snprintf((char *)presenceResBuf, sizeof(presenceResBuf), "%u:%u:%s",
 170                                 resPtr->sequenceNum, maxAge, resourceType->resourcetypename);
 171                     }
 172                     else
 173                     {
 174                         snprintf((char *)presenceResBuf, sizeof(presenceResBuf), "%u:%u",
 175                                 resPtr->sequenceNum, maxAge);
 176                     }
 177                     ehResponse.ehResult = OC_EH_OK;
 178                     ehResponse.payload = presenceResBuf;
 179                     ehResponse.payloadSize = strlen((const char *)presenceResBuf) + 1;
 180                     ehResponse.persistentBufferFlag = 0;
 181                     ehResponse.requestHandle = (OCRequestHandle) request;
 182                     ehResponse.resourceHandle = (OCResourceHandle) resPtr;
 183                     strcpy((char *)ehResponse.resourceUri, (const char *)resourceObserver->resUri);
 184                     result = OCDoResponse(&ehResponse);
 185                 }
 186             }
 187             #endif
 188         }
 189         resourceObserver = resourceObserver->next;
 190     }
 191     if (numObs == 0)
 192     {
 193         OC_LOG(INFO, TAG, PCF("Resource has no observers"));
 194         result = OC_STACK_NO_OBSERVERS;
 195     }
 196     return result;
 197 }
 198
 199 OCStackResult SendListObserverNotification (OCResource * resource,
 200         OCObservationId  *obsIdList, uint8_t numberOfIds,
 201         unsigned char *notificationJSONPayload, uint32_t maxAge,
 202         OCQualityOfService qos)
 203 {
 204     uint8_t numIds = numberOfIds;
 205     ResourceObserver *observation = NULL;
 206     uint8_t numSentNotification = 0;
 207     OCServerRequest * request = NULL;
 208     OCStackResult result = OC_STACK_ERROR;
 209     OCEntityHandlerResponse ehResponse = {0};
 210
 211     OC_LOG(INFO, TAG, PCF("Entering SendListObserverNotification"));
 212     while(numIds)
 213     {
 214         OC_LOG_V(INFO, TAG, "Need to notify observation id %d", *obsIdList);
 215         observation = NULL;
 216         observation = GetObserverUsingId (*obsIdList);
 217         if(observation)
 218         {
 219             // Found observation - verify if it matches the resource handle
 220             if (observation->resource == resource)
 221             {
 222                 qos = DetermineObserverQoS(OC_REST_GET, observation, qos);
 223
 224
 225 #ifdef CA_INT
 226                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 227                         0, resource->sequenceNum, qos, observation->query,
 228                         NULL, NULL, &observation->token,
 229                         observation->addr, observation->resUri, 0,
 230                         &(observation->addressInfo), observation->connectivityType);
 231 #else
 232                 result = AddServerRequest(&request, 0, 0, 0, 1, OC_REST_GET,
 233                         0, resource->sequenceNum, qos, observation->query,
 234                         NULL, NULL, &observation->token,
 235                         observation->addr, observation->resUri, 0);
 236 #endif //CA_INT
 237
 238                 request->observeResult = OC_STACK_OK;
 239                 if(request && result == OC_STACK_OK)
 240                 {
 241                     memset(&ehResponse, 0, sizeof(OCEntityHandlerResponse));
 242                     ehResponse.ehResult = OC_EH_OK;
 243                     ehResponse.payload = (unsigned char *) OCMalloc(MAX_RESPONSE_LENGTH);
 244                     if(!ehResponse.payload)
 245                     {
 246                         FindAndDeleteServerRequest(request);
 247                         continue;
 248                     }
 249                     strcpy((char *)ehResponse.payload, (const char *)notificationJSONPayload);
 250                     ehResponse.payloadSize = strlen((const char *)ehResponse.payload) + 1;
 251                     ehResponse.persistentBufferFlag = 0;
 252                     ehResponse.requestHandle = (OCRequestHandle) request;
 253                     ehResponse.resourceHandle = (OCResourceHandle) resource;
 254                     result = OCDoResponse(&ehResponse);
 255                     if(result == OC_STACK_OK)
 256                     {
 257                         OCFree(ehResponse.payload);
 258                         FindAndDeleteServerRequest(request);
 259                     }
 260                 }
 261                 else
 262                 {
 263                     FindAndDeleteServerRequest(request);
 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 #ifdef CA_INT
 310 OCStackResult AddObserver (const char         *resUri,
 311                            const char         *query,
 312                            OCObservationId    obsId,
 313                            CAToken_t          *token,
 314                            OCDevAddr          *addr,
 315                            OCResource         *resHandle,
 316                            OCQualityOfService qos,
 317                            CAAddress_t          *addressInfo,
 318                            CAConnectivityType_t connectivityType)
 319 #else
 320 OCStackResult AddObserver (const char         *resUri,
 321                            const char         *query,
 322                            OCObservationId    obsId,
 323                            OCCoAPToken        *token,
 324                            OCDevAddr          *addr,
 325                            OCResource         *resHandle,
 326                            OCQualityOfService qos)
 327 #endif // CA_INT
 328 {
 329     ResourceObserver *obsNode = NULL;
 330
 331     obsNode = (ResourceObserver *) OCCalloc(1, sizeof(ResourceObserver));
 332     if (obsNode)
 333     {
 334         obsNode->observeId = obsId;
 335
 336         obsNode->resUri = (unsigned char *)OCMalloc(strlen(resUri)+1);
 337         VERIFY_NON_NULL (obsNode->resUri);
 338         memcpy (obsNode->resUri, resUri, strlen(resUri)+1);
 339
 340         obsNode->qos = qos;
 341         if(query)
 342         {
 343             obsNode->query = (unsigned char *)OCMalloc(strlen(query)+1);
 344             VERIFY_NON_NULL (obsNode->query);
 345             memcpy (obsNode->query, query, strlen(query)+1);
 346         }
 347
 348 #ifdef CA_INT
 349         obsNode->token = (CAToken_t)OCMalloc(CA_MAX_TOKEN_LEN+1);
 350         VERIFY_NON_NULL (obsNode->token);
 351         memset(obsNode->token, 0, CA_MAX_TOKEN_LEN + 1);
 352         memcpy(obsNode->token, *token, CA_MAX_TOKEN_LEN);
 353 #else
 354         obsNode->token.tokenLength = token->tokenLength;
 355         memcpy (obsNode->token.token, token->token, token->tokenLength);
 356 #endif // CA_INT
 357
 358         obsNode->addr = (OCDevAddr *)OCMalloc(sizeof(OCDevAddr));
 359         VERIFY_NON_NULL (obsNode->addr);
 360         memcpy (obsNode->addr, addr, sizeof(OCDevAddr));
 361 #ifdef CA_INT
 362         obsNode->addressInfo = *addressInfo;
 363         obsNode->connectivityType = connectivityType;
 364 #endif
 365         obsNode->resource = resHandle;
 366         LL_APPEND (serverObsList, obsNode);
 367         return OC_STACK_OK;
 368     }
 369
 370 exit:
 371     if (obsNode)
 372     {
 373         OCFree(obsNode->resUri);
 374         OCFree(obsNode->query);
 375         OCFree(obsNode->addr);
 376         OCFree(obsNode);
 377     }
 378     return OC_STACK_NO_MEMORY;
 379 }
 380
 381 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
 382 {
 383     ResourceObserver *out = NULL;
 384
 385     if (observeId)
 386     {
 387         LL_FOREACH (serverObsList, out)
 388         {
 389             if (out->observeId == observeId)
 390             {
 391                 return out;
 392             }
 393         }
 394     }
 395     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 396     return NULL;
 397 }
 398
 399 #ifdef CA_INT
 400 ResourceObserver* GetObserverUsingToken (const CAToken_t * token)
 401 #else
 402 ResourceObserver* GetObserverUsingToken (const OCCoAPToken * token)
 403 #endif
 404 {
 405     ResourceObserver *out = NULL;
 406
 407     if(token)
 408     {
 409         LL_FOREACH (serverObsList, out)
 410         {
 411 #ifdef CA_INT
 412             OC_LOG(INFO, TAG,PCF("comparing tokens"));
 413             OC_LOG_BUFFER(INFO, TAG, token, CA_MAX_TOKEN_LEN);
 414             OC_LOG_BUFFER(INFO, TAG, out->token, CA_MAX_TOKEN_LEN);
 415             if((memcmp(out->token, *token, CA_MAX_TOKEN_LEN) == 0))
 416             {
 417                 return out;
 418             }
 419 #else
 420             OC_LOG(INFO, TAG,PCF("comparing tokens"));
 421             OC_LOG_BUFFER(INFO, TAG, token->token, token->tokenLength);
 422             OC_LOG_BUFFER(INFO, TAG, out->token.token, out->token.tokenLength);
 423             if((out->token.tokenLength == token->tokenLength) &&
 424                (memcmp(out->token.token, token->token, token->tokenLength) == 0))
 425             {
 426                 return out;
 427             }
 428 #endif // CA_INT
 429         }
 430     }
 431     OC_LOG(INFO, TAG, PCF("Observer node not found!!"));
 432     return NULL;
 433 }
 434
 435 #ifdef CA_INT
 436 OCStackResult DeleteObserverUsingToken (CAToken_t * token)
 437 #else
 438 OCStackResult DeleteObserverUsingToken (OCCoAPToken * token)
 439 #endif
 440 {
 441     ResourceObserver *obsNode = NULL;
 442
 443     obsNode = GetObserverUsingToken (token);
 444     if (obsNode)
 445     {
 446         OC_LOG_V(INFO, TAG, PCF("deleting tokens"));
 447 #ifdef CA_INT
 448         OC_LOG_BUFFER(INFO, TAG, obsNode->token, CA_MAX_TOKEN_LEN);
 449 #else
 450         OC_LOG_BUFFER(INFO, TAG, obsNode->token.token, obsNode->token.tokenLength);
 451 #endif
 452         LL_DELETE (serverObsList, obsNode);
 453         OCFree(obsNode->resUri);
 454         OCFree(obsNode->query);
 455         OCFree(obsNode->addr);
 456         OCFree(obsNode);
 457     }
 458     // it is ok if we did not find the observer...
 459     return OC_STACK_OK;
 460 }
 461
 462 void DeleteObserverList()
 463 {
 464     ResourceObserver *out = NULL;
 465     ResourceObserver *tmp = NULL;
 466     LL_FOREACH_SAFE (serverObsList, out, tmp)
 467     {
 468         DeleteObserverUsingToken (&(out->token));
 469     }
 470     serverObsList = NULL;
 471 }
 472
 473 #ifdef CA_INT
 474 OCStackResult
 475 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
 476                            OCHeaderOption *ocHdrOpt,
 477                            uint8_t numOptions,
 478                            uint8_t observeFlag)
 479 {
 480     CAHeaderOption_t *tmpHdrOpt = NULL;
 481
 482     tmpHdrOpt = (CAHeaderOption_t *) OCMalloc ((numOptions+1)*sizeof(CAHeaderOption_t));
 483     if (NULL == tmpHdrOpt)
 484     {
 485         return OC_STACK_NO_MEMORY;
 486     }
 487     tmpHdrOpt[0].protocolID = CA_COAP_ID;
 488     // TODO-CA: COAP_OPTION_OBSERVE is defined in CoAP header files which will be abstracted
 489     // from resource model. We have to define a new macro for this in the stack layer.
 490     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
 491     tmpHdrOpt[0].optionLength = sizeof(uint32_t);
 492     tmpHdrOpt[0].optionData[0] = observeFlag;
 493     for (uint8_t i = 0; i < numOptions; i++)
 494     {
 495         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
 496     }
 497
 498     *caHdrOpt = tmpHdrOpt;
 499     return OC_STACK_OK;
 500 }
 501
 502 OCStackResult
 503 GetObserveHeaderOption (uint32_t * observationOption,
 504                         CAHeaderOption_t *options,
 505                         uint8_t * numOptions)
 506 {
 507     *observationOption = OC_OBSERVE_NO_OPTION;
 508     uint8_t i = 0;
 509     uint8_t c = 0;
 510     for(i = 0; i < *numOptions; i++)
 511     {
 512         if(options[i].protocolID == CA_COAP_ID &&
 513                 options[i].optionID == COAP_OPTION_OBSERVE)
 514         {
 515             *observationOption = options[i].optionData[0];
 516             for(c = i; c < *numOptions-1; c++)
 517             {
 518                 options[i].protocolID = options[i+1].protocolID;
 519                 options[i].optionID = options[i+1].optionID;
 520                 options[i].optionLength = options[i+1].optionLength;
 521                 memcpy(options[i].optionData, options[i+1].optionData, options[i+1].optionLength);
 522             }
 523             (*numOptions)--;
 524             return OC_STACK_OK;
 525         }
 526     }
 527     return OC_STACK_OK;
 528 }
 529 #endif // CA_INT