#include "oic_string.h"
#include "ocpayload.h"
#include "ocserverrequest.h"
+#include "octhread.h"
#include "logger.h"
#include <coap/utlist.h>
#define VERIFY_NON_NULL(arg) { if (!arg) {OIC_LOG(FATAL, TAG, #arg " is NULL"); goto exit;} }
+#define MAX_OBSERVERS 3840
+
static struct ResourceObserver * g_serverObsList = NULL;
+static oc_mutex g_serverObsListMutex = NULL;
+
+static int observer_count = 0;
+
+static ResourceObserver* GetObserverUsingIdAsOwner (const OCObservationId observeId);
+
+static ResourceObserver* CloneObserverNode (ResourceObserver* observer)
+{
+ ResourceObserver* dupObsNode = NULL;
+ if (observer)
+ {
+ dupObsNode = (ResourceObserver *) OICCalloc(1, sizeof(ResourceObserver));
+ VERIFY_NON_NULL(dupObsNode);
+ memcpy(dupObsNode, observer, sizeof(ResourceObserver));
+
+ if (observer->resUri)
+ {
+ dupObsNode->resUri = OICStrdup(observer->resUri);
+ VERIFY_NON_NULL(dupObsNode->resUri);
+ }
+
+ if (observer->query)
+ {
+ dupObsNode->query = OICStrdup(observer->query);
+ VERIFY_NON_NULL(dupObsNode->query);
+ }
+
+ if (observer->token)
+ {
+ dupObsNode->token = (CAToken_t)OICMalloc(observer->tokenLength);
+ VERIFY_NON_NULL(dupObsNode->token);
+ memcpy(dupObsNode->token, observer->token, observer->tokenLength);
+ }
+
+ dupObsNode->next = NULL;
+ }
+
+ return dupObsNode;
+
+exit:
+ FreeObserver(dupObsNode);
+ return NULL;
+}
+
+static void FreeObserverList (ResourceObserver* list)
+{
+ ResourceObserver* head = list;
+ ResourceObserver* del = NULL;
+ while (NULL != head)
+ {
+ del = head;
+ head = head->next;
+
+ OICFree(del->resUri);
+ OICFree(del->query);
+ OICFree(del->token);
+ OICFree(del);
+ }
+}
+
+static ResourceObserver* CloneObserverList (ResourceObserver* obsList)
+{
+ ResourceObserver* dupList = NULL;
+ ResourceObserver* out = NULL;
+
+ LL_FOREACH(obsList, out)
+ {
+ ResourceObserver *obsNode = CloneObserverNode(out);
+ if (NULL == obsNode)
+ {
+ FreeObserverList(dupList);
+ dupList = NULL;
+ break;
+ }
+
+ LL_APPEND(dupList, obsNode);
+ }
+
+ return dupList;
+}
+
/**
* Determine observe QOS based on the QOS of the request.
* The qos passed as a parameter overrides what the client requested.
{
ehResult = observer->resource->entityHandler(OC_REQUEST_FLAG, &ehRequest,
observer->resource->entityHandlerCallbackParam);
- if (ehResult == OC_EH_ERROR)
+
+ // Clear server request on error case
+ if (!OCResultToSuccess(EntityHandlerCodeToOCStackCode(ehResult)))
{
FindAndDeleteServerRequest(request);
}
}
OCStackResult result = OC_STACK_ERROR;
- ResourceObserver * resourceObserver = g_serverObsList;
- uint8_t numObs = 0;
+ ResourceObserver * resourceObserver = NULL;
+ uint16_t numObs = 0;
OCServerRequest * request = NULL;
- OCEntityHandlerRequest ehRequest = {0};
- OCEntityHandlerResult ehResult = OC_EH_ERROR;
bool observeErrorFlag = false;
// Find clients that are observing this resource
+ oc_mutex_lock(g_serverObsListMutex);
+ resourceObserver = g_serverObsList;
while (resourceObserver)
{
if (resourceObserver->resource == resPtr)
if (!presenceResBuf)
{
+ oc_mutex_unlock(g_serverObsListMutex);
return OC_STACK_NO_MEMORY;
}
OICStrcpy(ehResponse.resourceUri, sizeof(ehResponse.resourceUri),
resourceObserver->resUri);
result = OCDoResponse(&ehResponse);
+ if (result != OC_STACK_OK)
+ {
+ OIC_LOG(ERROR, TAG, "Failed to send presence notification!");
+ FindAndDeleteServerRequest(request);
+ }
}
OCPresencePayloadDestroy(presenceResBuf);
resourceObserver = resourceObserver->next;
}
+ oc_mutex_unlock(g_serverObsListMutex);
+
if (numObs == 0)
{
OIC_LOG(INFO, TAG, "Resource has no observers");
}
OCStackResult SendListObserverNotification (OCResource * resource,
- OCObservationId *obsIdList, uint8_t numberOfIds,
+ OCObservationId *obsIdList, uint16_t numberOfIds,
const OCRepPayload *payload,
uint32_t maxAge,
OCQualityOfService qos)
return OC_STACK_INVALID_PARAM;
}
- uint8_t numIds = numberOfIds;
+ uint16_t numIds = numberOfIds;
ResourceObserver *observer = NULL;
- uint8_t numSentNotification = 0;
+ uint16_t numSentNotification = 0;
OCServerRequest * request = NULL;
OCStackResult result = OC_STACK_ERROR;
bool observeErrorFlag = false;
OIC_LOG(INFO, TAG, "Entering SendListObserverNotification");
while(numIds)
{
- observer = GetObserverUsingId (*obsIdList);
+ oc_mutex_lock(g_serverObsListMutex);
+ observer = GetObserverUsingIdAsOwner (*obsIdList);
if (observer)
{
// Found observer - verify if it matches the resource handle
if (!ehResponse.payload)
{
FindAndDeleteServerRequest(request);
+ oc_mutex_unlock(g_serverObsListMutex);
continue;
}
memcpy(ehResponse.payload, payload, sizeof(*payload));
// Increment only if OCDoResponse is successful
numSentNotification++;
-
- OICFree(ehResponse.payload);
- FindAndDeleteServerRequest(request);
}
else
{
OIC_LOG_V(INFO, TAG, "Error notifying observer id %d.", *obsIdList);
+ FindAndDeleteServerRequest(request);
}
+
// Reset Observer TTL.
observer->TTL =
GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
+
+ OICFree(ehResponse.payload);
}
else
{
FindAndDeleteServerRequest(request);
}
}
+
// Since we are in a loop, set an error flag to indicate
// at least one error occurred.
if (result != OC_STACK_OK)
}
}
}
+
+ oc_mutex_unlock(g_serverObsListMutex);
obsIdList++;
numIds--;
}
OCStackResult GenerateObserverId (OCObservationId *observationId)
{
- ResourceObserver *resObs = NULL;
+ bool found = false;
OIC_LOG(INFO, TAG, "Entering GenerateObserverId");
VERIFY_NON_NULL (observationId);
- do
- {
- *observationId = OCGetRandomByte();
- // Check if observation Id already exists
- resObs = GetObserverUsingId (*observationId);
- } while (NULL != resObs);
+ oc_mutex_lock(g_serverObsListMutex);
+ if (observer_count < MAX_OBSERVERS)
+ {
+ oc_mutex_unlock(g_serverObsListMutex);
+ do
+ {
+ do
+ {
+ *observationId = OCGetRandomTwoByte();
+ } while (0 == *observationId); //Make sure *observationId is not 0
+ // Check if observation Id already exists
+ found = IsObserverAvailable (*observationId);
+ } while (found);
OIC_LOG_V(INFO, TAG, "GeneratedObservation ID is %u", *observationId);
+ //oc_mutex_unlock(g_serverObsListMutex);
return OC_STACK_OK;
+ }
+ else
+ {
+ OIC_LOG_V(ERROR, TAG, "No more observers can be added");
+ oc_mutex_unlock(g_serverObsListMutex);
+
+ return OC_STACK_ERROR;
+ }
exit:
return OC_STACK_ERROR;
}
obsNode->devAddr = *devAddr;
obsNode->resource = resHandle;
+#ifdef WITH_PRESENCE
if ((strcmp(resUri, OC_RSRVD_PRESENCE_URI) == 0))
{
obsNode->TTL = 0;
}
else
+#endif
{
obsNode->TTL = GetTicks(MAX_OBSERVER_TTL_SECONDS * MILLISECONDS_PER_SECOND);
}
+ oc_mutex_lock(g_serverObsListMutex);
LL_APPEND (g_serverObsList, obsNode);
+ observer_count++;
+ oc_mutex_unlock(g_serverObsListMutex);
return OC_STACK_OK;
}
if (observeId)
{
+ oc_mutex_lock(g_serverObsListMutex);
+ LL_FOREACH (g_serverObsList, out)
+ {
+ if (out->observeId == observeId)
+ {
+ OIC_LOG(INFO, TAG, "Found in observer list");
+ ResourceObserver *observer = CloneObserverNode(out);
+ oc_mutex_unlock(g_serverObsListMutex);
+ return observer;
+ }
+ CheckTimedOutObserver(out);
+ }
+ oc_mutex_unlock(g_serverObsListMutex);
+ }
+ OIC_LOG(INFO, TAG, "Observer node not found!!");
+ return NULL;
+}
+
+static ResourceObserver* GetObserverUsingIdAsOwner (const OCObservationId observeId)
+{
+ ResourceObserver *out = NULL;
+
+ if (observeId)
+ {
LL_FOREACH (g_serverObsList, out)
{
if (out->observeId == observeId)
return NULL;
}
+bool IsObserverAvailable (const OCObservationId observeId)
+{
+ ResourceObserver *out = NULL;
+
+ if (observeId)
+ {
+ oc_mutex_lock(g_serverObsListMutex);
+ LL_FOREACH (g_serverObsList, out)
+ {
+ if (out->observeId == observeId)
+ {
+ oc_mutex_unlock(g_serverObsListMutex);
+ return true;
+ }
+ }
+ oc_mutex_unlock(g_serverObsListMutex);
+ }
+
+ return false;
+}
+
ResourceObserver* GetObserverUsingToken (const CAToken_t token, uint8_t tokenLength)
{
if (token)
OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
ResourceObserver *out = NULL;
+ oc_mutex_lock(g_serverObsListMutex);
+ LL_FOREACH (g_serverObsList, out)
+ {
+ /* de-annotate below line if want to see all token in cbList */
+ //OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)out->token, tokenLength);
+ if ((memcmp(out->token, token, tokenLength) == 0))
+ {
+ OIC_LOG(INFO, TAG, "Found in observer list");
+ ResourceObserver *observer = CloneObserverNode(out);
+ oc_mutex_unlock(g_serverObsListMutex);
+ return observer;
+ }
+ CheckTimedOutObserver(out);
+ }
+ oc_mutex_unlock(g_serverObsListMutex);
+ }
+ else
+ {
+ OIC_LOG(ERROR, TAG, "Passed in NULL token");
+ }
+
+ OIC_LOG(INFO, TAG, "Observer node not found!!");
+ return NULL;
+}
+
+static ResourceObserver* GetObserverUsingTokenAsOwner (const CAToken_t token, uint8_t tokenLength)
+{
+ if (token)
+ {
+ OIC_LOG(INFO, TAG, "Looking for token");
+ OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)token, tokenLength);
+
+ ResourceObserver *out = NULL;
LL_FOREACH (g_serverObsList, out)
{
/* de-annotate below line if want to see all token in cbList */
return OC_STACK_INVALID_PARAM;
}
- ResourceObserver *obsNode = GetObserverUsingToken (token, tokenLength);
+ oc_mutex_lock(g_serverObsListMutex);
+ ResourceObserver *obsNode = GetObserverUsingTokenAsOwner (token, tokenLength);
if (obsNode)
{
OIC_LOG_V(INFO, TAG, "deleting observer id %u with token", obsNode->observeId);
OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obsNode->token, tokenLength);
+
LL_DELETE (g_serverObsList, obsNode);
- OICFree(obsNode->resUri);
- OICFree(obsNode->query);
- OICFree(obsNode->token);
- OICFree(obsNode);
+ observer_count--;
+ FreeObserver(obsNode);
}
+ oc_mutex_unlock(g_serverObsListMutex);
+
// it is ok if we did not find the observer...
return OC_STACK_OK;
}
return OC_STACK_INVALID_PARAM;
}
+ oc_mutex_lock(g_serverObsListMutex);
+ ResourceObserver* obsDupList = CloneObserverList(g_serverObsList);
+ oc_mutex_unlock(g_serverObsListMutex);
+
ResourceObserver *out = NULL;
ResourceObserver *tmp = NULL;
- LL_FOREACH_SAFE(g_serverObsList, out, tmp)
+ LL_FOREACH_SAFE(obsDupList, out, tmp)
{
if (out)
{
}
}
+ FreeObserverList(obsDupList);
return OC_STACK_OK;
}
-void DeleteObserverList()
+OCStackResult DeleteObserverUsingResource(OCResource *res)
{
- ResourceObserver *out = NULL;
- ResourceObserver *tmp = NULL;
- LL_FOREACH_SAFE (g_serverObsList, out, tmp)
+ if (!res)
{
- if (out)
+ return OC_STACK_INVALID_PARAM;
+ }
+
+ oc_mutex_lock(g_serverObsListMutex);
+ ResourceObserver *obs = NULL;
+ ResourceObserver *next = NULL;
+ LL_FOREACH_SAFE(g_serverObsList, obs, next)
+ {
+ if (obs->resource == res)
{
- DeleteObserverUsingToken ((out->token), out->tokenLength);
+ OIC_LOG_V(INFO, TAG, "Deleting observer: id-%u, token-", obs->observeId);
+ OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)obs->token, obs->tokenLength);
+ LL_DELETE(g_serverObsList, obs);
+ FreeObserver(obs);
}
}
+ oc_mutex_unlock(g_serverObsListMutex);
+ return OC_STACK_OK;
+}
+
+void DeleteObserverList()
+{
+ oc_mutex_lock(g_serverObsListMutex);
+
+ ResourceObserver* head = g_serverObsList;
+ ResourceObserver* del = NULL;
+ while (NULL != head)
+ {
+ del = head;
+ head = head->next;
+
+ OIC_LOG_V(INFO, TAG, "deleting observer id %u with token", del->observeId);
+ OIC_LOG_BUFFER(INFO, TAG, (const uint8_t *)del->token, del->tokenLength);
+
+ FreeObserver(del);
+ }
+
g_serverObsList = NULL;
+ oc_mutex_unlock(g_serverObsListMutex);
}
/*
return OC_STACK_OK;
}
+OCStackResult InitializeObserverList()
+{
+ OIC_LOG(DEBUG, TAG, "InitializeObserverList IN");
+
+ if (NULL == g_serverObsListMutex)
+ {
+ g_serverObsListMutex = oc_mutex_new();
+ }
+
+ OIC_LOG(DEBUG, TAG, "InitializeObserverList OUT");
+ return OC_STACK_OK;
+}
+
+void TerminateObserverList()
+{
+ OIC_LOG(DEBUG, TAG, "TerminateObserverList IN");
+
+ if (NULL != g_serverObsListMutex)
+ {
+ oc_mutex_free(g_serverObsListMutex);
+ g_serverObsListMutex = NULL;
+ }
+
+ OIC_LOG(DEBUG, TAG, "TerminateObserverList OUT");
+}
+
+void FreeObserver (ResourceObserver* obsNode)
+{
+ if (NULL != obsNode)
+ {
+ OICFree(obsNode->resUri);
+ OICFree(obsNode->query);
+ OICFree(obsNode->token);
+ OICFree(obsNode);
+ }
+}
projects / platform / upstream / iotivity.git / blobdiff