X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;f=service%2Fnotification%2Fsrc%2Fconsumer%2FNSConsumerInternalTaskController.c;h=40a419240b6ea660677ed72501deca19c8410a7e;hb=refs%2Ftags%2Fsubmit%2Ftizen_4.0%2F20171010.021147;hp=49129b4ef1244ea3dc4eff11c5e4298ef67334c5;hpb=5d0c9be426113ad25185853ac72593b582717382;p=platform%2Fupstream%2Fiotivity.git

diff --git a/service/notification/src/consumer/NSConsumerInternalTaskController.c b/service/notification/src/consumer/NSConsumerInternalTaskController.c
index 49129b4..40a4192 100644
--- a/service/notification/src/consumer/NSConsumerInternalTaskController.c
+++ b/service/notification/src/consumer/NSConsumerInternalTaskController.c
@@ -26,16 +26,36 @@
 #include "oic_malloc.h"
 #include "oic_string.h"
 
-NSCacheList ** NSGetMessageCacheList()
+#define NS_RESERVED_MESSAGEID 10
+
+// MessageState storage structure
+typedef struct _NSMessageStateLL
 {
-    static NSCacheList * messageCache = NULL;
-    return & messageCache;
-}
+    uint64_t messageId;
+    NSSyncType state;
+    struct _NSMessageStateLL * next;
 
-void NSSetMessageCacheList(NSCacheList * cache)
+} NSMessageStateLL;
+
+typedef struct
 {
-    *(NSGetMessageCacheList()) = cache;
-}
+    NSMessageStateLL * head;
+    NSMessageStateLL * tail;
+
+} NSMessageStateList;
+
+// Mutex of MessageState storage
+pthread_mutex_t ** NSGetMessageListMutex();
+void NSLockMessageListMutex();
+void NSUnlockMessageListMutex();
+
+// Function for MessageState
+NSMessageStateList * NSGetMessageStateList();
+NSMessageStateLL * NSFindMessageState(uint64_t msgId);
+bool NSUpdateMessageState(uint64_t msgId, NSSyncType state);
+bool NSDeleteMessageState(uint64_t msgId);
+bool NSInsertMessageState(uint64_t msgId, NSSyncType state);
+void NSDestroyMessageStateList();
 
 NSCacheList ** NSGetProviderCacheList()
 {
@@ -48,47 +68,19 @@ void NSSetProviderCacheList(NSCacheList * cache)
     *(NSGetProviderCacheList()) = cache;
 }
 
-void NSDestroyMessageCacheList()
-{
-    NSCacheList * cache = *(NSGetMessageCacheList());
-    if (cache)
-    {
-        NSStorageDestroy(cache);
-    }
-
-    NSSetMessageCacheList(NULL);
-}
-
-void NSDestroyProviderCacheList()
+void NSDestroyInternalCachedList()
 {
     NSCacheList * cache = *(NSGetProviderCacheList());
     if (cache)
     {
-        NSStorageDestroy(cache);
+        NSConsumerStorageDestroy(cache);
     }
 
     NSSetProviderCacheList(NULL);
-}
-
-NSMessage * NSMessageCacheFind(const char * messageId)
-{
-    NS_VERIFY_NOT_NULL(messageId, NULL);
-
-    NSCacheList * MessageCache = *(NSGetMessageCacheList());
-    if (!MessageCache)
-    {
-        NS_LOG(DEBUG, "Message Cache Init");
-        MessageCache = NSStorageCreate();
-        NS_VERIFY_NOT_NULL(MessageCache, NULL);
-
-        MessageCache->cacheType = NS_CONSUMER_CACHE_MESSAGE;
-        NSSetMessageCacheList(MessageCache);
-    }
 
-    NSCacheElement * cacheElement = NSStorageRead(MessageCache, messageId);
-    NS_VERIFY_NOT_NULL(cacheElement, NULL);
-
-    return NSCopyMessage(((NSStoreMessage *) cacheElement->data)->msg);
+    NSDestroyMessageStateList();
+    pthread_mutex_destroy(*NSGetMessageListMutex());
+    NSOICFree(*NSGetMessageListMutex());
 }
 
 NSProvider_internal * NSProviderCacheFind(const char * providerId)
