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