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             NS_VERIFY_NOT_NULL(payloadTopicArray[i], NS_ERROR);
 310             OCRepPayloadSetPropString(payloadTopicArray[i], NS_ATTRIBUTE_TOPIC_NAME,
 311                     topics->topicName);
 312             OCRepPayloadSetPropInt(payloadTopicArray[i], NS_ATTRIBUTE_TOPIC_SELECTION,
 313                     (int) topics->state);
 314
 315             NSTopicLL * next = topics->next;
 316             NSOICFree(topics->topicName);
 317             NSOICFree(topics);
 318             topics = next;
 319         }
 320
 321         OCRepPayloadSetPropObjectArrayAsOwner(payload, NS_ATTRIBUTE_TOPIC_LIST,
 322                     payloadTopicArray, dimensions);
 323     }
 324     else
 325     {
 326         size_t dimensions[3] = { 0, 0, 0 };
 327
 328         OCRepPayloadSetPropObjectArrayAsOwner(payload, NS_ATTRIBUTE_TOPIC_LIST,
 329                 (OCRepPayload **) NULL, dimensions);
 330     }
 331
 332     copyReq = OICStrdup(entityHandlerRequest->query);
 333     char * reqInterface = NSGetValueFromQuery(copyReq, NS_QUERY_INTERFACE);
 334
 335     if (reqInterface && strcmp(reqInterface, NS_INTERFACE_BASELINE) == 0)
 336     {
 337         OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_BASELINE);
 338         OCResourcePayloadAddStringLL(&payload->interfaces, NS_INTERFACE_READ);
 339         OCResourcePayloadAddStringLL(&payload->types, NS_ROOT_TYPE);
 340     }
 341
 342     NSOICFree(copyReq);
 343     response.requestHandle = entityHandlerRequest->requestHandle;
 344     response.resourceHandle = entityHandlerRequest->resource;
 345     response.persistentBufferFlag = 0;
 346     response.ehResult = OC_EH_OK;
 347     response.payload = (OCPayload *) payload;
 348
 349     if (OCDoResponse(&response) != OC_STACK_OK)
 350     {
 351         NS_LOG(ERROR, "Fail to response topic list");
 352         OCRepPayloadDestroy(payload);
 353         return NS_ERROR;
 354     }
 355
 356     OCRepPayloadDestroy(payload);
 357     NS_LOG(DEBUG, "NSSendTopicList - OUT");
 358     return NS_OK;
 359 }
 360
 361 NSResult NSPostConsumerTopics(OCEntityHandlerRequest * entityHandlerRequest)
 362 {
 363     NS_LOG(DEBUG, "NSPostConsumerTopics() - IN");
 364
 365     char * consumerId = NULL;
 366     OCRepPayload * payload = (OCRepPayload *) entityHandlerRequest->payload;
 367     OCRepPayloadGetPropString(payload, NS_ATTRIBUTE_CONSUMER_ID, &consumerId);
 368
 369     if (!consumerId)
 370     {
 371         NS_LOG(DEBUG, "Invalid consumer ID");
 372         return NS_FAIL;
 373     }
 374
 375     NS_LOG_V(INFO_PRIVATE, "TOPIC consumer ID = %s", consumerId);
 376
 377     consumerTopicList->cacheType = NS_PROVIDER_CACHE_CONSUMER_TOPIC_CID;
 378
 379     while (NSProviderStorageDelete(consumerTopicList, consumerId) != NS_FAIL)
 380     {
 381     }
 382
 383     consumerTopicList->cacheType = NS_PROVIDER_CACHE_CONSUMER_TOPIC_NAME;
 384     OCRepPayload ** topicListPayload = NULL;
 385     OCRepPayloadValue * payloadValue = NULL;
 386     payloadValue = NSPayloadFindValue(payload, NS_ATTRIBUTE_TOPIC_LIST);
 387     size_t dimensionSize = calcDimTotal(payloadValue->arr.dimensions);
 388     size_t dimensions[3] = { dimensionSize, 0, 0 };
 389     OCRepPayloadGetPropObjectArray(payload, NS_ATTRIBUTE_TOPIC_LIST, &topicListPayload, dimensions);
 390
 391     for (int i = 0; i < (int) dimensionSize; i++)
 392     {
 393         char * topicName = NULL;
 394         int64_t topicState = 0;
 395
 396         OCRepPayloadGetPropString(topicListPayload[i], NS_ATTRIBUTE_TOPIC_NAME, &topicName);
 397         if (OCRepPayloadGetPropInt(topicListPayload[i], NS_ATTRIBUTE_TOPIC_SELECTION, &topicState))
 398         {
 399             NS_LOG_V(DEBUG, "Topic Name(state):  %s(%d)", topicName, (int)topicState);
 400         }
 401
 402         if (NS_TOPIC_SUBSCRIBED == (NSTopicState) topicState)
 403         {
 404             NSCacheTopicSubData * topicSubData = (NSCacheTopicSubData *) OICMalloc(
 405                     sizeof(NSCacheTopicSubData));
 406             NS_VERIFY_NOT_NULL(topicSubData, NS_FAIL);
 407
 408             OICStrcpy(topicSubData->id, NS_UUID_STRING_SIZE, consumerId);
 409             topicSubData->topicName = topicName;
 410
 411             NSCacheElement * newObj = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
 412
 413             if (!newObj)
 414             {
 415                 NSOICFree(topicSubData->topicName);
 416                 NSOICFree(topicSubData);
 417                 NSOICFree(consumerId);
 418
 419                 // Free topic list
 420                 for (size_t k = 0; k < dimensionSize; k++)
 421                 {
 422                     OCRepPayloadDestroy(topicListPayload[k]);
 423                 }
 424                 OICFree(topicListPayload);
 425
 426                 return NS_FAIL;
 427             }
 428
 429             newObj->data = (NSCacheData *) topicSubData;
 430             newObj->next = NULL;
 431
 432             if (NS_OK != NSProviderStorageWrite(consumerTopicList, newObj))
 433             {
 434                 NS_LOG(ERROR, "Fail to write cache");
 435             }
 436         }
 437     }
 438     NSSendTopicUpdationToConsumer(consumerId);
 439
 440     // Free topic list
 441     for (size_t k = 0; k < dimensionSize; k++)
 442     {
 443         OCRepPayloadDestroy(topicListPayload[k]);
 444     }
 445     OICFree(topicListPayload);
 446
 447     NSOICFree(consumerId);
 448     NS_LOG(DEBUG, "NSPostConsumerTopics() - OUT");
 449     return NS_OK;
 450 }
 451
 452 void * NSTopicSchedule(void * ptr)
 453 {
 454     if (ptr == NULL)
 455     {
 456         NS_LOG(DEBUG, "Create NSTopicSchedule");
 457     }
 458
 459     while (NSIsRunning[TOPIC_SCHEDULER])
 460     {
 461         sem_wait(&NSSemaphore[TOPIC_SCHEDULER]);
 462         pthread_mutex_lock(&NSMutex[TOPIC_SCHEDULER]);
 463
 464         if (NSHeadMsg[TOPIC_SCHEDULER] != NULL)
 465         {
 466             NSTask *node = NSHeadMsg[TOPIC_SCHEDULER];
 467             NSHeadMsg[TOPIC_SCHEDULER] = node->nextTask;
 468
 469             switch (node->taskType)
 470             {
 471                 case TASK_SEND_TOPICS:
 472                     NS_LOG(DEBUG, "CASE TASK_SEND_TOPICS : ");
 473                     NSSendTopicList((OCEntityHandlerRequest*) node->taskData);
 474                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 475                     break;
 476                 case TASK_SUBSCRIBE_TOPIC:
 477                 {
 478                     NS_LOG(DEBUG, "CASE TASK_SUBSCRIBE_TOPIC : ");
 479                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 480                     pthread_mutex_lock(topicSyncResult->mutex);
 481                     NSCacheElement * newObj = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
 482                     NSCacheTopicSubData * subData =
 483                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 484                     if (!newObj)
 485                     {
 486                         NSOICFree(subData->topicName);
 487                         NSOICFree(subData);
 488                         pthread_cond_signal(topicSyncResult->condition);
 489                         pthread_mutex_unlock(topicSyncResult->mutex);
 490                     }
 491                     else
 492                     {
 493                         if (NSProviderStorageRead(registeredTopicList, subData->topicName))
 494                         {
 495                             newObj->data = topicSyncResult->topicData;
 496                             newObj->next = NULL;
 497
 498                             if (NSProviderStorageWrite(consumerTopicList, newObj) == NS_OK)
 499                             {
 500                                 NSSendTopicUpdationToConsumer(subData->id);
 501                                 topicSyncResult->result = NS_OK;
 502                             }
 503                         }
 504                         else
 505                         {
 506                             NSOICFree(subData->topicName);
 507                             NSOICFree(subData);
 508                             NSOICFree(newObj);
 509                         }
 510                     }
 511                     pthread_cond_signal(topicSyncResult->condition);
 512                     pthread_mutex_unlock(topicSyncResult->mutex);
 513                 }
 514                     break;
 515                 case TASK_UNSUBSCRIBE_TOPIC:
 516                 {
 517                     NS_LOG(DEBUG, "CASE TASK_UNSUBSCRIBE_TOPIC : ");
 518                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 519                     pthread_mutex_lock(topicSyncResult->mutex);
 520                     NSCacheTopicSubData * topicSubData =
 521                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 522
 523                     if (NSProviderDeleteConsumerTopic(consumerTopicList, topicSubData) == NS_OK)
 524                     {
 525                         NSSendTopicUpdationToConsumer(topicSubData->id);
 526                         topicSyncResult->result = NS_OK;
 527                     }
 528
 529                     NSOICFree(topicSubData->topicName);
 530                     NSOICFree(topicSubData);
 531                     pthread_cond_signal(topicSyncResult->condition);
 532                     pthread_mutex_unlock(topicSyncResult->mutex);
 533
 534                 }
 535                     break;
 536                 case TASK_REGISTER_TOPIC:
 537                 {
 538                     NS_LOG(DEBUG, "CASE TASK_ADD_TOPIC : ");
 539                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 540
 541                     pthread_mutex_lock(topicSyncResult->mutex);
 542                     topicSyncResult->result = NSRegisterTopic(
 543                             (const char *) topicSyncResult->topicData);
 544                     pthread_cond_signal(topicSyncResult->condition);
 545                     pthread_mutex_unlock(topicSyncResult->mutex);
 546                 }
 547                     break;
 548                 case TASK_UNREGISTER_TOPIC:
 549                 {
 550                     NS_LOG(DEBUG, "CASE_TASK_DELETE_TOPIC : ");
 551                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 552                     pthread_mutex_lock(topicSyncResult->mutex);
 553                     topicSyncResult->result = NSUnregisterTopic(
 554                             (const char *) topicSyncResult->topicData);
 555                     pthread_cond_signal(topicSyncResult->condition);
 556                     pthread_mutex_unlock(topicSyncResult->mutex);
 557                 }
 558                     break;
 559                 case TASK_POST_TOPIC:
 560                 {
 561                     NS_LOG(DEBUG, "TASK_POST_TOPIC : ");
 562                     NSPostConsumerTopics((OCEntityHandlerRequest*) node->taskData);
 563                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 564                 }
 565                     break;
 566                 case TASK_GET_TOPICS:
 567                 {
 568                     NS_LOG(DEBUG, "TASK_GET_TOPICS : ");
 569                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 570                     pthread_mutex_lock(topicSync->mutex);
 571                     NSTopicLL * topics = NSProviderGetTopicsCacheData(registeredTopicList);
 572                     topicSync->topics = topics;
 573                     pthread_cond_signal(topicSync->condition);
 574                     pthread_mutex_unlock(topicSync->mutex);
 575                 }
 576                     break;
 577                 case TAST_GET_CONSUMER_TOPICS:
 578                 {
 579                     NS_LOG(DEBUG, "TASK_GET_CONSUMER_TOPICS : ");
 580                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 581                     pthread_mutex_lock(topicSync->mutex);
 582                     NSTopicLL * topics = NSProviderGetConsumerTopicsCacheData(registeredTopicList,
 583                             consumerTopicList, topicSync->consumerId);
 584                     topicSync->topics = topics;
 585                     pthread_cond_signal(topicSync->condition);
 586                     pthread_mutex_unlock(topicSync->mutex);
 587                 }
 588                     break;
 589                 default:
 590                     break;
 591             }
 592
 593             NSOICFree(node);
 594         }
 595
 596         pthread_mutex_unlock(&NSMutex[TOPIC_SCHEDULER]);
 597     }
 598
 599     NS_LOG(DEBUG, "Destroy NSTopicSchedule");
 600     return NULL;
 601 }