@@ -99,62 +91,36 @@ NSProvider_internal * NSProviderCacheFind(const char * providerId)
     if (!ProviderCache)
     {
         NS_LOG(DEBUG, "Provider Cache Init");
-        ProviderCache = NSStorageCreate();
+        ProviderCache = NSConsumerStorageCreate();
         NS_VERIFY_NOT_NULL(ProviderCache, NULL);
 
         ProviderCache->cacheType = NS_CONSUMER_CACHE_PROVIDER;
         NSSetProviderCacheList(ProviderCache);
     }
 
-    NSCacheElement * cacheElement = NSStorageRead(ProviderCache, providerId);
+    NSCacheElement * cacheElement = NSConsumerStorageRead(ProviderCache, providerId);
     NS_VERIFY_NOT_NULL(cacheElement, NULL);
 
-    return NSCopyProvider((NSProvider_internal *) cacheElement->data);
-}
-
-void NSRemoveCacheElementMessage(NSCacheElement * obj)
-{
-    NSRemoveMessage(((NSStoreMessage *)obj->data)->msg);
-    NSOICFree(obj->data);
-    NSOICFree(obj);
+    return NSCopyProvider_internal((NSProvider_internal *) cacheElement->data);
 }
 
-NSResult NSMessageCacheUpdate(NSMessage * msg, NSSyncType type)
+NSProvider_internal * NSFindProviderFromAddr(OCDevAddr * addr)
 {
-    NS_VERIFY_NOT_NULL(msg, NS_ERROR);
+    NS_VERIFY_NOT_NULL(addr, NULL);
 
-    NSCacheList * MessageCache = *(NSGetMessageCacheList());
-    if (!MessageCache)
+    NSCacheList * ProviderCache = *(NSGetProviderCacheList());
+    if (!ProviderCache)
     {
-        NS_LOG(DEBUG, "Message Cache Init");
-        MessageCache = NSStorageCreate();
-        NS_VERIFY_NOT_NULL(MessageCache, NS_ERROR);
-
-        MessageCache->cacheType = NS_CONSUMER_CACHE_MESSAGE;
-        NSSetMessageCacheList(MessageCache);
+        NS_LOG(DEBUG, "Provider Cache does not intialized.");
+        return NULL;
     }
 
-    NSStoreMessage * sMsg = (NSStoreMessage *)OICMalloc(sizeof(NSStoreMessage));
-    NS_VERIFY_NOT_NULL(sMsg, NS_ERROR);
-
-    NSCacheElement * obj = (NSCacheElement *)OICMalloc(sizeof(NSCacheElement));
-    NS_VERIFY_NOT_NULL_WITH_POST_CLEANING(obj, NS_ERROR, NSOICFree(sMsg));
-
-    sMsg->status = type;
-    sMsg->msg = NSCopyMessage(msg);
-
-    obj->data = (NSCacheData *) sMsg;
-    obj->next = NULL;
-
-    NS_LOG(DEBUG, "try to write to storage");
-    NSResult ret = NSStorageWrite(MessageCache, obj);
-    NS_VERIFY_NOT_NULL_WITH_POST_CLEANING(ret == NS_OK ? (void *) 1 : NULL,
-            NS_ERROR, NSRemoveCacheElementMessage(obj));
+    NSCacheElement * cacheElement =
+            NSGetProviderFromAddr(ProviderCache, addr->addr, addr->port);
 
-    NSRemoveCacheElementMessage(obj);
+    NS_VERIFY_NOT_NULL(cacheElement, NULL);
 
-    NS_LOG(DEBUG, "Update message done");
-    return NS_OK;
+    return NSCopyProvider_internal((NSProvider_internal *) cacheElement->data);
 }
 
 NSResult NSProviderCacheUpdate(NSProvider_internal * provider)
@@ -163,7 +129,7 @@ NSResult NSProviderCacheUpdate(NSProvider_internal * provider)
     if (!ProviderCache)
     {
         NS_LOG(DEBUG, "Provider Cache Init");
-        ProviderCache = NSStorageCreate();
+        ProviderCache = NSConsumerStorageCreate();
         NS_VERIFY_NOT_NULL(ProviderCache, NS_ERROR);
 
         ProviderCache->cacheType = NS_CONSUMER_CACHE_PROVIDER;
@@ -179,10 +145,12 @@ NSResult NSProviderCacheUpdate(NSProvider_internal * provider)
     obj->next = NULL;
 
     NS_LOG(DEBUG, "try to write to storage");
-    NSResult ret = NSStorageWrite(ProviderCache, obj);
+    NSResult ret = NSConsumerStorageWrite(ProviderCache, obj);
     NS_VERIFY_NOT_NULL_WITH_POST_CLEANING(ret == NS_OK ? (void *) 1 : NULL,
             NS_ERROR, NSOICFree(obj));
 
+    NSOICFree(obj);
+
     return NS_OK;
 }
 
