X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;ds=sidebyside;f=service%2Fnotification%2Fsrc%2Fprovider%2FNSProviderSubscription.c;h=a275660ac6d1b304bd610ff7bf708ac531b727aa;hb=c9c470b7fa5ee700e00daf3d57dce92445714060;hp=54a27c1dda5e90a8712d43c72f30fe736aa17efb;hpb=7e4bb7c179c43df48044dfbf0ecedfbb341bf6c7;p=platform%2Fupstream%2Fiotivity.git

diff --git a/service/notification/src/provider/NSProviderSubscription.c b/service/notification/src/provider/NSProviderSubscription.c
index 54a27c1..a275660 100644
--- a/service/notification/src/provider/NSProviderSubscription.c
+++ b/service/notification/src/provider/NSProviderSubscription.c
@@ -19,55 +19,50 @@
 //-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=
 
 #include "NSProviderSubscription.h"
+#include "NSProviderListener.h"
+
+NSCacheList * consumerSubList;
 
 NSResult NSInitSubscriptionList()
 {
     NS_LOG(DEBUG, "NSInitSubscriptionList - IN");
 
-    consumerSubList = NSCacheCreate();
+    consumerSubList = NSProviderStorageCreate();
+    NS_VERIFY_NOT_NULL(consumerSubList, NS_FAIL);
     consumerSubList->cacheType = NS_PROVIDER_CACHE_SUBSCRIBER;
 
     NS_LOG(DEBUG, "NSInitSubscriptionList - OUT");
     return NS_OK;
 }
 
-NSResult NSSetSubscriptionAcceptPolicy(NSAccessPolicy policy)
+NSResult NSSetSubscriptionAccessPolicy(bool policy)
 {
     NS_LOG(DEBUG, "NSSetSubscriptionAcceptPolicy - IN");
 
-    if (policy == NS_ACCEPTER_PROVIDER)
+    if (policy == NS_POLICY_PROVIDER)
     {
-        OIC_LOG(INFO, SUBSCRIPTION_TAG, "Place Provider as a subscription accepter");
         NS_LOG(DEBUG, "Place Provider as a subscription accepter");
     }
-    else if (policy == NS_ACCEPTER_CONSUMER)
+    else if (policy == NS_POLICY_CONSUMER)
     {
-        OIC_LOG(INFO, SUBSCRIPTION_TAG, "Place Consumer as a subscription accepter");
         NS_LOG(DEBUG, "Place Consumer as a subscription accepter");
     }
 
-    NSSubscriptionAccepter = policy;
+    NSSetPolicy(policy);
 
     NS_LOG(DEBUG, "NSSetSubscriptionAcceptPolicy - OUT");
     return NS_OK;
 }
 
-int NSGetSubscriptionAccepter()
-{
-    return NSSubscriptionAccepter;
-}
-
 NSResult NSSendAccessPolicyResponse(OCEntityHandlerRequest *entityHandlerRequest)
 {
-    OIC_LOG(DEBUG, NOTIFICATION_TAG, "Send Notification Policy to consumer");
     NS_LOG(DEBUG, "NSSendAccessPolicyResponse - IN");
 
     // put notification resource
-    OCResourceHandle notificationResourceHandle;
-    if (NSPutNotificationResource(NSGetSubscriptionAccepter(), &notificationResourceHandle)
+    OCResourceHandle notificationResourceHandle = NULL;
+    if (NSPutNotificationResource(NSGetPolicy(), &notificationResourceHandle)
             != NS_OK)
     {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, PCF("Failed to put notification resource"));
         NS_LOG(ERROR, "Fail to put notification resource");
         return NS_ERROR;
     }
@@ -82,15 +77,30 @@ NSResult NSSendAccessPolicyResponse(OCEntityHandlerRequest *entityHandlerRequest
     OCRepPayload* payload = OCRepPayloadCreate();
     if (!payload)
     {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, PCF("Failed to allocate Payload"));
         NS_LOG(ERROR, "payload is NULL");
         return NS_ERROR;
     }
 
-    OCRepPayloadSetUri(payload, NSGetNotificationUri());
-    OCRepPayloadSetPropInt(payload, NS_ATTRIBUTE_POLICY, NSGetSubscriptionAccepter());
-    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_MESSAGE, NSGetNotificationMessageUri());
-    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_SYNC, NSGetNotificationSyncUri());
+    NS_LOG_V(INFO_PRIVATE, "NS Provider ID: %s", NSGetProviderInfo()->providerId);
+
+    char * copyReq = OICStrdup(entityHandlerRequest->query);
+    char * reqInterface = NSGetValueFromQuery(copyReq, NS_QUERY_INTERFACE);
+
+    if (reqInterface && strcmp(reqInterface, NS_INTERFACE_BASELINE) == 0)
+    {
+        OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_BASELINE);
+        OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_READ);
+        OCResourcePayloadAddStringLL(&payload->types, NS_ROOT_TYPE);
+    }
+
+    NSOICFree(copyReq);
+    OCRepPayloadSetUri(payload, NS_ROOT_URI);
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_PROVIDER_ID, NSGetProviderInfo()->providerId);
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_VERSION, VERSION);
+    OCRepPayloadSetPropBool(payload, NS_ATTRIBUTE_POLICY, NSGetPolicy());
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_MESSAGE, NS_COLLECTION_MESSAGE_URI);
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_SYNC, NS_COLLECTION_SYNC_URI);
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_TOPIC, NS_COLLECTION_TOPIC_URI);
 
     response.requestHandle = entityHandlerRequest->requestHandle;
     response.resourceHandle = entityHandlerRequest->resource;
