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 #include "octhread.h"
  33 #include "logger.h"
  34
  35 #include <coap/utlist.h>
  36 #include <coap/pdu.h>
  37 #include <coap/coap.h>
  38
  39 // Module Name
  40 #define MOD_NAME "ocobserve"
  41
  42 #define TAG  "OIC_RI_OBSERVE"
  43
  44 #define VERIFY_NON_NULL(arg) { if (!arg) {OIC_LOG(FATAL, TAG, #arg " is NULL"); goto exit;} }
  45
  46 #define MAX_OBSERVERS 3840
  47
  48 static struct ResourceObserver * g_serverObsList = NULL;
  49 static oc_mutex g_serverObsListMutex = NULL;
  50
  51 static int observer_count = 0;
  52
  53 static ResourceObserver* GetObserverUsingIdAsOwner (const OCObservationId observeId);
  54
  55 static ResourceObserver* CloneObserverNode (ResourceObserver* observer)
  56 {
  57     ResourceObserver* dupObsNode = NULL;
  58     if (observer)
  59     {
  60         dupObsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
  61         VERIFY_NON_NULL(dupObsNode);
  62         memcpy(dupObsNode, observer, sizeof(ResourceObserver));
  63
  64         if (observer->resUri)
  65         {
  66             dupObsNode->resUri = OICStrdup(observer->resUri);
  67             VERIFY_NON_NULL(dupObsNode->resUri);
  68         }
  69
  70         if (observer->query)
  71         {
  72             dupObsNode->query = OICStrdup(observer->query);
  73             VERIFY_NON_NULL(dupObsNode->query);
  74         }
  75
  76         if (observer->token)
  77         {
  78             dupObsNode->token = (CAToken_t)OICMalloc(observer->tokenLength);
  79             VERIFY_NON_NULL(dupObsNode->token);
  80             memcpy(dupObsNode->token, observer->token, observer->tokenLength);
  81         }
  82
  83         dupObsNode->next = NULL;
  84     }
  85
  86     return dupObsNode;
  87
  88 exit:
  89     FreeObserver(dupObsNode);
  90     return NULL;
  91 }
  92
  93 static void FreeObserverList (ResourceObserver* list)
  94 {
  95     ResourceObserver* head = list;
  96     ResourceObserver* del = NULL;
  97     while (NULL != head)
  98     {
  99         del = head;
 100         head = head->next;
 101
 102         OICFree(del->resUri);
 103         OICFree(del->query);
 104         OICFree(del->token);
 105         OICFree(del);
 106     }
 107 }
 108
 109 static ResourceObserver* CloneObserverList (ResourceObserver* obsList)
 110 {
 111     ResourceObserver* dupList = NULL;
 112     ResourceObserver* out = NULL;
 113
 114     LL_FOREACH(obsList, out)
 115     {
 116         ResourceObserver *obsNode = CloneObserverNode(out);
 117         if (NULL == obsNode)
 118         {
 119             FreeObserverList(dupList);
 120             dupList = NULL;
 121             break;
 122         }
 123
 124         LL_APPEND(dupList, obsNode);
 125     }
 126
 127     return dupList;
 128 }
 129
 130 /**
 131  * Determine observe QOS based on the QOS of the request.
 132  * The qos passed as a parameter overrides what the client requested.
 133  * If we want the client preference taking high priority make:
 134  *     qos = resourceObserver->qos;
 135  *
 136  * @param method RESTful method.
 137  * @param resourceObserver Observer.
 138  * @param appQoS Quality of service.
 139  * @return The quality of service of the observer.
 140  */
 141 static OCQualityOfService DetermineObserverQoS(OCMethod method,
 142         ResourceObserver * resourceObserver, OCQualityOfService appQoS)
 143 {
 144     if (!resourceObserver)
 145     {
 146         OIC_LOG(ERROR, TAG, "DetermineObserverQoS called with invalid resourceObserver");
 147         return OC_NA_QOS;
 148     }
 149
 150     OCQualityOfService decidedQoS = appQoS;
 151     if (appQoS == OC_NA_QOS)
 152     {
 153         decidedQoS = resourceObserver->qos;
 154     }
 155
 156     if (appQoS != OC_HIGH_QOS)
 157     {
 158         OIC_LOG_V(INFO, TAG, "Current NON count for this observer is %d",
 159                 resourceObserver->lowQosCount);
 160 #ifdef WITH_PRESENCE
 161         if ((resourceObserver->forceHighQos \
 162                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT) \
 163                 && method != OC_REST_PRESENCE)
 164 #else
 165         if (resourceObserver->forceHighQos \
 166                 || resourceObserver->lowQosCount >= MAX_OBSERVER_NON_COUNT)
 167 #endif
 168         {
 169             resourceObserver->lowQosCount = 0;
 170             // at some point we have to to send CON to check on the
 171             // availability of observer
 172             OIC_LOG(INFO, TAG, "This time we are sending the  notification as High qos");
 173             decidedQoS = OC_HIGH_QOS;
 174         }
 175         else
 176         {
 177             (resourceObserver->lowQosCount)++;
 178         }
 179     }
 180     return decidedQoS;
 181 }
 182
 183 /**
 184  * Create a get request and pass to entityhandler to notify specific observer.
 185  *
 186  * @param observer Observer that need to be notified.
 187  * @param qos Quality of service of resource.
 188  *
 189  * @return ::OC_STACK_OK on success, some other value upon failure.
 190  */
 191 static OCStackResult SendObserveNotification(ResourceObserver *observer,
 192                                              OCQualityOfService qos)
 193 {
 194     OCStackResult result = OC_STACK_ERROR;
 195     OCServerRequest * request = NULL;
 196     OCEntityHandlerRequest ehRequest = {0};
 197     OCEntityHandlerResult ehResult = OC_EH_ERROR;
 198
 199     result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 200                               0, observer->resource->sequenceNum, qos,
 201                               observer->query, NULL, NULL,
 202                               observer->token, observer->tokenLength,
 203                               observer->resUri, 0, observer->acceptFormat,
 204                               &observer->devAddr);
 205
 206     if (request)
 207     {
 208         request->observeResult = OC_STACK_OK;
 209         if (result == OC_STACK_OK)
 210         {
 211             result = FormOCEntityHandlerRequest(
 212                         &ehRequest,
 213                         (OCRequestHandle) request->requestId,
 214                         request->method,
 215                         &request->devAddr,
 216                         (OCResourceHandle) observer->resource,
 217                         request->query,
 218                         PAYLOAD_TYPE_REPRESENTATION,
 219                         request->payload,
 220                         request->payloadSize,
 221                         request->numRcvdVendorSpecificHeaderOptions,
 222                         request->rcvdVendorSpecificHeaderOptions,
 223                         OC_OBSERVE_NO_OPTION,
 224                         0,
 225                         request->coapID);
 226             if (result == OC_STACK_OK)
 227             {
 228                 ehResult = observer->resource->entityHandler(OC_REQUEST_FLAG, &ehRequest,
 229                                     observer->resource->entityHandlerCallbackParam);
 230
 231                 // Clear server request on error case
 232                 if (!OCResultToSuccess(EntityHandlerCodeToOCStackCode(ehResult)))
 233                 {
 234                     FindAndDeleteServerRequest(request);
 235                 }
 236                 // Reset Observer TTL.
 237                 observer->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 238             }
 239             OCPayloadDestroy(ehRequest.payload);
 240         }
 241     }
 242
 243     return result;
 244 }
 245
 246 #ifdef WITH_PRESENCE
 247 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 248         OCPresenceTrigger trigger, OCResourceType *resourceType, OCQualityOfService qos)
 249 #else
 250 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 251         OCQualityOfService qos)
 252 #endif
 253 {
 254     OIC_LOG(INFO, TAG, "Entering SendObserverNotification");
 255     if (!resPtr)
 256     {
 257         return OC_STACK_INVALID_PARAM;
 258     }
 259
 260     OCStackResult result = OC_STACK_ERROR;
 261     ResourceObserver * resourceObserver = NULL;
 262     uint16_t numObs = 0;
 263     OCServerRequest * request = NULL;
 264     bool observeErrorFlag = false;
 265
 266     // Find clients that are observing this resource
 267     oc_mutex_lock(g_serverObsListMutex);
 268     resourceObserver = g_serverObsList;
 269     while (resourceObserver)
 270     {
 271         if (resourceObserver->resource == resPtr)
 272         {
 273             numObs++;
 274 #ifdef WITH_PRESENCE
 275             if (method != OC_REST_PRESENCE)
 276             {
 277 #endif
 278                 qos = DetermineObserverQoS(method, resourceObserver, qos);
 279                 result = SendObserveNotification(resourceObserver, qos);
 280 #ifdef WITH_PRESENCE
 281             }
 282             else
 283             {
 284                 OCEntityHandlerResponse ehResponse = {0};
 285
 286                 //This is effectively the implementation for the presence entity handler.
 287                 OIC_LOG(DEBUG, TAG, "This notification is for Presence");
 288                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 289                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 290                         NULL, NULL,
 291                         resourceObserver->token, resourceObserver->tokenLength,
 292                         resourceObserver->resUri, 0, resourceObserver->acceptFormat,
 293                         &resourceObserver->devAddr);
 294
 295                 if (result == OC_STACK_OK)
 296                 {
 297                     OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
 298                             resPtr->sequenceNum, maxAge, trigger,
 299                             resourceType ? resourceType->resourcetypename : NULL);
 300
 301                     if (!presenceResBuf)
 302                     {
 303                         oc_mutex_unlock(g_serverObsListMutex);
 304                         return OC_STACK_NO_MEMORY;
 305                     }
 306
 307                     if (result == OC_STACK_OK)
 308                     {
 309                         ehResponse.ehResult = OC_EH_OK;
 310                         ehResponse.payload = (OCPayload*)presenceResBuf;
 311                         ehResponse.persistentBufferFlag = 0;
 312                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
 313                         ehResponse.resourceHandle = (OCResourceHandle) resPtr;
 314                         OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
 315                                 resourceObserver->resUri);
 316                         result = OCDoResponse(&ehResponse);
 317                         if (result != OC_STACK_OK)
 318                         {
 319                             OIC_LOG(ERROR, TAG, "Failed to send presence notification!");
 320                             FindAndDeleteServerRequest(request);
 321                         }
 322                     }
 323
 324                     OCPresencePayloadDestroy(presenceResBuf);
 325                 }
 326             }
 327 #endif
 328
 329             // Since we are in a loop, set an error flag to indicate at least one error occurred.
 330             if (result != OC_STACK_OK)
 331             {
 332                 observeErrorFlag = true;
 333             }
 334         }
 335         resourceObserver = resourceObserver->next;
 336     }
 337
 338     oc_mutex_unlock(g_serverObsListMutex);
 339
 340     if (numObs == 0)
 341     {
 342         OIC_LOG(INFO, TAG, "Resource has no observers");
 343         result = OC_STACK_NO_OBSERVERS;
 344     }
 345     else if (observeErrorFlag)
 346     {
 347         OIC_LOG(ERROR, TAG, "Observer notification error");
 348         result = OC_STACK_ERROR;
 349     }
 350     return result;
 351 }
 352
 353 OCStackResult SendListObserverNotification (OCResource * resource,
 354         OCObservationId  *obsIdList, uint16_t numberOfIds,
 355         const OCRepPayload *payload,
 356         uint32_t maxAge,
 357         OCQualityOfService qos)
 358 {
 359     (void)maxAge;
 360     if (!resource || !obsIdList || !payload)
 361     {
 362         return OC_STACK_INVALID_PARAM;
 363     }
 364
 365     uint16_t numIds = numberOfIds;
 366     ResourceObserver *observer = NULL;
 367     uint16_t numSentNotification = 0;
 368     OCServerRequest * request = NULL;
 369     OCStackResult result = OC_STACK_ERROR;
 370     bool observeErrorFlag = false;
 371
 372     OIC_LOG(INFO, TAG, "Entering SendListObserverNotification");
 373     while(numIds)
 374     {
 375         oc_mutex_lock(g_serverObsListMutex);
 376         observer = GetObserverUsingIdAsOwner (*obsIdList);
 377         if (observer)
 378         {
 379             // Found observer - verify if it matches the resource handle
 380             if (observer->resource == resource)
 381             {
 382                 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
 383
 384
 385                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 386                         0, resource->sequenceNum, qos, observer->query,
 387                         NULL, NULL, observer->token, observer->tokenLength,
 388                         observer->resUri, 0, observer->acceptFormat,
 389                         &observer->devAddr);
 390
 391                 if (request)
 392                 {
 393                     request->observeResult = OC_STACK_OK;
 394                     if (result == OC_STACK_OK)
 395                     {
 396                         OCEntityHandlerResponse ehResponse = {0};
 397                         ehResponse.ehResult = OC_EH_OK;
 398                         ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
 399                         if (!ehResponse.payload)
 400                         {
 401                             FindAndDeleteServerRequest(request);
 402                             oc_mutex_unlock(g_serverObsListMutex);
 403                             continue;
 404                         }
 405                         memcpy(ehResponse.payload, payload, sizeof(*payload));
 406                         ehResponse.persistentBufferFlag = 0;
 407                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
 408                         ehResponse.resourceHandle = (OCResourceHandle) resource;
 409                         result = OCDoResponse(&ehResponse);
 410                         if (result == OC_STACK_OK)
 411                         {
 412                             OIC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
 413
 414                             // Increment only if OCDoResponse is successful
 415                             numSentNotification++;
 416                         }
 417                         else
 418                         {
 419                             OIC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
 420                             FindAndDeleteServerRequest(request);
 421                         }
 422
 423                         // Reset Observer TTL.
 424                         observer->TTL =
 425                                 GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 426
 427                         OICFree(ehResponse.payload);
 428                     }
 429                     else
 430                     {
 431                         FindAndDeleteServerRequest(request);
 432                     }
 433                 }
 434
 435                 // Since we are in a loop, set an error flag to indicate
 436                 // at least one error occurred.
 437                 if (result != OC_STACK_OK)
 438                 {
 439                     observeErrorFlag = true;
 440                 }
 441             }
 442         }
 443
 444         oc_mutex_unlock(g_serverObsListMutex);
 445         obsIdList++;
 446         numIds--;
 447     }
 448
 449     if (numSentNotification == numberOfIds && !observeErrorFlag)
 450     {
 451         return OC_STACK_OK;
 452     }
 453     else if (numSentNotification == 0)
 454     {
 455         return OC_STACK_NO_OBSERVERS;
 456     }
 457     else
 458     {
 459         OIC_LOG(ERROR, TAG, "Observer notification error");
 460         return OC_STACK_ERROR;
 461     }
 462 }
 463
 464 OCStackResult GenerateObserverId (OCObservationId *observationId)
 465 {
 466     bool found = false;
 467
 468     OIC_LOG(INFO, TAG, "Entering GenerateObserverId");
 469     VERIFY_NON_NULL (observationId);
 470
 471     oc_mutex_lock(g_serverObsListMutex);
 472
 473     if (observer_count < MAX_OBSERVERS)
 474     {
 475         oc_mutex_unlock(g_serverObsListMutex);
 476         do
 477         {
 478             do
 479             {
 480                 *observationId = OCGetRandomTwoByte();
 481             } while (0 == *observationId); //Make sure *observationId is not 0
 482             // Check if observation Id already exists
 483             found = IsObserverAvailable (*observationId);
 484         } while (found);
 485     OIC_LOG_V(INFO, TAG, "GeneratedObservation ID is %u", *observationId);
 486     //oc_mutex_unlock(g_serverObsListMutex);
 487
 488     return OC_STACK_OK;
 489     }
 490     else
 491     {
 492         OIC_LOG_V(ERROR, TAG, "No more observers can be added");
 493         oc_mutex_unlock(g_serverObsListMutex);
 494
 495         return OC_STACK_ERROR;
 496     }
 497 exit:
 498     return OC_STACK_ERROR;
 499 }
 500
 501 OCStackResult AddObserver (const char         *resUri,
 502                            const char         *query,
 503                            OCObservationId    obsId,
 504                            CAToken_t          token,
 505                            uint8_t            tokenLength,
 506                            OCResource         *resHandle,
 507                            OCQualityOfService qos,
 508                            OCPayloadFormat    acceptFormat,
 509                            const OCDevAddr    *devAddr)
 510 {
 511     // Check if resource exists and is observable.
 512     if (!resHandle)
 513     {
 514         return OC_STACK_INVALID_PARAM;
 515     }
 516     if (!(resHandle->resourceProperties & OC_OBSERVABLE))
 517     {
 518         return OC_STACK_RESOURCE_ERROR;
 519     }
 520
 521     if (!resUri || !token)
 522     {
 523         return OC_STACK_INVALID_PARAM;
 524     }
 525
 526     ResourceObserver *obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
 527     if (obsNode)
 528     {
 529         obsNode->observeId = obsId;
 530
 531         obsNode->resUri = OICStrdup(resUri);
 532         VERIFY_NON_NULL (obsNode->resUri);
 533
 534         obsNode->qos = qos;
 535         obsNode->acceptFormat = acceptFormat;
 536         if (query)
 537         {
 538             obsNode->query = OICStrdup(query);
 539             VERIFY_NON_NULL (obsNode->query);
 540         }
 541         // If tokenLength is zero, the return value depends on the
 542         // particular library implementation (it may or may not be a null pointer).
 543         if (tokenLength)
 544         {
 545             obsNode->token = (CAToken_t)OICMalloc(tokenLength);
 546             VERIFY_NON_NULL (obsNode->token);
 547             memcpy(obsNode->token, token, tokenLength);
 548         }
 549         obsNode->tokenLength = tokenLength;
 550
 551         obsNode->devAddr = *devAddr;
 552         obsNode->resource = resHandle;
 553
 554 #ifdef WITH_PRESENCE
 555         if ((strcmp(resUri, OC_RSRVD_PRESENCE_URI) == 0))
 556         {
 557             obsNode->TTL = 0;
 558         }
 559         else
 560 #endif
 561         {
 562             obsNode->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 563         }
 564
 565         oc_mutex_lock(g_serverObsListMutex);
 566         LL_APPEND (g_serverObsList, obsNode);
 567         observer_count++;
 568         oc_mutex_unlock(g_serverObsListMutex);
 569
 570         return OC_STACK_OK;
 571     }
 572
 573 exit:
 574     if (obsNode)
 575     {
 576         OICFree(obsNode->resUri);
 577         OICFree(obsNode->query);
 578         OICFree(obsNode);
 579     }
 580     return OC_STACK_NO_MEMORY;
 581 }
 582
 583 /*
 584  * This function checks if the node is past its time to live and
 585  * deletes it if timed-out. Calling this function with a  presence callback
 586  * with ttl set to 0 will not delete anything as presence nodes have
 587  * their own mechanisms for timeouts. A null argument will cause the function to
 588  * silently return.
 589  */
 590 static void CheckTimedOutObserver(ResourceObserver* observer)
 591 {
 592     if (!observer || observer->TTL == 0)
 593     {
 594         return;
 595     }
 596
 597     coap_tick_t now;
 598     coap_ticks(&now);
 599
 600     if (observer->TTL < now)
 601     {
 602         // Send confirmable notification message to observer.
 603         OIC_LOG(INFO, TAG, "Sending High-QoS notification to observer");
 604         SendObserveNotification(observer, OC_HIGH_QOS);
 605     }
 606 }
 607
 608 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
 609 {
 610     ResourceObserver *out = NULL;
 611
 612     if (observeId)
 613     {
 614         oc_mutex_lock(g_serverObsListMutex);
 615         LL_FOREACH (g_serverObsList, out)
 616         {
 617             if (out->observeId == observeId)
 618             {
 619                 OIC_LOG(INFO, TAG, "Found in observer list");
 620                 ResourceObserver *observer = CloneObserverNode(out);
 621                 oc_mutex_unlock(g_serverObsListMutex);
 622                 return observer;
 623             }
 624             CheckTimedOutObserver(out);
 625         }
 626         oc_mutex_unlock(g_serverObsListMutex);
 627     }
 628     OIC_LOG(INFO, TAG, "Observer node not found!!");
 629     return NULL;
 630 }
 631
 632 static ResourceObserver* GetObserverUsingIdAsOwner (const OCObservationId observeId)
 633 {
 634     ResourceObserver *out = NULL;
 635
 636     if (observeId)
 637     {
 638         LL_FOREACH (g_serverObsList, out)
 639         {
 640             if (out->observeId == observeId)
 641             {
 642                 return out;
 643             }
 644             CheckTimedOutObserver(out);
 645         }
 646     }
 647     OIC_LOG(INFO, TAG, "Observer node not found!!");
 648     return NULL;
 649 }
 650
 651 bool IsObserverAvailable (const OCObservationId observeId)
 652 {
 653     ResourceObserver *out = NULL;
 654
 655     if (observeId)
 656     {
 657         oc_mutex_lock(g_serverObsListMutex);
 658         LL_FOREACH (g_serverObsList, out)
 659         {
 660             if (out->observeId == observeId)
 661             {
 662                 oc_mutex_unlock(g_serverObsListMutex);
 663                 return true;
 664             }
 665         }
 666         oc_mutex_unlock(g_serverObsListMutex);
 667     }
 668
 669     return false;
 670 }
 671
 672 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
 673 {
 674     if (token)
 675     {
 676         OIC_LOG(INFO, TAG, "Looking for token");
 677         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 678
 679         ResourceObserver *out = NULL;
 680         oc_mutex_lock(g_serverObsListMutex);
 681         LL_FOREACH (g_serverObsList, out)
 682         {
 683             /* de-annotate below line if want to see all token in cbList */
 684             //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 685             if ((memcmp(out->token, token, tokenLength) == 0))
 686             {
 687                 OIC_LOG(INFO, TAG, "Found in observer list");
 688                 ResourceObserver *observer = CloneObserverNode(out);
 689                 oc_mutex_unlock(g_serverObsListMutex);
 690                 return observer;
 691             }
 692             CheckTimedOutObserver(out);
 693         }
 694         oc_mutex_unlock(g_serverObsListMutex);
 695     }
 696     else
 697     {
 698         OIC_LOG(ERROR, TAG, "Passed in NULL token");
 699     }
 700
 701     OIC_LOG(INFO, TAG, "Observer node not found!!");
 702     return NULL;
 703 }
 704
 705 static ResourceObserver* GetObserverUsingTokenAsOwner (const CAToken_t token, uint8_t tokenLength)
 706 {
 707     if (token)
 708     {
 709         OIC_LOG(INFO, TAG, "Looking for token");
 710         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 711
 712         ResourceObserver *out = NULL;
 713         LL_FOREACH (g_serverObsList, out)
 714         {
 715             /* de-annotate below line if want to see all token in cbList */
 716             //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 717             if ((memcmp(out->token, token, tokenLength) == 0))
 718             {
 719                 OIC_LOG(INFO, TAG, "Found in observer list");
 720                 return out;
 721             }
 722             CheckTimedOutObserver(out);
 723         }
 724     }
 725     else
 726     {
 727         OIC_LOG(ERROR, TAG, "Passed in NULL token");
 728     }
 729
 730     OIC_LOG(INFO, TAG, "Observer node not found!!");
 731     return NULL;
 732 }
 733
 734 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
 735 {
 736     if (!token)
 737     {
 738         return OC_STACK_INVALID_PARAM;
 739     }
 740
 741     oc_mutex_lock(g_serverObsListMutex);
 742     ResourceObserver *obsNode = GetObserverUsingTokenAsOwner (token, tokenLength);
 743     if (obsNode)
 744     {
 745         OIC_LOG_V(INFO, TAG, "deleting observer id  %u with token", obsNode->observeId);
 746         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
 747
 748         LL_DELETE (g_serverObsList, obsNode);
 749         observer_count--;
 750         FreeObserver(obsNode);
 751     }
 752     oc_mutex_unlock(g_serverObsListMutex);
 753
 754     // it is ok if we did not find the observer...
 755     return OC_STACK_OK;
 756 }
 757
 758 OCStackResult DeleteObserverUsingDevAddr(const OCDevAddr *devAddr)
 759 {
 760     if (!devAddr)
 761     {
 762         return OC_STACK_INVALID_PARAM;
 763     }
 764
 765     oc_mutex_lock(g_serverObsListMutex);
 766     ResourceObserver* obsDupList = CloneObserverList(g_serverObsList);
 767     oc_mutex_unlock(g_serverObsListMutex);
 768
 769     ResourceObserver *out = NULL;
 770     ResourceObserver *tmp = NULL;
 771     LL_FOREACH_SAFE(obsDupList, out, tmp)
 772     {
 773         if (out)
 774         {
 775             if ((strcmp(out->devAddr.addr, devAddr->addr) == 0)
 776                     && out->devAddr.port == devAddr->port)
 777             {
 778                 OIC_LOG_V(INFO, TAG, "deleting observer id  %u with %s:%u",
 779                           out->observeId, out->devAddr.addr, out->devAddr.port);
 780                 OCStackFeedBack(out->token, out->tokenLength, OC_OBSERVER_NOT_INTERESTED);
 781             }
 782         }
 783     }
 784
 785     FreeObserverList(obsDupList);
 786     return OC_STACK_OK;
 787 }
 788
 789 OCStackResult DeleteObserverUsingResource(OCResource *res)
 790 {
 791     if (!res)
 792     {
 793         return OC_STACK_INVALID_PARAM;
 794     }
 795
 796     oc_mutex_lock(g_serverObsListMutex);
 797     ResourceObserver *obs = NULL;
 798     ResourceObserver *next = NULL;
 799     LL_FOREACH_SAFE(g_serverObsList, obs, next)
 800     {
 801         if (obs->resource == res)
 802         {
 803             OIC_LOG_V(INFO, TAG, "Deleting observer: id-%u, token-", obs->observeId);
 804             OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obs->token, obs->tokenLength);
 805             LL_DELETE(g_serverObsList, obs);
 806             FreeObserver(obs);
 807         }
 808     }
 809     oc_mutex_unlock(g_serverObsListMutex);
 810     return OC_STACK_OK;
 811 }
 812
 813 void DeleteObserverList()
 814 {
 815     oc_mutex_lock(g_serverObsListMutex);
 816
 817     ResourceObserver* head = g_serverObsList;
 818     ResourceObserver* del = NULL;
 819     while (NULL != head)
 820     {
 821         del = head;
 822         head = head->next;
 823
 824         OIC_LOG_V(INFO, TAG, "deleting observer id  %u with token", del->observeId);
 825         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)del->token, del->tokenLength);
 826
 827         FreeObserver(del);
 828     }
 829
 830     g_serverObsList = NULL;
 831     oc_mutex_unlock(g_serverObsListMutex);
 832 }
 833
 834 /*
 835  * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
 836  * option, which breaks the protocol abstraction requirement between RI & CA, and
 837  * has to be fixed in the future. The function below adds the header option for observe.
 838  * It should be noted that the observe header option is assumed to be the first option
 839  * in the list of user defined header options and hence it is inserted at the front
 840  * of the header options list and number of options adjusted accordingly.
 841  */
 842 OCStackResult
 843 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
 844                            OCHeaderOption *ocHdrOpt,
 845                            uint8_t numOptions,
 846                            uint8_t observeFlag)
 847 {
 848     if (!caHdrOpt)
 849     {
 850         return OC_STACK_INVALID_PARAM;
 851     }
 852
 853     if (numOptions > 0 && !ocHdrOpt)
 854     {
 855         OIC_LOG (INFO, TAG, "options are NULL though number is non zero");
 856         return OC_STACK_INVALID_PARAM;
 857     }
 858
 859     CAHeaderOption_t *tmpHdrOpt = NULL;
 860
 861     tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
 862     if (NULL == tmpHdrOpt)
 863     {
 864         return OC_STACK_NO_MEMORY;
 865     }
 866     tmpHdrOpt[0].protocolID = CA_COAP_ID;
 867     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
 868     tmpHdrOpt[0].optionLength = sizeof(uint8_t);
 869     tmpHdrOpt[0].optionData[0] = observeFlag;
 870     for (uint8_t i = 0; i < numOptions; i++)
 871     {
 872         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
 873     }
 874
 875     *caHdrOpt = tmpHdrOpt;
 876     return OC_STACK_OK;
 877 }
 878
 879 /*
 880  * CA layer passes observe information to the RI layer as a header option, which
 881  * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
 882  * in the future. The function below removes the observe header option and processes it.
 883  * It should be noted that the observe header option is always assumed to be the first
 884  * option in the list of user defined header options and hence it is deleted from the
 885  * front of the header options list and the number of options is adjusted accordingly.
 886  */
 887 OCStackResult
 888 GetObserveHeaderOption (uint32_t * observationOption,
 889                         CAHeaderOption_t *options,
 890                         uint8_t * numOptions)
 891 {
 892     if (!observationOption)
 893     {
 894         return OC_STACK_INVALID_PARAM;
 895     }
 896
 897     if (!options || !numOptions)
 898     {
 899         OIC_LOG (INFO, TAG, "No options present");
 900         return OC_STACK_OK;
 901     }
 902
 903     for(uint8_t i = 0; i < *numOptions; i++)
 904     {
 905         if (options[i].protocolID == CA_COAP_ID &&
 906                 options[i].optionID == COAP_OPTION_OBSERVE)
 907         {
 908             *observationOption = options[i].optionData[0];
 909             for(uint8_t c = i; c < *numOptions-1; c++)
 910             {
 911                 options[i] = options[i+1];
 912             }
 913             (*numOptions)--;
 914             return OC_STACK_OK;
 915         }
 916     }
 917     return OC_STACK_OK;
 918 }
 919
 920 OCStackResult InitializeObserverList()
 921 {
 922     OIC_LOG(DEBUG, TAG, "InitializeObserverList IN");
 923
 924     if (NULL == g_serverObsListMutex)
 925     {
 926         g_serverObsListMutex = oc_mutex_new();
 927     }
 928
 929     OIC_LOG(DEBUG, TAG, "InitializeObserverList OUT");
 930     return OC_STACK_OK;
 931 }
 932
 933 void TerminateObserverList()
 934 {
 935     OIC_LOG(DEBUG, TAG, "TerminateObserverList IN");
 936
 937     if (NULL != g_serverObsListMutex)
 938     {
 939         oc_mutex_free(g_serverObsListMutex);
 940         g_serverObsListMutex = NULL;
 941     }
 942
 943     OIC_LOG(DEBUG, TAG, "TerminateObserverList OUT");
 944 }
 945
 946 void FreeObserver (ResourceObserver* obsNode)
 947 {
 948     if (NULL != obsNode)
 949     {
 950         OICFree(obsNode->resUri);
 951         OICFree(obsNode->query);
 952         OICFree(obsNode->token);
 953         OICFree(obsNode);
 954     }
 955 }