47224cd152e2af5fdee34d008bb6b1e7c6d70068
[platform/upstream/iotivity.git] / 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     uint8_t uuid[UUID_SIZE] = {0,};
78     char uuidString[UUID_STRING_SIZE] = {0,};
79     OCRandomUuidResult randomRet = OCGenerateUuid(uuid);
80     NS_VERIFY_NOT_NULL(randomRet == RAND_UUID_OK ? (void *) 1 : NULL, NS_ERROR);
81     randomRet = OCConvertUuidToString(uuid, uuidString);
82     NS_VERIFY_NOT_NULL(randomRet == RAND_UUID_OK ? (void *) 1 : NULL, NS_ERROR);
83
84     NSSetConsumerId(uuidString);
85     NS_LOG_V(DEBUG, "Consumer ID : %s", *NSGetConsumerId());
86
87     NS_LOG(DEBUG, "listener init");
88     NSResult ret = NSConsumerListenerInit();
89     NS_VERIFY_NOT_NULL(ret == NS_OK ? (void *) 1 : NULL, NS_ERROR);
90
91     NS_LOG(DEBUG, "system init");
92     ret = NSConsumerSystemInit();
93     NS_VERIFY_NOT_NULL(ret == NS_OK ? (void *) 1 : NULL, NS_ERROR);
94
95     NS_LOG(DEBUG, "queue thread init");
96     handle = NSThreadInit(NSConsumerMsgHandleThreadFunc, NULL);
97     NS_VERIFY_NOT_NULL(handle, NS_ERROR);
98     NSSetMsgHandleThreadHandle(handle);
99
100     NS_LOG(DEBUG, "create queue");
101     queue = NSCreateQueue();
102     NS_VERIFY_NOT_NULL(queue, NS_ERROR);
103     NSSetMsgHandleQueue(queue);
104
105     return NS_OK;
106 }
107
108 NSResult NSConsumerPushEvent(NSTask * task)
109 {
110     NSConsumerThread * thread = NSThreadInit(NSConsumerMsgPushThreadFunc, (void *) task);
111     NS_VERIFY_NOT_NULL(thread, NS_ERROR);
112
113     NSDestroyThreadHandle(thread);
114
115     return NS_OK;
116 }
117
118 void NSConsumerMessageHandlerExit()
119 {
120
121     NSConsumerListenerTermiate();
122     NSCancelAllSubscription();
123     NSThreadStop(*(NSGetMsgHandleThreadHandle()));
124     NSDestroyQueue(*(NSGetMsgHandleQueue()));
125     NSSetMsgHandleQueue(NULL);
126
127     NSDestroyMessageCacheList();
128     NSDestroyProviderCacheList();
129 }
130
131 void * NSConsumerMsgHandleThreadFunc(void * threadHandle)
132 {
133     NSConsumerQueue * queue = *(NSGetMsgHandleQueue());;
134     NSConsumerQueueObject * obj = NULL;
135
136     NS_LOG(DEBUG, "create thread for consumer message handle");
137     NSConsumerThread * queueHandleThread = (NSConsumerThread *) threadHandle;
138     NS_VERIFY_NOT_NULL(queueHandleThread, NULL);
139
140     while (true)
141     {
142         if (!queue)
143         {
144             queue = *(NSGetMsgHandleQueue());
145             usleep(2000);
146             continue;
147         }
148
149         if (!queueHandleThread->isStarted && NSIsQueueEmpty(queue))
150         {
151             NS_LOG(ERROR, "msg handler thread will be terminated");
152             break;
153         }
154
155         if (NSIsQueueEmpty(queue))
156         {
157             usleep(2000);
158             continue;
159         }
160
161         NSThreadLock(queueHandleThread);
162         NS_LOG(DEBUG, "msg handler working");
163         obj = NSPopQueue(queue);
164
165         if (obj)
166         {
167             NSConsumerTaskProcessing((NSTask *)(obj->data));
168         }
169
170         NSThreadUnlock(queueHandleThread);
171
172     }
173
174     return NULL;
175 }
176
177 void * NSConsumerMsgPushThreadFunc(void * data)
178 {
179     NSThreadDetach();
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         NSPushQueue(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         {
268             NSConsumerCommunicationTaskProcessing(task);
269             break;
270         }
271         case TASK_CONSUMER_GET_TOPIC_LIST:
272         case TASK_CONSUMER_SELECT_TOPIC_LIST:
273         {
274             NSConsumerCommunicationTaskProcessing(task);
275             break;
276         }
277         case TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL:
278         case TASK_CONSUMER_PROVIDER_DELETED:
279         {
280             NSProvider_internal * data = NULL;
281
282             if (task->taskType == TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL)
283             {
284                 data = NSConsumerFindNSProvider(((NSProvider *) task->taskData)->providerId);
285                 NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
286                         data, NSProviderDeletedPostClean(task, NULL, NULL));
287             }
288             else if (task->taskType == TASK_CONSUMER_PROVIDER_DELETED)
289             {
290                 data = NSFindProviderFromAddr((OCDevAddr *) task->taskData);
291                 NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
292                         data, NSProviderDeletedPostClean(task, NULL, NULL));
293             }
294
295             NSProvider_internal * data2 = NSCopyProvider_internal(data);
296             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
297                         data2, NSProviderDeletedPostClean(task, data, NULL));
298
299             NSTask * conTask = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, data);
300             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
301                         conTask, NSProviderDeletedPostClean(task, data, data2));
302             NSConsumerCommunicationTaskProcessing(conTask);
303
304             NSTask * conTask2 = NSMakeTask(TASK_CONSUMER_REQ_SUBSCRIBE_CANCEL, data2);
305             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(
306                         conTask, NSProviderDeletedPostClean(task, NULL, data2));
307             NSConsumerInternalTaskProcessing(conTask2);
308
309             NSProviderDeletedPostClean(task, NULL, NULL);
310             break;
311         }
312         case TASK_RECV_SYNCINFO:
313         case TASK_CONSUMER_RECV_MESSAGE:
314         case TASK_CONSUMER_SENT_REQ_OBSERVE:
315         case TASK_CONSUMER_RECV_PROVIDER_CHANGED:
316         case TASK_MAKE_SYNCINFO:
317         case TASK_CONSUMER_REQ_TOPIC_URI:
318         case TASK_CONSUMER_RECV_TOPIC_LIST:
319         {
320             NSConsumerInternalTaskProcessing(task);
321             break;
322         }
323         case TASK_CONSUMER_PROVIDER_DISCOVERED:
324         {
325             NSTask * getTopicTask = (NSTask *)OICMalloc(sizeof(NSTask));
326             NS_VERIFY_NOT_NULL_WITH_POST_CLEANING_V(getTopicTask,
327                         NSRemoveProvider_internal((NSProvider_internal *) task->taskData));
328             getTopicTask->nextTask = NULL;
329             getTopicTask->taskData =
330                     (void *) NSCopyProvider_internal((NSProvider_internal *) task->taskData);
331             getTopicTask->taskType = TASK_CONSUMER_REQ_TOPIC_LIST;
332             NSConsumerCommunicationTaskProcessing(getTopicTask);
333             NSConsumerInternalTaskProcessing(task);
334             break;
335         }
336         default:
337         {
338             NS_LOG(ERROR, "Unknown type of task");
339             break;
340         }
341     }
342 }
343
344 NSMessage * NSConsumerFindNSMessage(const char* messageId)
345 {
346     NS_VERIFY_NOT_NULL(messageId, NULL);
347
348     return NSMessageCacheFind(messageId);
349 }
350
351 NSProvider_internal * NSConsumerFindNSProvider(const char * providerId)
352 {
353     NS_VERIFY_NOT_NULL(providerId, NULL);
354
355     return NSProviderCacheFind(providerId);
356 }