@@ -101,8 +111,8 @@ NSResult NSSendAccessPolicyResponse(OCEntityHandlerRequest *entityHandlerRequest
     // Send Response
     if (OCDoResponse(&response) != OC_STACK_OK)
     {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, PCF("Fail to send response"));
         NS_LOG(ERROR, "Fail to AccessPolicy send response");
+        OCRepPayloadDestroy(payload);
         return NS_ERROR;
     }
     OCRepPayloadDestroy(payload);
@@ -114,103 +124,161 @@ NSResult NSSendAccessPolicyResponse(OCEntityHandlerRequest *entityHandlerRequest
 
 void NSHandleSubscription(OCEntityHandlerRequest *entityHandlerRequest, NSResourceType resourceType)
 {
-
-    OIC_LOG(INFO, SUBSCRIPTION_TAG, "Start to subscription process");
     NS_LOG(DEBUG, "NSHandleSubscription - IN");
 
+    char * copyReq = OICStrdup(entityHandlerRequest->query);
+    char * id = NSGetValueFromQuery(copyReq, NS_QUERY_CONSUMER_ID);
+
+    if (!id)
+    {
+        NSOICFree(copyReq);
+        NSFreeOCEntityHandlerRequest(entityHandlerRequest);
+        NS_LOG(ERROR, "Invalid ConsumerID");
+        return;
+    }
+
+    NS_LOG_V(INFO_PRIVATE, "consumerId = %s", id);
     if (resourceType == NS_RESOURCE_MESSAGE)
     {
         NS_LOG(DEBUG, "resourceType == NS_RESOURCE_MESSAGE");
         NSCacheElement * element = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
+        NS_VERIFY_NOT_NULL_V(element);
         NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
+        if (!subData)
+        {
+            NS_LOG(ERROR, "Memory assignment to subData failed");
+            NSOICFree(element);
+            return;
+        }
+        OICStrcpy(subData->id, UUID_STRING_SIZE, id);
+        NS_LOG_V(INFO_PRIVATE, "SubList ID = [%s]", subData->id);
+
+        NS_LOG_V(INFO_PRIVATE, "Consumer Address: %s", entityHandlerRequest->devAddr.addr);
+
+        subData->remote_messageObId = subData->messageObId = 0;
+
+        bool iSRemoteServer = false;
+
+#if(defined WITH_CLOUD)
+        iSRemoteServer = NSIsRemoteServerAddress(entityHandlerRequest->devAddr.addr);
+        if (iSRemoteServer)
+        {
+            NS_LOG(DEBUG, "Requested by remote server");
+            subData->remote_messageObId = entityHandlerRequest->obsInfo.obsId;
+            NS_LOG_V(DEBUG, "SubList message observation ID = [%d]", subData->remote_messageObId);
+        }
+#endif
+
+        if (!iSRemoteServer)
+        {
+            NS_LOG(DEBUG, "Requested by local consumer");
+            subData->messageObId = entityHandlerRequest->obsInfo.obsId;
+            NS_LOG_V(DEBUG, "SubList message observation ID = [%d]", subData->messageObId);
+        }
 
-        subData->id = OICStrdup(entityHandlerRequest->devAddr.addr);
         subData->isWhite = false;
-        subData->messageObId = entityHandlerRequest->obsInfo.obsId;
+        subData->remote_syncObId = 0;
         subData->syncObId = 0;
 
         element->data = (void*) subData;
         element->next = NULL;
 
-        NS_LOG_V(DEBUG, "SubList IP[ID] = [%s]", subData->id);
-        NS_LOG_V(DEBUG, "SubList message observation ID = [%d]", subData->messageObId);
-
-        if (NSCacheWrite(consumerSubList, element) != NS_OK)
+        if (NSProviderStorageWrite(consumerSubList, element) != NS_OK)
         {
             NS_LOG(DEBUG, "fail to write cache");
         }
 
-        if (NSGetSubscriptionAccepter() == NS_ACCEPTER_PROVIDER)
+        bool currPolicy = NSGetPolicy();
+        NSAskAcceptanceToUser(NSCopyOCEntityHandlerRequest(entityHandlerRequest));
+
+        if (currPolicy == NS_POLICY_PROVIDER)
         {
             NS_LOG(DEBUG, "NSGetSubscriptionAccepter == NS_ACCEPTER_PROVIDER");
-            NSAskAcceptanceToUser(entityHandlerRequest);
         }
-        else if (NSGetSubscriptionAccepter() == NS_ACCEPTER_CONSUMER)
+        else if (currPolicy == NS_POLICY_CONSUMER)
         {
             NS_LOG(DEBUG, "NSGetSubscriptionAccepter == NS_ACCEPTER_CONSUMER");
-            NSSendSubscriptionResponse(entityHandlerRequest, true);
+            NSSendConsumerSubResponse(NSCopyOCEntityHandlerRequest(entityHandlerRequest));
         }
+
+        NSFreeOCEntityHandlerRequest(entityHandlerRequest);
     }
     else if (resourceType == NS_RESOURCE_SYNC)
     {
         NS_LOG(DEBUG, "resourceType == NS_RESOURCE_SYNC");
         NSCacheElement * element = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
-
+        NS_VERIFY_NOT_NULL_V(element);
         NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
+        if (!subData)
+        {
+            NS_LOG(ERROR, "Memory assignment to subData failed");
+            NSOICFree(element);
+            return;
+        }
+
+        OICStrcpy(subData->id, UUID_STRING_SIZE, id);
+        NS_LOG_V(INFO_PRIVATE, "SubList ID = [%s]", subData->id);
+
+        NS_LOG_V(INFO_PRIVATE, "Consumer Address: %s", entityHandlerRequest->devAddr.addr);
+
+        subData->remote_syncObId = subData->syncObId = 0;
+        bool isRemoteServer = false;
+
+#if (defined WITH_CLOUD)
+        isRemoteServer = NSIsRemoteServerAddress(entityHandlerRequest->devAddr.addr);
+        if (isRemoteServer)
+        {
+            NS_LOG(DEBUG, "Requested by remote server");
+            subData->remote_syncObId = entityHandlerRequest->obsInfo.obsId;
+            NS_LOG_V(DEBUG, "SubList sync observation ID = [%d]", subData->remote_syncObId);
+        }
+#endif
+
+        if (!isRemoteServer)
+        {
+            NS_LOG(DEBUG, "Requested by local consumer");
+            subData->syncObId = entityHandlerRequest->obsInfo.obsId;
+            NS_LOG_V(DEBUG, "SubList sync observation ID = [%d]", subData->syncObId);
+        }
 
-        subData->id = OICStrdup(entityHandlerRequest->devAddr.addr);
         subData->isWhite = false;
-        subData->syncObId = entityHandlerRequest->obsInfo.obsId;
         subData->messageObId = 0;
+        subData->remote_messageObId = 0;
 
         element->data = (void*) subData;
         element->next = NULL;
 
-        NS_LOG_V(DEBUG, "SubList IP[ID] = [%s]", subData->id);
-        NS_LOG_V(DEBUG, "SubList sync observation ID = [%d]", subData->syncObId);
-
-        if (NSCacheWrite(consumerSubList, element) != NS_OK)
+        if (NS_OK != NSProviderStorageWrite(consumerSubList, element))
         {
             NS_LOG(ERROR, "Fail to write cache");
         }
+
+        NSFreeOCEntityHandlerRequest(entityHandlerRequest);
     }
+    NSOICFree(copyReq);
 
     NS_LOG(DEBUG, "NSHandleSubscription - OUT");
 }
 
 void NSHandleUnsubscription(OCEntityHandlerRequest *entityHandlerRequest)
 {
-    OIC_LOG(INFO, SUBSCRIPTION_TAG, "Start to unsubscription process");
     NS_LOG(DEBUG, "NSHandleUnsubscription - IN");
 
-    NSCacheElement * element = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
-    NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
-    subData->id = OICStrdup(entityHandlerRequest->devAddr.addr);
-    subData->isWhite = false;
-    subData->messageObId = entityHandlerRequest->obsInfo.obsId;
-
-    element->data = (void*) subData;
-    element->next = NULL;
-
-    NS_LOG_V(DEBUG, "SubList IP[ID] = [%s]", subData->id);
-    NS_LOG_V(DEBUG, "SubList observation ID = [%d]", subData->syncObId);
+    consumerSubList->cacheType = NS_PROVIDER_CACHE_SUBSCRIBER_OBSERVE_ID;
 
-    if (NSCacheWrite(consumerSubList, element) != NS_OK)
-    {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, "fail to write consumer white list");
-        NS_LOG(ERROR, "fail to write consumer white list");
-    }
+    while (NSProviderStorageDelete(consumerSubList, (char *)
+            &(entityHandlerRequest->obsInfo.obsId)) != NS_FAIL);
 
-    NS_LOG(DEBUG, "NSHandleUnsubscription - IN");
+    consumerSubList->cacheType = NS_PROVIDER_CACHE_SUBSCRIBER;
     NSFreeOCEntityHandlerRequest(entityHandlerRequest);
+    NS_LOG(DEBUG, "NSHandleUnsubscription - OUT");
 }
 
 void NSAskAcceptanceToUser(OCEntityHandlerRequest *entityHandlerRequest)
 {
-    OIC_LOG(DEBUG, SUBSCRIPTION_TAG, "Ask for user to Allow or Deny");
     NS_LOG(DEBUG, "NSAskAcceptanceToUser - IN");
 
-    NSPushQueue(RESPONSE_SCHEDULER, TASK_CB_SUBSCRIPTION, entityHandlerRequest);
+    NSPushQueue(CALLBACK_RESPONSE_SCHEDULER, TASK_CB_SUBSCRIPTION, entityHandlerRequest);
 
     NS_LOG(DEBUG, "NSAskAcceptanceToUser - OUT");
 }