@@ -192,21 +160,29 @@ void NSCancelAllSubscription()
     if (!ProviderCache)
     {
         NS_LOG(DEBUG, "Provider Cache Init");
-        ProviderCache = NSStorageCreate();
+        ProviderCache = NSConsumerStorageCreate();
         NS_VERIFY_NOT_NULL_V(ProviderCache);
 
         ProviderCache->cacheType = NS_CONSUMER_CACHE_PROVIDER;
         NSSetProviderCacheList(ProviderCache);
     }
 
-    NSCacheElement * obj = NULL;
-    while ((obj = NSPopProviderCacheList(ProviderCache)))
+    NSCacheElement * obj = NSPopProviderCacheList(ProviderCache);
+    while (obj)
     {
         NS_LOG(DEBUG, "build NSTask");
-        NSTask * task = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, (void *) obj->data);
+        NSProvider * prov = NSCopyProvider((NSProvider_internal *) obj->data);
+        NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(prov,
+                    NSRemoveProvider_internal((void *) obj->data));
+
+        NSTask * task = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, prov);
         NS_VERIFY_NOT_NULL_V(task);
 
         NSConsumerPushEvent(task);
+        NSRemoveProvider_internal((void *) obj->data);
+        NSOICFree(obj);
+
+        obj = NSPopProviderCacheList(ProviderCache);
     }
 }
 
@@ -216,6 +192,33 @@ void NSConsumerHandleProviderDiscovered(NSProvider_internal * provider)
 
     bool isAdded = true;
     bool isSubscribing = false;
+
+    NSProvider_internal * providerCacheDataFromAddr
+        = NSFindProviderFromAddr(provider->connection->addr);
+
+    if (providerCacheDataFromAddr)
+    {
+        if (!strcmp(providerCacheDataFromAddr->providerId, provider->providerId))
+        {
+            NSProviderConnectionInfo * infos = providerCacheDataFromAddr->connection;
+            while (infos)
+            {
+                isSubscribing |= infos->isSubscribing;
+                infos = infos->next;
+            }
+
+            if (isSubscribing == false)
+            {
+                NSProvider * providerForCb = NSCopyProvider(providerCacheDataFromAddr);
+                NSProviderChanged(providerForCb, NS_DISCOVERED);
+                NSRemoveProvider(providerForCb);
+            }
+            NSRemoveProvider_internal(providerCacheDataFromAddr);
+            return ;
+        }
+        NSRemoveProvider_internal(providerCacheDataFromAddr);
+    }
+
     NSProvider_internal * providerCacheData = NSProviderCacheFind(provider->providerId);
 
     if (providerCacheData == NULL)
@@ -232,7 +235,7 @@ void NSConsumerHandleProviderDiscovered(NSProvider_internal * provider)
             isSubscribing |= infos->isSubscribing;
             if (infos->addr->adapter == newAdapter && infos->isSubscribing == true)
             {
-                NS_LOG_V(DEBUG, "This provider already discovered : %s:%d",
+                NS_LOG_V(INFO_PRIVATE, "This provider already discovered : %s:%d",
                          infos->addr->addr, infos->addr->port);
                 NS_LOG_V(DEBUG, "Subscription : %d", infos->isSubscribing);
                 return;
@@ -242,7 +245,8 @@ void NSConsumerHandleProviderDiscovered(NSProvider_internal * provider)
     }
 
     NSResult ret = NSProviderCacheUpdate(provider);
-    NS_VERIFY_NOT_NULL_V(ret == NS_OK ? (void *) 1 : NULL);
+    NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(ret == NS_OK ? (void *) 1 : NULL,
+            NSRemoveProvider_internal(providerCacheData));
 
     if (isAdded == false)
     {
@@ -257,19 +261,22 @@ void NSConsumerHandleProviderDiscovered(NSProvider_internal * provider)
     if (provider->accessPolicy == NS_SELECTION_CONSUMER && isSubscribing == false)
     {
         NS_LOG(DEBUG, "accepter is NS_ACCEPTER_CONSUMER, Callback to user");
-        NSDiscoveredProvider((NSProvider *) provider);
+        NSProvider * prov = NSCopyProvider(provider);
+        NSProviderChanged(prov, NS_DISCOVERED);
+        NSRemoveProvider(prov);
     }
     else
     {
         NS_LOG(DEBUG, "accepter is NS_ACCEPTER_PROVIDER, request subscribe");
-        NSProvider_internal * subProvider = NSCopyProvider(provider);
+        NSProvider * subProvider = NSCopyProvider(provider);
         NSTask * task = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE, (void *) subProvider);
-        NS_VERIFY_NOT_NULL_V(task);
+        NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(task,
+                NSRemoveProvider_internal(providerCacheData));
 
         NSConsumerPushEvent(task);
     }
 
-    NSRemoveProvider(providerCacheData);
+    NSRemoveProvider_internal(providerCacheData);
 }
 
 void NSConsumerHandleProviderDeleted(NSProvider_internal * provider)
