service/notification/src/provider/NSProviderTopic.c

   1 //******************************************************************
   2 //
   3 // Copyright 2016 Samsung Electronics 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 "NSProviderTopic.h"
  22 #include "oic_string.h"
  23 #include "oic_malloc.h"
  24 #include <pthread.h>
  25
  26 NSCacheList * consumerTopicList;
  27 NSCacheList * registeredTopicList;
  28
  29 NSResult NSSendTopicUpdation();
  30
  31 NSResult NSInitTopicList()
  32 {
  33     NS_LOG(DEBUG, "NSInitTopicList - IN");
  34
  35     consumerTopicList = NSProviderStorageCreate();
  36     NS_VERIFY_NOT_NULL(consumerTopicList, NS_FAIL);
  37     consumerTopicList->cacheType = NS_PROVIDER_CACHE_CONSUMER_TOPIC_NAME;
  38
  39     registeredTopicList = NSProviderStorageCreate();
  40     NS_VERIFY_NOT_NULL(registeredTopicList, NS_FAIL);
  41     registeredTopicList->cacheType = NS_PROVIDER_CACHE_REGISTER_TOPIC;
  42
  43     NS_LOG(DEBUG, "NSInitTopicList - OUT");
  44     return NS_OK;
  45 }
  46
  47 size_t NSProviderGetTopicListSize(NSTopicLL * firstElement)
  48 {
  49     if (!firstElement)
  50     {
  51         return 0;
  52     }
  53
  54     int cnt = 0;
  55
  56     NSTopicLL * iter = firstElement;
  57
  58     while (iter)
  59     {
  60         cnt++;
  61         iter = iter->next;
  62     }
  63
  64     return cnt;
  65 }
  66
  67 NSResult NSRegisterTopic(const char * topicName)
  68 {
  69     NS_LOG(DEBUG, "NSWriteTopicsToStorage()");
  70
  71     NSCacheTopicData * data = (NSCacheTopicData *) OICMalloc(sizeof(NSCacheTopicData));
  72     NS_VERIFY_NOT_NULL(data, NS_FAIL);
  73     data->topicName = (char *) topicName;
  74     data->state = NS_TOPIC_UNSUBSCRIBED;
  75
  76     NSCacheElement * element = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
  77     if (!element)
  78     {
  79         NSOICFree(data->topicName);
  80         NSOICFree(data);
  81         return NS_FAIL;
  82     }
  83
  84     element->data = (void *) data;
  85     element->next = NULL;
  86
  87     if (NSProviderStorageWrite(registeredTopicList, element) != NS_OK)
  88     {
  89         NS_LOG(DEBUG, "fail to write cache");
  90         return NS_FAIL;
  91     }
  92
  93     NSSendTopicUpdation();
  94     NS_LOG(DEBUG, "NSWriteTopicsToStorage() NS_OK");
  95     return NS_OK;
  96 }
  97
  98 NSResult NSUnregisterTopic(const char * topicName)
  99 {
 100     NS_LOG(DEBUG, "NSDeleteTopics()");
 101     NSResult result = NS_OK;
 102
 103     if (!topicName)
 104     {
 105         NS_LOG(ERROR, "topicName is NULL");
 106         return NS_ERROR;
 107     }
 108
 109     result = NSProviderStorageDelete(registeredTopicList, topicName);
 110
 111     while (NSProviderStorageDelete(consumerTopicList, topicName) != NS_FAIL)
 112     {
 113     }
 114
 115     if (result == NS_OK)
 116     {
 117         NSSendTopicUpdation();
 118     }
 119
 120     return result;
 121 }
 122
 123 NSResult NSSendTopicUpdation()
 124 {
 125     NS_LOG(DEBUG, "NSSendTopicUpdation - IN");
 126
 127     OCRepPayload* payload = OCRepPayloadCreate();
 128
 129     if (!payload)
 130     {
 131         NS_LOG(ERROR, "fail to create playload");
 132         return NS_ERROR;
 133     }
 134
 135     OCResourceHandle rHandle = NULL;
 136     if (NSPutMessageResource(NULL, &rHandle) != NS_OK)
 137     {
 138         NS_LOG(ERROR, "Fail to put message resource");
 139         return NS_ERROR;
 140     }
 141
 142     OCRepPayloadSetUri(payload, NS_COLLECTION_MESSAGE_URI);
 143     OCRepPayloadSetPropInt(payload, NS_ATTRIBUTE_MESSAGE_ID, NS_TOPIC);
 144     OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_PROVIDER_ID, NSGetProviderInfo()->providerId);
 145
 146     OCObservationId obArray[3839] =
 147     { 0, };
 148     int obCount = 0;
 149
 150     NSCacheElement * it = consumerSubList->head;
 151
 152     while (it)
 153     {
 154         NSCacheSubData * subData = (NSCacheSubData *) it->data;
 155
 156         if (subData->isWhite)
 157         {
 158             if (subData->messageObId != 0)
 159             {
 160                 obArray[obCount++] = subData->messageObId;
 161             }
 162
 163 #if (defined WITH_CLOUD)
 164             if (subData->remote_messageObId != 0)
 165             {
 166                 obArray[obCount++] = subData->remote_messageObId;
 167             }
 168 #endif
 169         }
 170
 171         it = it->next;
 172     }
 173
 174     if (!obCount)
 175     {
 176         NS_LOG(ERROR, "observer count is zero");
 177         return NS_ERROR;
 178     }
 179
 180     if (OCNotifyListOfObservers(rHandle, obArray, obCount, payload, OC_HIGH_QOS) != OC_STACK_OK)
 181     {
 182         NS_LOG(ERROR, "fail to send topic updation");
 183         OCRepPayloadDestroy(payload);
 184         return NS_ERROR;
 185
 186     }
 187     OCRepPayloadDestroy(payload);
 188
 189     NS_LOG(DEBUG, "NSSendTopicUpdation - OUT");
 190     return NS_OK;
 191 }
 192
 193 NSResult NSSendTopicUpdationToConsumer(char *consumerId)
 194 {
 195     NS_LOG(DEBUG, "NSSendTopicUpdationToConsumer - IN");
 196
 197     OCRepPayload* payload = OCRepPayloadCreate();
 198
 199     if (!payload)
 200     {
 201         NS_LOG(ERROR, "fail to create playload");
 202         return NS_ERROR;
 203     }
 204
 205     OCResourceHandle rHandle = NULL;
 206     if (NSPutMessageResource(NULL, &rHandle) != NS_OK)
 207     {
 208         NS_LOG(ERROR, "Fail to put message resource");
 209         return NS_ERROR;
 210     }
 211
 212     OCRepPayloadSetUri(payload, NS_COLLECTION_MESSAGE_URI);
 213     OCRepPayloadSetPropInt(payload, NS_ATTRIBUTE_MESSAGE_ID, NS_TOPIC);
 214     OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_PROVIDER_ID, NSGetProviderInfo()->providerId);
 215
 216     NSCacheElement * element = NSProviderStorageRead(consumerSubList, consumerId);
 217
 218     if (element == NULL)
 219     {
 220         NS_LOG(ERROR, "element is NULL");
 221         return NS_ERROR;
 222     }
 223
 224     NSCacheSubData * subData = (NSCacheSubData*) element->data;
 225
 226     if (OCNotifyListOfObservers(rHandle, (OCObservationId*) &subData->messageObId, 1, payload,
 227             OC_HIGH_QOS) != OC_STACK_OK)
 228     {
 229         NS_LOG(ERROR, "fail to send topic updation");
 230         OCRepPayloadDestroy(payload);
 231         return NS_ERROR;
 232     }
 233
 234     OCRepPayloadDestroy(payload);
 235
 236     NS_LOG(DEBUG, "NSSendTopicUpdationToConsumer - OUT");
 237     return NS_OK;
 238 }
 239
 240 NSResult NSSendTopicList(OCEntityHandlerRequest * entityHandlerRequest)
 241 {
 242     NS_LOG(DEBUG, "NSSendTopicList - IN");
 243
 244     char * copyReq = OICStrdup(entityHandlerRequest->query);
 245     char * id = NSGetValueFromQuery(copyReq, NS_QUERY_CONSUMER_ID);
 246     NSTopicLL * topics = NULL;
 247
 248     if (!id)
 249     {
 250         NS_LOG(DEBUG, "Send registered topic list");
 251         topics = NSProviderGetTopicsCacheData(registeredTopicList);
 252     }
 253     else
 254     {
 255         NS_LOG(DEBUG, "Send subscribed topic list to consumer");
 256         topics = NSProviderGetConsumerTopicsCacheData(registeredTopicList, consumerTopicList, id);
 257         if (!topics)
 258         {
 259             topics = NSProviderGetTopicsCacheData(registeredTopicList);
 260         }
 261     }
 262
 263     // make response for the Get Request
 264     OCEntityHandlerResponse response;
 265     response.numSendVendorSpecificHeaderOptions = 0;
 266     memset(response.sendVendorSpecificHeaderOptions, 0,
 267             sizeof response.sendVendorSpecificHeaderOptions);
 268     memset(response.resourceUri, 0, sizeof response.resourceUri);
 269
 270     OCRepPayload* payload = OCRepPayloadCreate();
 271     if (!payload)
 272     {
 273         NS_LOG(ERROR, "payload is NULL");
 274         NSOICFree(copyReq);
 275         return NS_ERROR;
 276     }
 277
 278     OCRepPayloadSetUri(payload, NS_COLLECTION_TOPIC_URI);
 279     if (id)
 280     {
 281         OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_CONSUMER_ID, id);
 282     }
 283     OCRepPayloadSetPropString(payload, NS_ATTRIBUTE_PROVIDER_ID, NSGetProviderInfo()->providerId);
 284     NSOICFree(copyReq);
 285
 286     if (topics)
 287     {
 288         NS_LOG(DEBUG, "topicList is NULL");
 289         size_t dimensionSize = (size_t) NSProviderGetTopicListSize(topics);
 290         NS_LOG_V(DEBUG, "dimensionSize = %d", (int)dimensionSize);
 291
 292         if (!dimensionSize)
 293         {
 294             return NS_ERROR;
 295         }
 296
 297         OCRepPayload** payloadTopicArray = (OCRepPayload **) OICMalloc(
 298                 sizeof(OCRepPayload *) * dimensionSize);
 299         NS_VERIFY_NOT_NULL(payloadTopicArray, NS_ERROR);
 300
 301         size_t dimensions[3] = { dimensionSize, 0, 0 };
 302
 303         for (int i = 0; i < (int) dimensionSize; i++)
 304         {
 305             NS_LOG_V(DEBUG, "topicName = %s", topics->topicName);
 306             NS_LOG_V(DEBUG, "topicState = %d",(int) topics->state);
 307
 308             payloadTopicArray[i] = OCRepPayloadCreate();
 309             if (!payloadTopicArray[i])
 310             {
 311                 NS_LOG_V(ERROR, "payloadTopicArray[%d] is NULL", i);
 312                 NSOICFree(payloadTopicArray);
 313                 return NS_ERROR;
 314             }
 315
 316             OCRepPayloadSetPropString(payloadTopicArray[i], NS_ATTRIBUTE_TOPIC_NAME,
 317                     topics->topicName);
 318             OCRepPayloadSetPropInt(payloadTopicArray[i], NS_ATTRIBUTE_TOPIC_SELECTION,
 319                     (int) topics->state);
 320
 321             NSTopicLL * next = topics->next;
 322             NSOICFree(topics->topicName);
 323             NSOICFree(topics);
 324             topics = next;
 325         }
 326
 327         OCRepPayloadSetPropObjectArrayAsOwner(payload, NS_ATTRIBUTE_TOPIC_LIST,
 328                     payloadTopicArray, dimensions);
 329     }
 330     else
 331     {
 332         size_t dimensions[3] = { 0, 0, 0 };
 333
 334         OCRepPayloadSetPropObjectArrayAsOwner(payload, NS_ATTRIBUTE_TOPIC_LIST,
 335                 (OCRepPayload **) NULL, dimensions);
 336     }
 337
 338     copyReq = OICStrdup(entityHandlerRequest->query);
 339     char * reqInterface = NSGetValueFromQuery(copyReq, NS_QUERY_INTERFACE);
 340
 341     if (reqInterface && strcmp(reqInterface, NS_INTERFACE_BASELINE) == 0)
 342     {
 343         OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_BASELINE);
 344         OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_READ);
 345         OCResourcePayloadAddStringLL(&payload->types, NS_ROOT_TYPE);
 346     }
 347
 348     NSOICFree(copyReq);
 349     response.requestHandle = entityHandlerRequest->requestHandle;
 350     response.resourceHandle = entityHandlerRequest->resource;
 351     response.persistentBufferFlag = 0;
 352     response.ehResult = OC_EH_OK;
 353     response.payload = (OCPayload *) payload;
 354
 355     if (OCDoResponse(&response) != OC_STACK_OK)
 356     {
 357         NS_LOG(ERROR, "Fail to response topic list");
 358         OCRepPayloadDestroy(payload);
 359         return NS_ERROR;
 360     }
 361
 362     OCRepPayloadDestroy(payload);
 363     NS_LOG(DEBUG, "NSSendTopicList - OUT");
 364     return NS_OK;
 365 }
 366
 367 NSResult NSPostConsumerTopics(OCEntityHandlerRequest * entityHandlerRequest)
 368 {
 369     NS_LOG(DEBUG, "NSPostConsumerTopics() - IN");
 370
 371     char * consumerId = NULL;
 372     OCRepPayload * payload = (OCRepPayload *) entityHandlerRequest->payload;
 373     OCRepPayloadGetPropString(payload, NS_ATTRIBUTE_CONSUMER_ID, &consumerId);
 374
 375     if (!consumerId)
 376     {
 377         NS_LOG(DEBUG, "Invalid consumer ID");
 378         return NS_FAIL;
 379     }
 380
 381     NS_LOG_V(INFO_PRIVATE, "TOPIC consumer ID = %s", consumerId);
 382
 383     consumerTopicList->cacheType = NS_PROVIDER_CACHE_CONSUMER_TOPIC_CID;
 384
 385     while (NSProviderStorageDelete(consumerTopicList, consumerId) != NS_FAIL)
 386     {
 387     }
 388
 389     consumerTopicList->cacheType = NS_PROVIDER_CACHE_CONSUMER_TOPIC_NAME;
 390     OCRepPayload ** topicListPayload = NULL;
 391     OCRepPayloadValue * payloadValue = NULL;
 392     payloadValue = NSPayloadFindValue(payload, NS_ATTRIBUTE_TOPIC_LIST);
 393     size_t dimensionSize = calcDimTotal(payloadValue->arr.dimensions);
 394     size_t dimensions[3] = { dimensionSize, 0, 0 };
 395     OCRepPayloadGetPropObjectArray(payload, NS_ATTRIBUTE_TOPIC_LIST, &topicListPayload, dimensions);
 396
 397     for (int i = 0; i < (int) dimensionSize; i++)
 398     {
 399         char * topicName = NULL;
 400         int64_t topicState = 0;
 401
 402         OCRepPayloadGetPropString(topicListPayload[i], NS_ATTRIBUTE_TOPIC_NAME, &topicName);
 403         if (OCRepPayloadGetPropInt(topicListPayload[i], NS_ATTRIBUTE_TOPIC_SELECTION, &topicState))
 404         {
 405             NS_LOG_V(DEBUG, "Topic Name(state):  %s(%d)", topicName, (int)topicState);
 406         }
 407
 408         if (NS_TOPIC_SUBSCRIBED == (NSTopicState) topicState)
 409         {
 410             NSCacheTopicSubData * topicSubData = (NSCacheTopicSubData *) OICMalloc(
 411                     sizeof(NSCacheTopicSubData));
 412             NS_VERIFY_NOT_NULL_EXIT(topicSubData);
 413
 414             OICStrcpy(topicSubData->id, NS_UUID_STRING_SIZE, consumerId);
 415             topicSubData->topicName = topicName;
 416
 417             NSCacheElement * newObj = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
 418
 419             if (!newObj)
 420             {
 421                 NSOICFree(topicSubData->topicName);
 422                 NSOICFree(topicSubData);
 423                 NSOICFree(consumerId);
 424
 425                 // Free topic list
 426                 for (size_t k = 0; k < dimensionSize; k++)
 427                 {
 428                     OCRepPayloadDestroy(topicListPayload[k]);
 429                 }
 430                 OICFree(topicListPayload);
 431
 432                 return NS_FAIL;
 433             }
 434
 435             newObj->data = (NSCacheData *) topicSubData;
 436             newObj->next = NULL;
 437
 438             if (NS_OK != NSProviderStorageWrite(consumerTopicList, newObj))
 439             {
 440                 NS_LOG(ERROR, "Fail to write cache");
 441             }
 442         }
 443     }
 444     NSSendTopicUpdationToConsumer(consumerId);
 445
 446     // Free topic list
 447     for (size_t k = 0; k < dimensionSize; k++)
 448     {
 449         OCRepPayloadDestroy(topicListPayload[k]);
 450     }
 451     OICFree(topicListPayload);
 452
 453     NSOICFree(consumerId);
 454     NS_LOG(DEBUG, "NSPostConsumerTopics() - OUT");
 455     return NS_OK;
 456
 457 exit:
 458     OICFree(topicListPayload);
 459     return NS_FAIL;
 460 }
 461
 462 void * NSTopicSchedule(void * ptr)
 463 {
 464     if (ptr == NULL)
 465     {
 466         NS_LOG(DEBUG, "Create NSTopicSchedule");
 467     }
 468
 469     while (NSIsRunning[TOPIC_SCHEDULER])
 470     {
 471         sem_wait(&NSSemaphore[TOPIC_SCHEDULER]);
 472         pthread_mutex_lock(&NSMutex[TOPIC_SCHEDULER]);
 473
 474         if (NSHeadMsg[TOPIC_SCHEDULER] != NULL)
 475         {
 476             NSTask *node = NSHeadMsg[TOPIC_SCHEDULER];
 477             NSHeadMsg[TOPIC_SCHEDULER] = node->nextTask;
 478
 479             switch (node->taskType)
 480             {
 481                 case TASK_SEND_TOPICS:
 482                     NS_LOG(DEBUG, "CASE TASK_SEND_TOPICS : ");
 483                     NSSendTopicList((OCEntityHandlerRequest*) node->taskData);
 484                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 485                     break;
 486                 case TASK_SUBSCRIBE_TOPIC:
 487                 {
 488                     NS_LOG(DEBUG, "CASE TASK_SUBSCRIBE_TOPIC : ");
 489                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 490                     pthread_mutex_lock(topicSyncResult->mutex);
 491                     NSCacheElement * newObj = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
 492                     NSCacheTopicSubData * subData =
 493                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 494                     if (!newObj)
 495                     {
 496                         NSOICFree(subData->topicName);
 497                         NSOICFree(subData);
 498                         pthread_cond_signal(topicSyncResult->condition);
 499                         pthread_mutex_unlock(topicSyncResult->mutex);
 500                     }
 501                     else
 502                     {
 503                         if (NSProviderStorageRead(registeredTopicList, subData->topicName))
 504                         {
 505                             newObj->data = topicSyncResult->topicData;
 506                             newObj->next = NULL;
 507
 508                             if (NSProviderStorageWrite(consumerTopicList, newObj) == NS_OK)
 509                             {
 510                                 if(subData)
 511                                 {
 512                                 NSSendTopicUpdationToConsumer(subData->id);
 513                                 }
 514                                 topicSyncResult->result = NS_OK;
 515                             }
 516                         }
 517                         else
 518                         {
 519                             NSOICFree(subData->topicName);
 520                             NSOICFree(subData);
 521                             NSOICFree(newObj);
 522                         }
 523                     }
 524                     pthread_cond_signal(topicSyncResult->condition);
 525                     pthread_mutex_unlock(topicSyncResult->mutex);
 526                 }
 527                     break;
 528                 case TASK_UNSUBSCRIBE_TOPIC:
 529                 {
 530                     NS_LOG(DEBUG, "CASE TASK_UNSUBSCRIBE_TOPIC : ");
 531                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 532                     pthread_mutex_lock(topicSyncResult->mutex);
 533                     NSCacheTopicSubData * topicSubData =
 534                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 535
 536                     if (NSProviderDeleteConsumerTopic(consumerTopicList, topicSubData) == NS_OK)
 537                     {
 538                         NSSendTopicUpdationToConsumer(topicSubData->id);
 539                         topicSyncResult->result = NS_OK;
 540                     }
 541
 542                     NSOICFree(topicSubData->topicName);
 543                     NSOICFree(topicSubData);
 544                     pthread_cond_signal(topicSyncResult->condition);
 545                     pthread_mutex_unlock(topicSyncResult->mutex);
 546
 547                 }
 548                     break;
 549                 case TASK_REGISTER_TOPIC:
 550                 {
 551                     NS_LOG(DEBUG, "CASE TASK_ADD_TOPIC : ");
 552                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 553
 554                     pthread_mutex_lock(topicSyncResult->mutex);
 555                     topicSyncResult->result = NSRegisterTopic(
 556                             (const char *) topicSyncResult->topicData);
 557                     pthread_cond_signal(topicSyncResult->condition);
 558                     pthread_mutex_unlock(topicSyncResult->mutex);
 559                 }
 560                     break;
 561                 case TASK_UNREGISTER_TOPIC:
 562                 {
 563                     NS_LOG(DEBUG, "CASE_TASK_DELETE_TOPIC : ");
 564                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 565                     pthread_mutex_lock(topicSyncResult->mutex);
 566                     topicSyncResult->result = NSUnregisterTopic(
 567                             (const char *) topicSyncResult->topicData);
 568                     pthread_cond_signal(topicSyncResult->condition);
 569                     pthread_mutex_unlock(topicSyncResult->mutex);
 570                 }
 571                     break;
 572                 case TASK_POST_TOPIC:
 573                 {
 574                     NS_LOG(DEBUG, "TASK_POST_TOPIC : ");
 575                     NSPostConsumerTopics((OCEntityHandlerRequest*) node->taskData);
 576                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 577                 }
 578                     break;
 579                 case TASK_GET_TOPICS:
 580                 {
 581                     NS_LOG(DEBUG, "TASK_GET_TOPICS : ");
 582                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 583                     pthread_mutex_lock(topicSync->mutex);
 584                     NSTopicLL * topics = NSProviderGetTopicsCacheData(registeredTopicList);
 585                     topicSync->topics = topics;
 586                     pthread_cond_signal(topicSync->condition);
 587                     pthread_mutex_unlock(topicSync->mutex);
 588                 }
 589                     break;
 590                 case TAST_GET_CONSUMER_TOPICS:
 591                 {
 592                     NS_LOG(DEBUG, "TASK_GET_CONSUMER_TOPICS : ");
 593                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 594                     pthread_mutex_lock(topicSync->mutex);
 595                     NSTopicLL * topics = NSProviderGetConsumerTopicsCacheData(registeredTopicList,
 596                             consumerTopicList, topicSync->consumerId);
 597                     topicSync->topics = topics;
 598                     pthread_cond_signal(topicSync->condition);
 599                     pthread_mutex_unlock(topicSync->mutex);
 600                 }
 601                     break;
 602                 default:
 603                     break;
 604             }
 605
 606             NSOICFree(node);
 607         }
 608
 609         pthread_mutex_unlock(&NSMutex[TOPIC_SCHEDULER]);
 610     }
 611
 612     NS_LOG(DEBUG, "Destroy NSTopicSchedule");
 613     return NULL;
 614 }