service/notification/src/consumer/NSConsumerScheduler.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 "NSConsumerScheduler.h"
  22
  23 #include <stdlib.h>
  24 #include <stdbool.h>
  25 #include <unistd.h>
  26
  27 #include "oic_malloc.h"
  28 #include "oic_string.h"
  29 #include "ocrandom.h"
  30
  31 #include "NSStructs.h"
  32 #include "NSConstants.h"
  33 #include "NSConsumerCommon.h"
  34 #include "NSConsumerCommunication.h"
  35
  36 #include "NSThread.h"
  37 #include "NSConsumerQueue.h"
  38
  39 #include "NSConsumerDiscovery.h"
  40 #include "NSConsumerInternalTaskController.h"
  41 #include "NSConsumerNetworkEventListener.h"
  42 #include "NSConsumerSystem.h"
  43
  44 void * NSConsumerMsgHandleThreadFunc(void * handle);
  45
  46 void * NSConsumerMsgPushThreadFunc(void * data);
  47
  48 void NSConsumerTaskProcessing(NSTask * task);
  49
  50 NSConsumerThread ** NSGetMsgHandleThreadHandle()
  51 {
  52     static NSConsumerThread * handle = NULL;
  53     return & handle;
  54 }
  55
  56 void NSSetMsgHandleThreadHandle(NSConsumerThread * handle)
  57 {
  58    *(NSGetMsgHandleThreadHandle()) = handle;
  59 }
  60
  61 NSConsumerQueue ** NSGetMsgHandleQueue()
  62 {
  63     static NSConsumerQueue * queue = NULL;
  64     return & queue;
  65 }
  66
  67 void NSSetMsgHandleQueue(NSConsumerQueue * queue)
  68 {
  69    *(NSGetMsgHandleQueue()) = queue;
  70 }
  71
  72 NSResult NSConsumerMessageHandlerInit()
  73 {
  74     NSConsumerThread * handle = NULL;
  75     NSConsumerQueue * queue = NULL;
  76
  77     char * consumerUuid = (char *)OCGetServerInstanceIDString();
  78     NS_VERIFY_NOT_NULL(consumerUuid, NS_ERROR);
  79
  80     NSSetConsumerId(consumerUuid);
  81     NS_LOG_V(DEBUG, "Consumer ID : %s", *NSGetConsumerId());
  82
  83     NS_LOG(DEBUG, "listener init");
  84     NSResult ret = NSConsumerListenerInit();
  85     NS_VERIFY_NOT_NULL(ret == NS_OK ? (void *) 1 : NULL, NS_ERROR);
  86
  87     NS_LOG(DEBUG, "system init");
  88     ret = NSConsumerSystemInit();
  89     NS_VERIFY_NOT_NULL(ret == NS_OK ? (void *) 1 : NULL, NS_ERROR);
  90
  91     NS_LOG(DEBUG, "create queue");
  92     queue = NSCreateQueue();
  93     NS_VERIFY_NOT_NULL(queue, NS_ERROR);
  94     NSSetMsgHandleQueue(queue);
  95
  96     NS_LOG(DEBUG, "queue thread init");
  97     handle = NSThreadInit(NSConsumerMsgHandleThreadFunc, NULL);
  98     NS_VERIFY_NOT_NULL(handle, NS_ERROR);
  99     NSSetMsgHandleThreadHandle(handle);
 100
 101     return NS_OK;
 102 }
 103
 104 NSResult NSConsumerPushEvent(NSTask * task)
 105 {
 106     NSConsumerThread * thread = NSThreadInit(NSConsumerMsgPushThreadFunc, (void *) task);
 107     NS_VERIFY_NOT_NULL(thread, NS_ERROR);
 108
 109     NSDestroyThreadHandle(thread);
 110     NSOICFree(thread);
 111
 112     return NS_OK;
 113 }
 114
 115 void NSConsumerMessageHandlerExit()
 116 {
 117
 118     NSConsumerListenerTermiate();
 119     NSCancelAllSubscription();
 120
 121     NSConsumerThread * thread = *(NSGetMsgHandleThreadHandle());
 122     NSThreadStop(thread);
 123     NSSetMsgHandleThreadHandle(NULL);
 124
 125     NSConsumerQueue * queue = *(NSGetMsgHandleQueue());
 126     NSDestroyQueue(queue);
 127     NSSetMsgHandleQueue(NULL);
 128
 129     NSDestroyInternalCachedList();
 130 }
 131
 132 void * NSConsumerMsgHandleThreadFunc(void * threadHandle)
 133 {
 134     NSConsumerQueue * queue = *(NSGetMsgHandleQueue());;
 135     NSConsumerQueueObject * obj = NULL;
 136
 137     NS_LOG(DEBUG, "create thread for consumer message handle");
 138     NSConsumerThread * queueHandleThread = (NSConsumerThread *) threadHandle;
 139     NS_VERIFY_NOT_NULL(queueHandleThread, NULL);
 140
 141     while (true)
 142     {
 143         if (!queue)
 144         {
 145             queue = *(NSGetMsgHandleQueue());
 146             usleep(2000);
 147             continue;
 148         }
 149
 150         if (!queueHandleThread->isStarted && NSIsQueueEmpty(queue))
 151         {
 152             NS_LOG(ERROR, "msg handler thread will be terminated");
 153             break;
 154         }
 155
 156         if (NSIsQueueEmpty(queue))
 157         {
 158             usleep(2000);
 159             continue;
 160         }
 161
 162         NSThreadLock(queueHandleThread);
 163         NS_LOG(DEBUG, "msg handler working");
 164         obj = NSPopQueue(queue);
 165
 166         if (obj)
 167         {
 168             NSConsumerTaskProcessing((NSTask *)(obj->data));
 169             NSOICFree(obj);
 170         }
 171
 172         NSThreadUnlock(queueHandleThread);
 173
 174     }
 175
 176     return NULL;
 177 }
 178
 179 void * NSConsumerMsgPushThreadFunc(void * data)
 180 {
 181     NSConsumerQueueObject * obj = NULL;
 182     NSConsumerQueue * queue = NULL;
 183
 184     NS_LOG(DEBUG, "get queueThread handle");
 185     NSConsumerThread * msgHandleThread = *(NSGetMsgHandleThreadHandle());
 186     NS_VERIFY_NOT_NULL_WITH_POST_CLEANING(msgHandleThread, NULL, NSOICFree(data));
 187
 188     NS_LOG(DEBUG, "create queue object");
 189     obj = (NSConsumerQueueObject *)OICMalloc(sizeof(NSConsumerQueueObject));
 190     NS_VERIFY_NOT_NULL_WITH_POST_CLEANING(obj, NULL, NSOICFree(data));
 191
 192     obj->data = data;
 193     obj->next = NULL;
 194
 195     NSThreadLock(msgHandleThread);
 196
 197     queue = *(NSGetMsgHandleQueue());
 198     if (!queue)
 199     {
 200         NS_LOG(ERROR, "NSQueue is null. can not insert to queue");
 201         NSOICFree(data);
 202         NSOICFree(obj);
 203     }
 204     else
 205     {
 206         NSPushConsumerQueue(queue, obj);
 207     }
 208
 209     NSThreadUnlock(msgHandleThread);
 210
 211     return NULL;
 212 }
 213
 214 void NSProviderDeletedPostClean(
 215         NSTask * task, NSProvider_internal * prov1, NSProvider_internal * prov2)
 216 {
 217     if (task && task->taskData)
 218     {
 219         if (task->taskType == TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL)
 220         {
 221             NSRemoveProvider((NSProvider *) task->taskData);
 222         }
 223         else if (task->taskType == TASK_CONSUMER_PROVIDER_DELETED)
 224         {
 225             NSOICFree(task->taskData);
 226         }
 227         NSOICFree(task);
 228     }
 229
 230     if (prov1)
 231     {
 232         NSRemoveProvider_internal(prov1);
 233     }
 234
 235     if (prov2)
 236     {
 237         NSRemoveProvider_internal(prov2);
 238     }
 239 }
 240
 241 void NSConsumerTaskProcessing(NSTask * task)
 242 {
 243     switch (task->taskType)
 244     {
 245         case TASK_EVENT_CONNECTED:
 246         case TASK_EVENT_CONNECTED_TCP:
 247         case TASK_CONSUMER_REQ_DISCOVER:
 248         {
 249             NSConsumerDiscoveryTaskProcessing(task);
 250             break;
 251         }
 252         case TASK_CONSUMER_REQ_SUBSCRIBE:
 253         {
 254             NSProvider_internal * prov =
 255                     NSConsumerFindNSProvider(((NSProvider *)task->taskData)->providerId);
 256             NS_VERIFY_NOT_NULL_V(prov);
 257             NSTask * subTask = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE, prov);
 258             NS_VERIFY_NOT_NULL_V(subTask);
 259             NSConsumerCommunicationTaskProcessing(subTask);
 260
 261             NSRemoveProvider((NSProvider *)task->taskData);
 262             NSOICFree(task);
 263             break;
 264         }
 265         case TASK_SEND_SYNCINFO:
 266         case TASK_CONSUMER_REQ_TOPIC_LIST:
 267         case TASK_CONSUMER_SELECT_TOPIC_LIST:
 268         {
 269             NSConsumerCommunicationTaskProcessing(task);
 270             break;
 271         }
 272         case TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL:
 273         case TASK_CONSUMER_PROVIDER_DELETED:
 274         {
 275             NSProvider_internal * data = NULL;
 276
 277             if (task->taskType == TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL)
 278             {
 279                 data = NSConsumerFindNSProvider(((NSProvider *) task->taskData)->providerId);
 280                 NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
 281                         data, NSProviderDeletedPostClean(task, NULL, NULL));
 282             }
 283             else if (task->taskType == TASK_CONSUMER_PROVIDER_DELETED)
 284             {
 285                 data = NSFindProviderFromAddr((OCDevAddr *) task->taskData);
 286                 NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
 287                         data, NSProviderDeletedPostClean(task, NULL, NULL));
 288             }
 289
 290             NSProvider_internal * data2 = NSCopyProvider_internal(data);
 291             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
 292                         data2, NSProviderDeletedPostClean(task, data, NULL));
 293
 294             NSTask * conTask = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, data);
 295             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
 296                         conTask, NSProviderDeletedPostClean(task, data, data2));
 297             NSConsumerCommunicationTaskProcessing(conTask);
 298
 299             NSTask * conTask2 = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, data2);
 300             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
 301                         conTask, NSProviderDeletedPostClean(task, NULL, data2));
 302             NSConsumerInternalTaskProcessing(conTask2);
 303
 304             NSProviderDeletedPostClean(task, NULL, NULL);
 305             break;
 306         }
 307         case TASK_RECV_SYNCINFO:
 308         case TASK_CONSUMER_RECV_MESSAGE:
 309         case TASK_CONSUMER_SENT_REQ_OBSERVE:
 310         case TASK_CONSUMER_RECV_PROVIDER_CHANGED:
 311         case TASK_MAKE_SYNCINFO:
 312         case TASK_CONSUMER_REQ_TOPIC_URI:
 313         case TASK_CONSUMER_RECV_TOPIC_LIST:
 314         {
 315             NSConsumerInternalTaskProcessing(task);
 316             break;
 317         }
 318         case TASK_CONSUMER_PROVIDER_DISCOVERED:
 319         {
 320             NSTask * getTopicTask = (NSTask *)OICMalloc(sizeof(NSTask));
 321             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(getTopicTask,
 322                         NSRemoveProvider_internal((void *) task->taskData));
 323             getTopicTask->nextTask = NULL;
 324             getTopicTask->taskData =
 325                     (void *) NSCopyProvider_internal((NSProvider_internal *) task->taskData);
 326             getTopicTask->taskType = TASK_CONSUMER_REQ_TOPIC_LIST;
 327             NSConsumerCommunicationTaskProcessing(getTopicTask);
 328             NSConsumerInternalTaskProcessing(task);
 329             break;
 330         }
 331         default:
 332         {
 333             NS_LOG(ERROR, "Unknown type of task");
 334             break;
 335         }
 336     }
 337 }
 338
 339 NSProvider_internal * NSConsumerFindNSProvider(const char * providerId)
 340 {
 341     NS_VERIFY_NOT_NULL(providerId, NULL);
 342
 343     return NSProviderCacheFind(providerId);
 344 }