@@ -279,50 +286,96 @@ void NSConsumerHandleProviderDeleted(NSProvider_internal * provider)
     NSCacheList * providerCache = *(NSGetProviderCacheList());
     NS_VERIFY_NOT_NULL_V(providerCache);
 
-    NSResult ret = NSStorageDelete(providerCache, provider->providerId);
+    NSResult ret = NSConsumerStorageDelete(providerCache, provider->providerId);
     NS_VERIFY_NOT_NULL_V(ret == NS_OK ? (void *)1 : NULL);
+
+    NS_LOG_V(INFO_PRIVATE, "Stopped Provider : %s", provider->providerId);
+    NSProvider * prov = NSCopyProvider(provider);
+    NSProviderChanged(prov, NS_STOPPED);
+    NSRemoveProvider(prov);
 }
 
-void NSConsumerHandleRecvSubscriptionConfirmed(NSMessage * msg)
+void NSConsumerHandleSubscribeSucceed(NSProvider_internal * provider)
+{
+    NS_VERIFY_NOT_NULL_V(provider);
+
+    NSCacheList * ProviderCache = *(NSGetProviderCacheList());
+
+    NSCacheElement * cacheElement = NSConsumerStorageRead(ProviderCache, provider->providerId);
+    NS_VERIFY_NOT_NULL_V(cacheElement);
+
+    pthread_mutex_t * mutex = NSGetCacheMutex();
+    pthread_mutex_lock(mutex);
+
+    NS_VERIFY_NOT_NULL_V(cacheElement);
+    NSProvider_internal * prov = (NSProvider_internal *)cacheElement->data;
+    NSProviderConnectionInfo *infos = prov->connection;
+    while(infos)
+    {
+        infos->isSubscribing = true;
+        infos = infos->next;
+    }
+
+    pthread_mutex_unlock(mutex);
+}
+
+void NSConsumerHandleRecvProviderChanged(NSMessage * msg)
 {
     NS_VERIFY_NOT_NULL_V(msg);
 
-    NS_LOG_V(DEBUG, "confirmed by : %s", msg->providerId);
-    NSProvider_internal * provider = NSProviderCacheFind(msg->providerId);
-    NS_VERIFY_NOT_NULL_V(provider);
+    NS_LOG_V(INFO_PRIVATE, "confirmed by : %s", msg->providerId);
 
-    if (provider->connection->next == NULL)
+    NSCacheList * ProviderCache = *(NSGetProviderCacheList());
+
+    NSCacheElement * cacheElement = NSConsumerStorageRead(ProviderCache, msg->providerId);
+    NS_VERIFY_NOT_NULL_V(cacheElement);
+
+    pthread_mutex_t * mutex = NSGetCacheMutex();
+    pthread_mutex_lock(mutex);
+    NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(cacheElement, pthread_mutex_unlock(mutex));
+    NSProvider_internal * provider = (NSProvider_internal *) cacheElement->data;
+    if (provider->state == (NSProviderState) msg->messageId)
     {
-        NS_LOG(DEBUG, "call back to user");
-        NSSubscriptionAccepted((NSProvider *) provider);
+        NS_LOG_V(DEBUG, "Already receive message(ALLOW/DENY) : %d", (int) msg->messageId);
+        pthread_mutex_unlock(mutex);
+        return;
     }
+
+    NS_LOG_V(DEBUG, "Provider State Changed %d -> %d",
+             (int) provider->state, (int) msg->messageId);
+    NS_LOG(DEBUG, "call back to user");
+    provider->state = (NSProviderState) msg->messageId;
+
+    NSProvider * prov = NSCopyProvider(provider);
+
+    pthread_mutex_unlock(mutex);
+    NSProviderChanged(prov, (NSProviderState) msg->messageId);
+    NSRemoveProvider(prov);
 }
 
 void NSConsumerHandleRecvMessage(NSMessage * msg)
 {
     NS_VERIFY_NOT_NULL_V(msg);
 
-    NSResult ret = NSMessageCacheUpdate(msg, NS_SYNC_UNREAD);
-    NS_VERIFY_NOT_NULL_V(ret == NS_OK ? (void *) 1 : NULL);
-
-    NSMessagePost((NSMessage *) msg);
+    if (NSInsertMessageState(msg->messageId, NS_SYNC_UNREAD))
+    {
+        NSMessagePost(msg);
+    }
 }
 
 void NSConsumerHandleRecvSyncInfo(NSSyncInfo * sync)
 {
     NS_VERIFY_NOT_NULL_V(sync);
 
-    char msgId[NS_DEVICE_ID_LENGTH] = { 0, };
-    snprintf(msgId, NS_DEVICE_ID_LENGTH, "%lld", (long long int)sync->messageId);
-
-    NSMessage * msg = NSMessageCacheFind(msgId);
-    NS_VERIFY_NOT_NULL_V(msg);
-
-    NSResult ret = NSMessageCacheUpdate(msg, sync->state);
-    NS_VERIFY_NOT_NULL_V(ret == NS_OK ? (void *) 1 : NULL);
-    NSRemoveMessage(msg);
+    if (NSUpdateMessageState(sync->messageId, sync->state))
+    {
+        NSNotificationSync(sync);
+    }
 
-    NSNotificationSync(sync);
+    if (sync->state == NS_SYNC_DELETED)
+    {
+        NSDeleteMessageState(sync->messageId);
+    }
 }
 
 void NSConsumerHandleMakeSyncInfo(NSSyncInfo * sync)
