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                 return NS_FAIL;
 416             }
 417
 418             newObj->data = (NSCacheData *) topicSubData;
 419             newObj->next = NULL;
 420
 421             if (NS_OK != NSProviderStorageWrite(consumerTopicList, newObj))
 422             {
 423                 NS_LOG(ERROR, "Fail to write cache");
 424             }
 425         }
 426     }
 427     NSSendTopicUpdationToConsumer(consumerId);
 428     NSOICFree(consumerId);
 429     NS_LOG(DEBUG, "NSPostConsumerTopics() - OUT");
 430     return NS_OK;
 431 }
 432
 433 void * NSTopicSchedule(void * ptr)
 434 {
 435     if (ptr == NULL)
 436     {
 437         NS_LOG(DEBUG, "Create NSTopicSchedule");
 438     }
 439
 440     while (NSIsRunning[TOPIC_SCHEDULER])
 441     {
 442         sem_wait(&NSSemaphore[TOPIC_SCHEDULER]);
 443         pthread_mutex_lock(&NSMutex[TOPIC_SCHEDULER]);
 444
 445         if (NSHeadMsg[TOPIC_SCHEDULER] != NULL)
 446         {
 447             NSTask *node = NSHeadMsg[TOPIC_SCHEDULER];
 448             NSHeadMsg[TOPIC_SCHEDULER] = node->nextTask;
 449
 450             switch (node->taskType)
 451             {
 452                 case TASK_SEND_TOPICS:
 453                     NS_LOG(DEBUG, "CASE TASK_SEND_TOPICS : ");
 454                     NSSendTopicList((OCEntityHandlerRequest*) node->taskData);
 455                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 456                     break;
 457                 case TASK_SUBSCRIBE_TOPIC:
 458                 {
 459                     NS_LOG(DEBUG, "CASE TASK_SUBSCRIBE_TOPIC : ");
 460                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 461                     pthread_mutex_lock(topicSyncResult->mutex);
 462                     NSCacheElement * newObj = (NSCacheElement *) OICMalloc(sizeof(NSCacheElement));
 463                     NSCacheTopicSubData * subData =
 464                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 465                     if (!newObj)
 466                     {
 467                         NSOICFree(subData->topicName);
 468                         NSOICFree(subData);
 469                         pthread_cond_signal(topicSyncResult->condition);
 470                         pthread_mutex_unlock(topicSyncResult->mutex);
 471                     }
 472                     else
 473                     {
 474                         if (NSProviderStorageRead(registeredTopicList, subData->topicName))
 475                         {
 476                             newObj->data = topicSyncResult->topicData;
 477                             newObj->next = NULL;
 478
 479                             if (NSProviderStorageWrite(consumerTopicList, newObj) == NS_OK)
 480                             {
 481                                 NSSendTopicUpdationToConsumer(subData->id);
 482                                 topicSyncResult->result = NS_OK;
 483                             }
 484                         }
 485                         else
 486                         {
 487                             NSOICFree(subData->topicName);
 488                             NSOICFree(subData);
 489                             NSOICFree(newObj);
 490                         }
 491                     }
 492                     pthread_cond_signal(topicSyncResult->condition);
 493                     pthread_mutex_unlock(topicSyncResult->mutex);
 494                 }
 495                     break;
 496                 case TASK_UNSUBSCRIBE_TOPIC:
 497                 {
 498                     NS_LOG(DEBUG, "CASE TASK_UNSUBSCRIBE_TOPIC : ");
 499                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 500                     pthread_mutex_lock(topicSyncResult->mutex);
 501                     NSCacheTopicSubData * topicSubData =
 502                             (NSCacheTopicSubData *) topicSyncResult->topicData;
 503
 504                     if (NSProviderDeleteConsumerTopic(consumerTopicList, topicSubData) == NS_OK)
 505                     {
 506                         NSSendTopicUpdationToConsumer(topicSubData->id);
 507                         topicSyncResult->result = NS_OK;
 508                     }
 509
 510                     NSOICFree(topicSubData->topicName);
 511                     NSOICFree(topicSubData);
 512                     pthread_cond_signal(topicSyncResult->condition);
 513                     pthread_mutex_unlock(topicSyncResult->mutex);
 514
 515                 }
 516                     break;
 517                 case TASK_REGISTER_TOPIC:
 518                 {
 519                     NS_LOG(DEBUG, "CASE TASK_ADD_TOPIC : ");
 520                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 521
 522                     pthread_mutex_lock(topicSyncResult->mutex);
 523                     topicSyncResult->result = NSRegisterTopic(
 524                             (const char *) topicSyncResult->topicData);
 525                     pthread_cond_signal(topicSyncResult->condition);
 526                     pthread_mutex_unlock(topicSyncResult->mutex);
 527                 }
 528                     break;
 529                 case TASK_UNREGISTER_TOPIC:
 530                 {
 531                     NS_LOG(DEBUG, "CASE_TASK_DELETE_TOPIC : ");
 532                     NSTopicSyncResult * topicSyncResult = (NSTopicSyncResult *) node->taskData;
 533                     pthread_mutex_lock(topicSyncResult->mutex);
 534                     topicSyncResult->result = NSUnregisterTopic(
 535                             (const char *) topicSyncResult->topicData);
 536                     pthread_cond_signal(topicSyncResult->condition);
 537                     pthread_mutex_unlock(topicSyncResult->mutex);
 538                 }
 539                     break;
 540                 case TASK_POST_TOPIC:
 541                 {
 542                     NS_LOG(DEBUG, "TASK_POST_TOPIC : ");
 543                     NSPostConsumerTopics((OCEntityHandlerRequest*) node->taskData);
 544                     NSFreeOCEntityHandlerRequest((OCEntityHandlerRequest*) node->taskData);
 545                 }
 546                     break;
 547                 case TASK_GET_TOPICS:
 548                 {
 549                     NS_LOG(DEBUG, "TASK_GET_TOPICS : ");
 550                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 551                     pthread_mutex_lock(topicSync->mutex);
 552                     NSTopicLL * topics = NSProviderGetTopicsCacheData(registeredTopicList);
 553                     topicSync->topics = topics;
 554                     pthread_cond_signal(topicSync->condition);
 555                     pthread_mutex_unlock(topicSync->mutex);
 556                 }
 557                     break;
 558                 case TAST_GET_CONSUMER_TOPICS:
 559                 {
 560                     NS_LOG(DEBUG, "TASK_GET_CONSUMER_TOPICS : ");
 561                     NSTopicSync * topicSync = (NSTopicSync *) node->taskData;
 562                     pthread_mutex_lock(topicSync->mutex);
 563                     NSTopicLL * topics = NSProviderGetConsumerTopicsCacheData(registeredTopicList,
 564                             consumerTopicList, topicSync->consumerId);
 565                     topicSync->topics = topics;
 566                     pthread_cond_signal(topicSync->condition);
 567                     pthread_mutex_unlock(topicSync->mutex);
 568                 }
 569                     break;
 570                 default:
 571                     break;
 572             }
 573
 574             NSOICFree(node);
 575         }
 576
 577         pthread_mutex_unlock(&NSMutex[TOPIC_SCHEDULER]);
 578     }
 579
 580     NS_LOG(DEBUG, "Destroy NSTopicSchedule");
 581     return NULL;
 582 }