@@ -222,91 +290,111 @@ NSResult NSSendResponse(const char * id, bool accepted)
     OCRepPayload* payload = OCRepPayloadCreate();
     if (!payload)
     {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, PCF("Failed to allocate Payload"));
         NS_LOG(ERROR, "fail to create playload");
         return NS_ERROR;
     }
 
-    OCResourceHandle rHandle;
+    OCResourceHandle rHandle = NULL;
     if (NSPutMessageResource(NULL, &rHandle) != NS_OK)
     {
-        OIC_LOG(ERROR, NOTIFICATION_TAG, PCF("Fail to put notification resource"));
         NS_LOG(ERROR, "Fail to put notification resource");
         return NS_ERROR;
     }
 
-    OCRepPayloadSetUri(payload, NSGetNotificationMessageUri());
-    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_ID, "0000-0000-0000-0000");
-    OCRepPayloadSetPropBool(payload, NS_ATTRIBUTE_ACCPETANCE, accepted);
+    OCRepPayloadSetUri(payload, NS_COLLECTION_MESSAGE_URI);
+    (accepted) ? OCRepPayloadSetPropInt(payload, NS_ATTRIBUTE_MESSAGE_ID, NS_ALLOW)
+        : OCRepPayloadSetPropInt(payload, NS_ATTRIBUTE_MESSAGE_ID, NS_DENY);
+    OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_PROVIDER_ID, NSGetProviderInfo()->providerId);
 