@@ -333,6 +386,7 @@ void NSConsumerHandleMakeSyncInfo(NSSyncInfo * sync)
     NS_VERIFY_NOT_NULL_V (provider);
 
     NSProviderConnectionInfo * connections = NSCopyProviderConnections(provider->connection);
+    NSRemoveProvider_internal((void *) provider);
     NS_VERIFY_NOT_NULL_V (connections);
 
     NSSyncInfo_internal * syncInfo = (NSSyncInfo_internal *)OICMalloc(sizeof(NSSyncInfo_internal));
@@ -349,6 +403,46 @@ void NSConsumerHandleMakeSyncInfo(NSSyncInfo * sync)
     NSConsumerPushEvent(syncTask);
 }
 
+void NSConsumerHandleGetTopicUri(NSMessage * msg)
+{
+    NS_VERIFY_NOT_NULL_V(msg);
+
+    NSProvider_internal * provider = NSProviderCacheFind(msg->providerId);
+    NS_VERIFY_NOT_NULL_V(provider);
+
+    NSTask * topicTask = NSMakeTask(TASK_CONSUMER_REQ_TOPIC_LIST, (void *) provider);
+    NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(topicTask, NSRemoveProvider_internal(provider));
+
+    NSConsumerPushEvent(topicTask);
+}
+
+void NSConsumerHandleRecvTopicLL(NSProvider_internal * provider)
+{
+    NS_VERIFY_NOT_NULL_V(provider);
+
+    NSRemoveConnections(provider->connection);
+    provider->connection = NULL;
+
+    NSProvider_internal * cachedProvider = NSProviderCacheFind(provider->providerId);
+    NS_VERIFY_NOT_NULL_V(cachedProvider);
+
+    if (!cachedProvider->topicLL && !provider->topicLL)
+    {
+        NS_LOG(DEBUG, "topic is null and previous status is same.");
+        NSRemoveProvider_internal(cachedProvider);
+        return;
+    }
+    NSRemoveProvider_internal(cachedProvider);
+
+    NSResult ret = NSProviderCacheUpdate(provider);
+    NS_VERIFY_NOT_NULL_V(ret == NS_OK ? (void *) 1 : NULL);
+
+    NS_LOG(DEBUG, "call back to user");
+    NSProvider * prov = NSCopyProvider(provider);
+    NSProviderChanged((NSProvider *) prov, (NSProviderState) NS_TOPIC);
+    NSRemoveProvider(prov);
+}
+
 void NSConsumerInternalTaskProcessing(NSTask * task)
 {
     NS_VERIFY_NOT_NULL_V(task);
@@ -356,10 +450,17 @@ void NSConsumerInternalTaskProcessing(NSTask * task)
     NS_LOG_V(DEBUG, "Receive Event : %d", (int)task->taskType);
     switch (task->taskType)
     {
-        case TASK_CONSUMER_RECV_SUBSCRIBE_CONFIRMED:
+        case TASK_CONSUMER_SENT_REQ_OBSERVE:
         {
-            NS_LOG(DEBUG, "Receive Subscribe confirm from provider.");
-            NSConsumerHandleRecvSubscriptionConfirmed((NSMessage *)task->taskData);
+            NS_LOG(DEBUG, "Receive Subscribe succeed from provider.");
+            NSConsumerHandleSubscribeSucceed((NSProvider_internal *)task->taskData);
+            NSRemoveProvider_internal((void *)task->taskData);
+            break;
+        }
+        case TASK_CONSUMER_RECV_PROVIDER_CHANGED:
+        {
+            NS_LOG(DEBUG, "Receive Provider Changed");
+            NSConsumerHandleRecvProviderChanged((NSMessage *)task->taskData);
             NSRemoveMessage((NSMessage *)task->taskData);
             break;
         }
@@ -374,7 +475,7 @@ void NSConsumerInternalTaskProcessing(NSTask * task)
         {
             NS_LOG(DEBUG, "Receive New Provider is discovered.");
             NSConsumerHandleProviderDiscovered((NSProvider_internal *)task->taskData);
-            NSRemoveProvider((NSProvider_internal *)task->taskData);
+            NSRemoveProvider_internal((NSProvider_internal *)task->taskData);
             break;
         }
         case TASK_RECV_SYNCINFO:
@@ -391,11 +492,25 @@ void NSConsumerInternalTaskProcessing(NSTask * task)
             NSOICFree(task->taskData);
             break;
         }
+        case TASK_CONSUMER_REQ_TOPIC_URI:
+        {
+            NS_LOG(DEBUG, "Request Topic Uri");
+            NSConsumerHandleGetTopicUri((NSMessage *)task->taskData);
+            NSRemoveMessage((NSMessage *)task->taskData);
+            break;
+        }
+        case TASK_CONSUMER_RECV_TOPIC_LIST:
+        {
+            NS_LOG(DEBUG, "Receive Topic List");
+            NSConsumerHandleRecvTopicLL((NSProvider_internal *)task->taskData);
+            NSRemoveProvider_internal((NSProvider_internal *)task->taskData);
+            break;
+        }
         case TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL:
         {
             NS_LOG(DEBUG, "Make Subscribe cancel from provider.");
             NSConsumerHandleProviderDeleted((NSProvider_internal *)task->taskData);
-            NSRemoveProvider((NSProvider_internal *)task->taskData);
+            NSRemoveProvider_internal((NSProvider_internal *)task->taskData);
             break;
         }
         default :
@@ -406,3 +521,202 @@ void NSConsumerInternalTaskProcessing(NSTask * task)
     }
     NSOICFree(task);
 }
