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