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
 227                 // Clear server request on error case
 228                 if (!OCResultToSuccess(EntityHandlerCodeToOCStackCode(ehResult)))
 229                 {
 230                     FindAndDeleteServerRequest(request);
 231                 }
 232                 // Reset Observer TTL.
 233                 observer->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 234             }
 235             OCPayloadDestroy(ehRequest.payload);
 236         }
 237     }
 238
 239     return result;
 240 }
 241
 242 #ifdef WITH_PRESENCE
 243 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 244         OCPresenceTrigger trigger, OCResourceType *resourceType, OCQualityOfService qos)
 245 #else
 246 OCStackResult SendAllObserverNotification (OCMethod method, OCResource *resPtr, uint32_t maxAge,
 247         OCQualityOfService qos)
 248 #endif
 249 {
 250     OIC_LOG(INFO, TAG, "Entering SendObserverNotification");
 251     if (!resPtr)
 252     {
 253         return OC_STACK_INVALID_PARAM;
 254     }
 255
 256     OCStackResult result = OC_STACK_ERROR;
 257     ResourceObserver * resourceObserver = NULL;
 258     uint8_t numObs = 0;
 259     OCServerRequest * request = NULL;
 260     bool observeErrorFlag = false;
 261
 262     // Find clients that are observing this resource
 263     oc_mutex_lock(g_serverObsListMutex);
 264     resourceObserver = g_serverObsList;
 265     while (resourceObserver)
 266     {
 267         if (resourceObserver->resource == resPtr)
 268         {
 269             numObs++;
 270 #ifdef WITH_PRESENCE
 271             if (method != OC_REST_PRESENCE)
 272             {
 273 #endif
 274                 qos = DetermineObserverQoS(method, resourceObserver, qos);
 275                 result = SendObserveNotification(resourceObserver, qos);
 276 #ifdef WITH_PRESENCE
 277             }
 278             else
 279             {
 280                 OCEntityHandlerResponse ehResponse = {0};
 281
 282                 //This is effectively the implementation for the presence entity handler.
 283                 OIC_LOG(DEBUG, TAG, "This notification is for Presence");
 284                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 285                         0, resPtr->sequenceNum, qos, resourceObserver->query,
 286                         NULL, NULL,
 287                         resourceObserver->token, resourceObserver->tokenLength,
 288                         resourceObserver->resUri, 0, resourceObserver->acceptFormat,
 289                         &resourceObserver->devAddr);
 290
 291                 if (result == OC_STACK_OK)
 292                 {
 293                     OCPresencePayload* presenceResBuf = OCPresencePayloadCreate(
 294                             resPtr->sequenceNum, maxAge, trigger,
 295                             resourceType ? resourceType->resourcetypename : NULL);
 296
 297                     if (!presenceResBuf)
 298                     {
 299                         oc_mutex_unlock(g_serverObsListMutex);
 300                         return OC_STACK_NO_MEMORY;
 301                     }
 302
 303                     if (result == OC_STACK_OK)
 304                     {
 305                         ehResponse.ehResult = OC_EH_OK;
 306                         ehResponse.payload = (OCPayload*)presenceResBuf;
 307                         ehResponse.persistentBufferFlag = 0;
 308                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
 309                         ehResponse.resourceHandle = (OCResourceHandle) resPtr;
 310                         OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
 311                                 resourceObserver->resUri);
 312                         result = OCDoResponse(&ehResponse);
 313                         if (result != OC_STACK_OK)
 314                         {
 315                             OIC_LOG(ERROR, TAG, "Failed to send presence notification!");
 316                             FindAndDeleteServerRequest(request);
 317                         }
 318                     }
 319
 320                     OCPresencePayloadDestroy(presenceResBuf);
 321                 }
 322             }
 323 #endif
 324
 325             // Since we are in a loop, set an error flag to indicate at least one error occurred.
 326             if (result != OC_STACK_OK)
 327             {
 328                 observeErrorFlag = true;
 329             }
 330         }
 331         resourceObserver = resourceObserver->next;
 332     }
 333
 334     oc_mutex_unlock(g_serverObsListMutex);
 335
 336     if (numObs == 0)
 337     {
 338         OIC_LOG(INFO, TAG, "Resource has no observers");
 339         result = OC_STACK_NO_OBSERVERS;
 340     }
 341     else if (observeErrorFlag)
 342     {
 343         OIC_LOG(ERROR, TAG, "Observer notification error");
 344         result = OC_STACK_ERROR;
 345     }
 346     return result;
 347 }
 348
 349 OCStackResult SendListObserverNotification (OCResource * resource,
 350         OCObservationId  *obsIdList, uint8_t numberOfIds,
 351         const OCRepPayload *payload,
 352         uint32_t maxAge,
 353         OCQualityOfService qos)
 354 {
 355     (void)maxAge;
 356     if (!resource || !obsIdList || !payload)
 357     {
 358         return OC_STACK_INVALID_PARAM;
 359     }
 360
 361     uint8_t numIds = numberOfIds;
 362     ResourceObserver *observer = NULL;
 363     uint8_t numSentNotification = 0;
 364     OCServerRequest * request = NULL;
 365     OCStackResult result = OC_STACK_ERROR;
 366     bool observeErrorFlag = false;
 367
 368     OIC_LOG(INFO, TAG, "Entering SendListObserverNotification");
 369     while(numIds)
 370     {
 371         oc_mutex_lock(g_serverObsListMutex);
 372         observer = GetObserverUsingIdAsOwner (*obsIdList);
 373         if (observer)
 374         {
 375             // Found observer - verify if it matches the resource handle
 376             if (observer->resource == resource)
 377             {
 378                 qos = DetermineObserverQoS(OC_REST_GET, observer, qos);
 379
 380
 381                 result = AddServerRequest(&request, 0, 0, 1, OC_REST_GET,
 382                         0, resource->sequenceNum, qos, observer->query,
 383                         NULL, NULL, observer->token, observer->tokenLength,
 384                         observer->resUri, 0, observer->acceptFormat,
 385                         &observer->devAddr);
 386
 387                 if (request)
 388                 {
 389                     request->observeResult = OC_STACK_OK;
 390                     if (result == OC_STACK_OK)
 391                     {
 392                         OCEntityHandlerResponse ehResponse = {0};
 393                         ehResponse.ehResult = OC_EH_OK;
 394                         ehResponse.payload = (OCPayload*)OCRepPayloadCreate();
 395                         if (!ehResponse.payload)
 396                         {
 397                             FindAndDeleteServerRequest(request);
 398                             oc_mutex_unlock(g_serverObsListMutex);
 399                             continue;
 400                         }
 401                         memcpy(ehResponse.payload, payload, sizeof(*payload));
 402                         ehResponse.persistentBufferFlag = 0;
 403                         ehResponse.requestHandle = (OCRequestHandle) request->requestId;
 404                         ehResponse.resourceHandle = (OCResourceHandle) resource;
 405                         result = OCDoResponse(&ehResponse);
 406                         if (result == OC_STACK_OK)
 407                         {
 408                             OIC_LOG_V(INFO, TAG, "Observer id %d notified.", *obsIdList);
 409
 410                             // Increment only if OCDoResponse is successful
 411                             numSentNotification++;
 412                         }
 413                         else
 414                         {
 415                             OIC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
 416                             FindAndDeleteServerRequest(request);
 417                         }
 418
 419                         // Reset Observer TTL.
 420                         observer->TTL =
 421                                 GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 422
 423                         OICFree(ehResponse.payload);
 424                     }
 425                     else
 426                     {
 427                         FindAndDeleteServerRequest(request);
 428                     }
 429                 }
 430
 431                 // Since we are in a loop, set an error flag to indicate
 432                 // at least one error occurred.
 433                 if (result != OC_STACK_OK)
 434                 {
 435                     observeErrorFlag = true;
 436                 }
 437             }
 438         }
 439
 440         oc_mutex_unlock(g_serverObsListMutex);
 441         obsIdList++;
 442         numIds--;
 443     }
 444
 445     if (numSentNotification == numberOfIds && !observeErrorFlag)
 446     {
 447         return OC_STACK_OK;
 448     }
 449     else if (numSentNotification == 0)
 450     {
 451         return OC_STACK_NO_OBSERVERS;
 452     }
 453     else
 454     {
 455         OIC_LOG(ERROR, TAG, "Observer notification error");
 456         return OC_STACK_ERROR;
 457     }
 458 }
 459
 460 OCStackResult GenerateObserverId (OCObservationId *observationId)
 461 {
 462     bool found = false;
 463
 464     OIC_LOG(INFO, TAG, "Entering GenerateObserverId");
 465     VERIFY_NON_NULL (observationId);
 466
 467     do
 468     {
 469         do
 470         {
 471             *observationId = OCGetRandomByte();
 472         } while (0 == *observationId); //Make sure *observationId is not 0
 473         // Check if observation Id already exists
 474         found = IsObserverAvailable (*observationId);
 475     } while (found);
 476
 477     OIC_LOG_V(INFO, TAG, "GeneratedObservation ID is %u", *observationId);
 478
 479     return OC_STACK_OK;
 480 exit:
 481     return OC_STACK_ERROR;
 482 }
 483
 484 OCStackResult AddObserver (const char         *resUri,
 485                            const char         *query,
 486                            OCObservationId    obsId,
 487                            CAToken_t          token,
 488                            uint8_t            tokenLength,
 489                            OCResource         *resHandle,
 490                            OCQualityOfService qos,
 491                            OCPayloadFormat    acceptFormat,
 492                            const OCDevAddr    *devAddr)
 493 {
 494     // Check if resource exists and is observable.
 495     if (!resHandle)
 496     {
 497         return OC_STACK_INVALID_PARAM;
 498     }
 499     if (!(resHandle->resourceProperties & OC_OBSERVABLE))
 500     {
 501         return OC_STACK_RESOURCE_ERROR;
 502     }
 503
 504     if (!resUri || !token)
 505     {
 506         return OC_STACK_INVALID_PARAM;
 507     }
 508
 509     ResourceObserver *obsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
 510     if (obsNode)
 511     {
 512         obsNode->observeId = obsId;
 513
 514         obsNode->resUri = OICStrdup(resUri);
 515         VERIFY_NON_NULL (obsNode->resUri);
 516
 517         obsNode->qos = qos;
 518         obsNode->acceptFormat = acceptFormat;
 519         if (query)
 520         {
 521             obsNode->query = OICStrdup(query);
 522             VERIFY_NON_NULL (obsNode->query);
 523         }
 524         // If tokenLength is zero, the return value depends on the
 525         // particular library implementation (it may or may not be a null pointer).
 526         if (tokenLength)
 527         {
 528             obsNode->token = (CAToken_t)OICMalloc(tokenLength);
 529             VERIFY_NON_NULL (obsNode->token);
 530             memcpy(obsNode->token, token, tokenLength);
 531         }
 532         obsNode->tokenLength = tokenLength;
 533
 534         obsNode->devAddr = *devAddr;
 535         obsNode->resource = resHandle;
 536
 537 #ifdef WITH_PRESENCE
 538         if ((strcmp(resUri, OC_RSRVD_PRESENCE_URI) == 0))
 539         {
 540             obsNode->TTL = 0;
 541         }
 542         else
 543 #endif
 544         {
 545             obsNode->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
 546         }
 547
 548         oc_mutex_lock(g_serverObsListMutex);
 549         LL_APPEND (g_serverObsList, obsNode);
 550         oc_mutex_unlock(g_serverObsListMutex);
 551
 552         return OC_STACK_OK;
 553     }
 554
 555 exit:
 556     if (obsNode)
 557     {
 558         OICFree(obsNode->resUri);
 559         OICFree(obsNode->query);
 560         OICFree(obsNode);
 561     }
 562     return OC_STACK_NO_MEMORY;
 563 }
 564
 565 /*
 566  * This function checks if the node is past its time to live and
 567  * deletes it if timed-out. Calling this function with a  presence callback
 568  * with ttl set to 0 will not delete anything as presence nodes have
 569  * their own mechanisms for timeouts. A null argument will cause the function to
 570  * silently return.
 571  */
 572 static void CheckTimedOutObserver(ResourceObserver* observer)
 573 {
 574     if (!observer || observer->TTL == 0)
 575     {
 576         return;
 577     }
 578
 579     coap_tick_t now;
 580     coap_ticks(&now);
 581
 582     if (observer->TTL < now)
 583     {
 584         // Send confirmable notification message to observer.
 585         OIC_LOG(INFO, TAG, "Sending High-QoS notification to observer");
 586         SendObserveNotification(observer, OC_HIGH_QOS);
 587     }
 588 }
 589
 590 ResourceObserver* GetObserverUsingId (const OCObservationId observeId)
 591 {
 592     ResourceObserver *out = NULL;
 593
 594     if (observeId)
 595     {
 596         oc_mutex_lock(g_serverObsListMutex);
 597         LL_FOREACH (g_serverObsList, out)
 598         {
 599             if (out->observeId == observeId)
 600             {
 601                 OIC_LOG(INFO, TAG, "Found in observer list");
 602                 ResourceObserver *observer = CloneObserverNode(out);
 603                 oc_mutex_unlock(g_serverObsListMutex);
 604                 return observer;
 605             }
 606             CheckTimedOutObserver(out);
 607         }
 608         oc_mutex_unlock(g_serverObsListMutex);
 609     }
 610     OIC_LOG(INFO, TAG, "Observer node not found!!");
 611     return NULL;
 612 }
 613
 614 static ResourceObserver* GetObserverUsingIdAsOwner (const OCObservationId observeId)
 615 {
 616     ResourceObserver *out = NULL;
 617
 618     if (observeId)
 619     {
 620         LL_FOREACH (g_serverObsList, out)
 621         {
 622             if (out->observeId == observeId)
 623             {
 624                 return out;
 625             }
 626             CheckTimedOutObserver(out);
 627         }
 628     }
 629     OIC_LOG(INFO, TAG, "Observer node not found!!");
 630     return NULL;
 631 }
 632
 633 bool IsObserverAvailable (const OCObservationId observeId)
 634 {
 635     ResourceObserver *out = NULL;
 636
 637     if (observeId)
 638     {
 639         oc_mutex_lock(g_serverObsListMutex);
 640         LL_FOREACH (g_serverObsList, out)
 641         {
 642             if (out->observeId == observeId)
 643             {
 644                 oc_mutex_unlock(g_serverObsListMutex);
 645                 return true;
 646             }
 647         }
 648         oc_mutex_unlock(g_serverObsListMutex);
 649     }
 650
 651     return false;
 652 }
 653
 654 ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
 655 {
 656     if (token)
 657     {
 658         OIC_LOG(INFO, TAG, "Looking for token");
 659         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 660
 661         ResourceObserver *out = NULL;
 662         oc_mutex_lock(g_serverObsListMutex);
 663         LL_FOREACH (g_serverObsList, out)
 664         {
 665             /* de-annotate below line if want to see all token in cbList */
 666             //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 667             if ((memcmp(out->token, token, tokenLength) == 0))
 668             {
 669                 OIC_LOG(INFO, TAG, "Found in observer list");
 670                 ResourceObserver *observer = CloneObserverNode(out);
 671                 oc_mutex_unlock(g_serverObsListMutex);
 672                 return observer;
 673             }
 674             CheckTimedOutObserver(out);
 675         }
 676         oc_mutex_unlock(g_serverObsListMutex);
 677     }
 678     else
 679     {
 680         OIC_LOG(ERROR, TAG, "Passed in NULL token");
 681     }
 682
 683     OIC_LOG(INFO, TAG, "Observer node not found!!");
 684     return NULL;
 685 }
 686
 687 static ResourceObserver* GetObserverUsingTokenAsOwner (const CAToken_t token, uint8_t tokenLength)
 688 {
 689     if (token)
 690     {
 691         OIC_LOG(INFO, TAG, "Looking for token");
 692         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
 693
 694         ResourceObserver *out = NULL;
 695         LL_FOREACH (g_serverObsList, out)
 696         {
 697             /* de-annotate below line if want to see all token in cbList */
 698             //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
 699             if ((memcmp(out->token, token, tokenLength) == 0))
 700             {
 701                 OIC_LOG(INFO, TAG, "Found in observer list");
 702                 return out;
 703             }
 704             CheckTimedOutObserver(out);
 705         }
 706     }
 707     else
 708     {
 709         OIC_LOG(ERROR, TAG, "Passed in NULL token");
 710     }
 711
 712     OIC_LOG(INFO, TAG, "Observer node not found!!");
 713     return NULL;
 714 }
 715
 716 OCStackResult DeleteObserverUsingToken (CAToken_t token, uint8_t tokenLength)
 717 {
 718     if (!token)
 719     {
 720         return OC_STACK_INVALID_PARAM;
 721     }
 722
 723     oc_mutex_lock(g_serverObsListMutex);
 724     ResourceObserver *obsNode = GetObserverUsingTokenAsOwner (token, tokenLength);
 725     if (obsNode)
 726     {
 727         OIC_LOG_V(INFO, TAG, "deleting observer id  %u with token", obsNode->observeId);
 728         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
 729
 730         LL_DELETE (g_serverObsList, obsNode);
 731         FreeObserver(obsNode);
 732     }
 733     oc_mutex_unlock(g_serverObsListMutex);
 734
 735     // it is ok if we did not find the observer...
 736     return OC_STACK_OK;
 737 }
 738
 739 OCStackResult DeleteObserverUsingDevAddr(const OCDevAddr *devAddr)
 740 {
 741     if (!devAddr)
 742     {
 743         return OC_STACK_INVALID_PARAM;
 744     }
 745
 746     oc_mutex_lock(g_serverObsListMutex);
 747     ResourceObserver* obsDupList = CloneObserverList(g_serverObsList);
 748     oc_mutex_unlock(g_serverObsListMutex);
 749
 750     ResourceObserver *out = NULL;
 751     ResourceObserver *tmp = NULL;
 752     LL_FOREACH_SAFE(obsDupList, out, tmp)
 753     {
 754         if (out)
 755         {
 756             if ((strcmp(out->devAddr.addr, devAddr->addr) == 0)
 757                     && out->devAddr.port == devAddr->port)
 758             {
 759                 OIC_LOG_V(INFO, TAG, "deleting observer id  %u with %s:%u",
 760                           out->observeId, out->devAddr.addr, out->devAddr.port);
 761                 OCStackFeedBack(out->token, out->tokenLength, OC_OBSERVER_NOT_INTERESTED);
 762             }
 763         }
 764     }
 765
 766     FreeObserverList(obsDupList);
 767     return OC_STACK_OK;
 768 }
 769
 770 OCStackResult DeleteObserverUsingResource(OCResource *res)
 771 {
 772     if (!res)
 773     {
 774         return OC_STACK_INVALID_PARAM;
 775     }
 776
 777     oc_mutex_lock(g_serverObsListMutex);
 778     ResourceObserver *obs = NULL;
 779     ResourceObserver *next = NULL;
 780     LL_FOREACH_SAFE(g_serverObsList, obs, next)
 781     {
 782         if (obs->resource == res)
 783         {
 784             OIC_LOG_V(INFO, TAG, "Deleting observer: id-%u, token-", obs->observeId);
 785             OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obs->token, obs->tokenLength);
 786             LL_DELETE(g_serverObsList, obs);
 787             FreeObserver(obs);
 788         }
 789     }
 790     oc_mutex_unlock(g_serverObsListMutex);
 791     return OC_STACK_OK;
 792 }
 793
 794 void DeleteObserverList()
 795 {
 796     oc_mutex_lock(g_serverObsListMutex);
 797
 798     ResourceObserver* head = g_serverObsList;
 799     ResourceObserver* del = NULL;
 800     while (NULL != head)
 801     {
 802         del = head;
 803         head = head->next;
 804
 805         OIC_LOG_V(INFO, TAG, "deleting observer id  %u with token", del->observeId);
 806         OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)del->token, del->tokenLength);
 807
 808         FreeObserver(del);
 809     }
 810
 811     g_serverObsList = NULL;
 812     oc_mutex_unlock(g_serverObsListMutex);
 813 }
 814
 815 /*
 816  * CA layer expects observe registration/de-reg/notiifcations to be passed as a header
 817  * option, which breaks the protocol abstraction requirement between RI & CA, and
 818  * has to be fixed in the future. The function below adds the header option for observe.
 819  * It should be noted that the observe header option is assumed to be the first option
 820  * in the list of user defined header options and hence it is inserted at the front
 821  * of the header options list and number of options adjusted accordingly.
 822  */
 823 OCStackResult
 824 CreateObserveHeaderOption (CAHeaderOption_t **caHdrOpt,
 825                            OCHeaderOption *ocHdrOpt,
 826                            uint8_t numOptions,
 827                            uint8_t observeFlag)
 828 {
 829     if (!caHdrOpt)
 830     {
 831         return OC_STACK_INVALID_PARAM;
 832     }
 833
 834     if (numOptions > 0 && !ocHdrOpt)
 835     {
 836         OIC_LOG (INFO, TAG, "options are NULL though number is non zero");
 837         return OC_STACK_INVALID_PARAM;
 838     }
 839
 840     CAHeaderOption_t *tmpHdrOpt = NULL;
 841
 842     tmpHdrOpt = (CAHeaderOption_t *) OICCalloc ((numOptions+1), sizeof(CAHeaderOption_t));
 843     if (NULL == tmpHdrOpt)
 844     {
 845         return OC_STACK_NO_MEMORY;
 846     }
 847     tmpHdrOpt[0].protocolID = CA_COAP_ID;
 848     tmpHdrOpt[0].optionID = COAP_OPTION_OBSERVE;
 849     tmpHdrOpt[0].optionLength = sizeof(uint8_t);
 850     tmpHdrOpt[0].optionData[0] = observeFlag;
 851     for (uint8_t i = 0; i < numOptions; i++)
 852     {
 853         memcpy (&(tmpHdrOpt[i+1]), &(ocHdrOpt[i]), sizeof(CAHeaderOption_t));
 854     }
 855
 856     *caHdrOpt = tmpHdrOpt;
 857     return OC_STACK_OK;
 858 }
 859
 860 /*
 861  * CA layer passes observe information to the RI layer as a header option, which
 862  * breaks the protocol abstraction requirement between RI & CA, and has to be fixed
 863  * in the future. The function below removes the observe header option and processes it.
 864  * It should be noted that the observe header option is always assumed to be the first
 865  * option in the list of user defined header options and hence it is deleted from the
 866  * front of the header options list and the number of options is adjusted accordingly.
 867  */
 868 OCStackResult
 869 GetObserveHeaderOption (uint32_t * observationOption,
 870                         CAHeaderOption_t *options,
 871                         uint8_t * numOptions)
 872 {
 873     if (!observationOption)
 874     {
 875         return OC_STACK_INVALID_PARAM;
 876     }
 877
 878     if (!options || !numOptions)
 879     {
 880         OIC_LOG (INFO, TAG, "No options present");
 881         return OC_STACK_OK;
 882     }
 883
 884     for(uint8_t i = 0; i < *numOptions; i++)
 885     {
 886         if (options[i].protocolID == CA_COAP_ID &&
 887                 options[i].optionID == COAP_OPTION_OBSERVE)
 888         {
 889             *observationOption = options[i].optionData[0];
 890             for(uint8_t c = i; c < *numOptions-1; c++)
 891             {
 892                 options[i] = options[i+1];
 893             }
 894             (*numOptions)--;
 895             return OC_STACK_OK;
 896         }
 897     }
 898     return OC_STACK_OK;
 899 }
 900
 901 OCStackResult InitializeObserverList()
 902 {
 903     OIC_LOG(DEBUG, TAG, "InitializeObserverList IN");
 904
 905     if (NULL == g_serverObsListMutex)
 906     {
 907         g_serverObsListMutex = oc_mutex_new();
 908     }
 909
 910     OIC_LOG(DEBUG, TAG, "InitializeObserverList OUT");
 911     return OC_STACK_OK;
 912 }
 913
 914 void TerminateObserverList()
 915 {
 916     OIC_LOG(DEBUG, TAG, "TerminateObserverList IN");
 917
 918     if (NULL != g_serverObsListMutex)
 919     {
 920         oc_mutex_free(g_serverObsListMutex);
 921         g_serverObsListMutex = NULL;
 922     }
 923
 924     OIC_LOG(DEBUG, TAG, "TerminateObserverList OUT");
 925 }
 926
 927 void FreeObserver (ResourceObserver* obsNode)
 928 {
 929     if (NULL != obsNode)
 930     {
 931         OICFree(obsNode->resUri);
 932         OICFree(obsNode->query);
 933         OICFree(obsNode->token);
 934         OICFree(obsNode);
 935     }
 936 }