+
+// implements of MessageState function
+pthread_mutex_t ** NSGetMessageListMutex()
+{
+    static pthread_mutex_t * g_mutex = NULL;
+    if (g_mutex == NULL)
+    {
+        g_mutex = (pthread_mutex_t *) OICMalloc(sizeof(pthread_mutex_t));
+        NS_VERIFY_NOT_NULL(g_mutex, NULL);
+
+        pthread_mutex_init(g_mutex, NULL);
+    }
+    return & g_mutex;
+}
+
+void NSLockMessageListMutex()
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    pthread_mutex_lock(*NSGetMessageListMutex());
+}
+
+void NSUnlockMessageListMutex()
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    pthread_mutex_unlock(*NSGetMessageListMutex());
+}
+
+NSMessageStateList ** NSGetMessageStateListAddr()
+{
+    static NSMessageStateList * g_messageStateList = NULL;
+    if (g_messageStateList == NULL)
+    {
+        g_messageStateList = (NSMessageStateList *)OICMalloc(sizeof(NSMessageStateList));
+        NS_VERIFY_NOT_NULL(g_messageStateList, NULL);
+
+        g_messageStateList->head = NULL;
+        g_messageStateList->tail = NULL;
+    }
+
+    return & g_messageStateList;
+}
+
+NSMessageStateList * NSGetMessageStateList()
+{
+    return * NSGetMessageStateListAddr();
+}
+
+NSMessageStateLL * NSFindMessageState(uint64_t msgId)
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    if (msgId <= NS_RESERVED_MESSAGEID)
+    {
+        return NULL;
+    }
+    NSMessageStateLL * iter = NULL;
+
+    NSLockMessageListMutex();
+    if (NSGetMessageStateList()->head == NULL)
+    {
+        NSUnlockMessageListMutex();
+        return NULL;
+    }
+
+    for (iter = NSGetMessageStateList()->head; iter; iter = iter->next)
+    {
+        if (iter->messageId == msgId)
+        {
+            NSUnlockMessageListMutex();
+            return iter;
+        }
+    }
+
+    NSUnlockMessageListMutex();
+    return NULL;
+}
+
+bool NSUpdateMessageState(uint64_t msgId, NSSyncType state)
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    if (msgId <= NS_RESERVED_MESSAGEID)
+    {
+        return false;
+    }
+    NSMessageStateLL * iter = NULL;
+
+    NSLockMessageListMutex();
+    for (iter = NSGetMessageStateList()->head; iter; iter = iter->next)
+    {
+        if (iter->messageId == msgId && state != iter->state)
+        {
+            iter->state = state;
+            NSUnlockMessageListMutex();
+            return true;
+        }
+    }
+
+    NSUnlockMessageListMutex();
+    return false;
+}
+
+bool NSDeleteMessageState(uint64_t msgId)
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    if (msgId <= NS_RESERVED_MESSAGEID)
+    {
+        return false;
+    }
+
+    NSMessageStateLL * iter = NULL;
+    NSMessageStateLL * prev = NULL;
+
+    NSLockMessageListMutex();
+    for (iter = NSGetMessageStateList()->head; iter; iter = iter->next)
+    {
+        if (iter->messageId == msgId)
+        {
+            if (iter == NSGetMessageStateList()->head)
+            {
+                NSGetMessageStateList()->head = NULL;
+                NSGetMessageStateList()->tail = NULL;
+            }
+            else if (iter == NSGetMessageStateList()->tail)
+            {
+                prev->next = NULL;
+                NSGetMessageStateList()->tail = prev;
+            }
+            else
+            {
+                prev->next = iter->next;
+            }
+            NSUnlockMessageListMutex();
+
+            NSOICFree(iter);
+            return true;
+        }
+        prev = iter;
+    }
+
+    NSUnlockMessageListMutex();
+    return false;
+}
+
+bool NSInsertMessageState(uint64_t msgId, NSSyncType state)
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    if (NSFindMessageState(msgId))
+    {
+        return false;
+    }
+
+    NSMessageStateLL * insertMsg = (NSMessageStateLL * )OICMalloc(sizeof(NSMessageStateLL));
+    NS_VERIFY_NOT_NULL(insertMsg, false);
+
+    insertMsg->messageId = msgId;
+    insertMsg->state = state;
+    insertMsg->next = NULL;
+
+    NSLockMessageListMutex();
+    if (NSGetMessageStateList()->head == NULL)
+    {
+        NSGetMessageStateList()->head = insertMsg;
+    }
+    else
+    {
+        NSGetMessageStateList()->tail->next = insertMsg;
+    }
+    NSGetMessageStateList()->tail = insertMsg;
+    NSUnlockMessageListMutex();
+
+    return true;
+}
+
+void NSDestroyMessageStateList()
+{
+    NS_LOG_V(DEBUG, "%s", __func__);
+    NSLockMessageListMutex();
+
+    NSMessageStateLL * iter = NSGetMessageStateList()->head;
+    while (iter)
+    {
+        NSMessageStateLL * del = iter;
+        iter = iter->next;
+        NSOICFree(del);
+    }
+
+    NSGetMessageStateList()->head = NULL;
+    NSGetMessageStateList()->tail = NULL;
+
+    NSUnlockMessageListMutex();
+
+    pthread_mutex_t * mu = *NSGetMessageListMutex();
+    pthread_mutex_destroy(mu);
+    NSOICFree(mu);
+    *NSGetMessageListMutex() = NULL;
+
+    NSMessageStateList * list = NSGetMessageStateList();
+    NSOICFree(list);
+    *NSGetMessageStateListAddr() = NULL;
+}