-    NSCacheElement * element = NSCacheRead(consumerSubList, id);
+    NSCacheElement * element = NSProviderStorageRead(consumerSubList, id);
 
-    if(element == NULL)
+    if (element == NULL)
     {
         NS_LOG(ERROR, "element is NULL");
         return NS_ERROR;
     }
+
     NSCacheSubData * subData = (NSCacheSubData*) element->data;
 
-    if (OCNotifyListOfObservers(rHandle, (OCObservationId*)&subData->messageObId, 1, payload, OC_HIGH_QOS)
-            != OC_STACK_OK)
+    if (OCNotifyListOfObservers(rHandle, (OCObservationId*)&subData->messageObId, 1,
+            payload, OC_LOW_QOS) != OC_STACK_OK)
     {
-        OIC_LOG(ERROR, SUBSCRIPTION_TAG, "fail to send Acceptance");
         NS_LOG(ERROR, "fail to send Acceptance");
         OCRepPayloadDestroy(payload);
         return NS_ERROR;
 
     }
-    OCRepPayloadDestroy(payload);
 
+    OCRepPayloadDestroy(payload);
     NS_LOG(DEBUG, "NSSendResponse - OUT");
     return NS_OK;
 }
 
-NSResult NSSendSubscriptionResponse(OCEntityHandlerRequest *entityHandlerRequest, bool accepted)
+NSResult NSSendConsumerSubResponse(OCEntityHandlerRequest * entityHandlerRequest)
 {
-    OIC_LOG(INFO, SUBSCRIPTION_TAG, "Send ACCEPT message to consumer");
     NS_LOG(DEBUG, "NSSendSubscriptionResponse - IN");
 
     if (!entityHandlerRequest)
     {
-        OIC_LOG (ERROR, LISTENER_TAG, "Invalid request pointer");
-        return OC_EH_ERROR;
+        NS_LOG(ERROR, "Invalid request pointer");
+        return NS_ERROR;
     }
 
-    if (accepted)
-    {
-        NS_LOG(DEBUG, "accepted is true");
-        NSCacheElement * element = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
-
-        NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
-        subData->id = OICStrdup(entityHandlerRequest->devAddr.addr);
-        subData->isWhite = true;
-        subData->messageObId = entityHandlerRequest->obsInfo.obsId;
-
-        element->data = (void*) subData;
-        element->next = NULL;
+    char * copyReq = OICStrdup(entityHandlerRequest->query);
+    char * id = NSGetValueFromQuery(copyReq, NS_QUERY_CONSUMER_ID);
 
-        if (NSCacheWrite(consumerSubList, element) != NS_OK)
-        {
-            OIC_LOG(ERROR, SUBSCRIPTION_TAG, "fail to write consumer white list");
-            NS_LOG(ERROR, "fail to write consumer white list");
-        }
+    if (!id)
+    {
+        NSOICFree(copyReq);
+        NSFreeOCEntityHandlerRequest(entityHandlerRequest);
+        NS_LOG(ERROR, "Invalid ConsumerID");
+        return NS_ERROR;
     }
 
-    NSSendResponse(entityHandlerRequest->devAddr.addr, accepted);
+    NSCacheUpdateSubScriptionState(consumerSubList, id, true);
+    NSSendResponse(id, true);
+    NSOICFree(copyReq);
     NSFreeOCEntityHandlerRequest(entityHandlerRequest);
-
     NS_LOG(DEBUG, "NSSendSubscriptionResponse - OUT");
     return NS_OK;
 }
 
-void * NSSubScriptionSchedule(void *ptr)
+#ifdef WITH_MQ
+void NSProviderMQSubscription(NSMQTopicAddress * topicAddr)
 {
+    char * serverUri = topicAddr->serverAddr;
+    char * topicName = topicAddr->topicName;
+
+    NS_LOG_V(DEBUG, "input Topic Name2 : %s", topicAddr->topicName);
+
+    OCDevAddr * addr = NSChangeAddress(serverUri);
+    OCCallbackData cbdata = { NULL, NULL, NULL };
+    cbdata.cb = NSProviderGetMQResponseCB;
+    cbdata.context = OICStrdup(topicName);
+    cbdata.cd = NSOICFree;
+
+    char requestUri[100] = "coap+tcp://";
+
+    NS_LOG_V(DEBUG, "requestUri1 = %s", requestUri);
+    OICStrcat(requestUri, strlen(requestUri)+strlen(serverUri)+1, serverUri);
+    NS_LOG_V(DEBUG, "requestUri2 = %s", requestUri);
+    OICStrcat(requestUri, strlen(requestUri)+ strlen("/oic/ps") + 1, "/oic/ps");
+    NS_LOG_V(DEBUG, "requestUri3 = %s", requestUri);
+    OCStackResult ret = OCDoResource(NULL, OC_REST_GET, requestUri, addr,
+                                     NULL, CT_DEFAULT, OC_HIGH_QOS, &cbdata, NULL, 0);
 
+    NSOCResultToSuccess(ret);
+
+    NSOICFree(topicAddr->serverAddr);
+    NSOICFree(topicAddr->topicName);
+    NSOICFree(topicAddr);
+}
+#endif
+
+void * NSSubScriptionSchedule(void *ptr)
+{
     if (ptr == NULL)
     {
-        OIC_LOG(INFO, SUBSCRIPTION_TAG, "Create NSSubscriptionSchedule");
         NS_LOG(DEBUG, "Create NSSubScriptionSchedule");
     }
 
@@ -341,34 +429,22 @@ void * NSSubScriptionSchedule(void *ptr)
                 case TASK_SEND_ALLOW:
                 {
                     NS_LOG(DEBUG, "CASE TASK_SEND_ALLOW : ");
-                    NSConsumer * consumer = (NSConsumer *) node->taskData;
-
-                    NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
-                    subData->id = OICStrdup(consumer->mId);
-                    subData->isWhite = true;
-                    subData->messageObId = 0;
-                    subData->syncObId = 0;
-
-                    NSCacheUpdateSubScriptionState(consumerSubList, subData);
-                    NSSendResponse(consumer->mId, true);
+                    char * consumerId = (char *) node->taskData;
 
+                    NSCacheUpdateSubScriptionState(consumerSubList, consumerId, true);
+                    NSSendResponse(consumerId, true);
+                    NSOICFree(consumerId);
                     break;
                 }
                 case TASK_SEND_DENY:
                 {
                     NS_LOG(DEBUG, "CASE TASK_SEND_DENY : ");
-                    NSConsumer * consumer = (NSConsumer *) node->taskData;
-                    int * pObId = (int *) consumer->mUserData;
-
-                    NSCacheSubData * subData = (NSCacheSubData *) OICMalloc(sizeof(NSCacheSubData));
-                    subData->id = OICStrdup(consumer->mId);
-                    subData->isWhite = false;
-                    subData->messageObId = 0;
-                    subData->syncObId = 0;
-
-                    NSCacheUpdateSubScriptionState(consumerSubList, subData);
-                    printf("observer ID = %d\n", *pObId);
-                    NSSendResponse(consumer->mId, false);
+                    char * consumerId = (char *) node->taskData;
+
+                    NSCacheUpdateSubScriptionState(consumerSubList, consumerId, false);
+                    NSSendResponse(consumerId, false);
+                    NSOICFree(consumerId);
+
                     break;
                 }
                 case TASK_SYNC_SUBSCRIPTION:
@@ -376,15 +452,22 @@ void * NSSubScriptionSchedule(void *ptr)
                     NSHandleSubscription((OCEntityHandlerRequest*) node->taskData,
                             NS_RESOURCE_SYNC);
                     break;
+#ifdef WITH_MQ
+                case TASK_MQ_REQ_SUBSCRIBE:
+                    NS_LOG(DEBUG, "CASE TASK_MQ_REQ_SUBSCRIBE : ");
+                    NSProviderMQSubscription((NSMQTopicAddress*) node->taskData);
+                    break;
+#endif
                 default:
                     break;
 
             }
-            OICFree(node);
+            NSOICFree(node);
         }
 
         pthread_mutex_unlock(&NSMutex[SUBSCRIPTION_SCHEDULER]);
 
     }
+    NS_LOG(INFO, "Destroy NSSubScriptionSchedule");
     return NULL;
 }