Coverage Report

Created: 2026-08-13 06:39

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/src/open62541/src/client/ua_client_subscriptions.c
Line
Count
Source
1
/* This Source Code Form is subject to the terms of the Mozilla Public
2
 * License, v. 2.0. If a copy of the MPL was not distributed with this
3
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
4
 *
5
 *    Copyright 2015-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
6
 *    Copyright 2015 (c) Oleksiy Vasylyev
7
 *    Copyright 2016 (c) Sten Grüner
8
 *    Copyright 2017-2018 (c) Thomas Stalder, Blue Time Concept SA
9
 *    Copyright 2016-2017 (c) Florian Palm
10
 *    Copyright 2017 (c) Frank Meerkötter
11
 *    Copyright 2017 (c) Stefan Profanter, fortiss GmbH
12
 *    Copyright 2026 (c) o6 Automation GmbH (Author: Julius Pfrommer)
13
 */
14
15
#include <open62541/client_highlevel.h>
16
#include <open62541/client_highlevel_async.h>
17
18
#include "ua_client_internal.h"
19
20
struct UA_Client_MonitoredItem_ForDelete {
21
    UA_Client *client;
22
    UA_Client_Subscription *sub;
23
    UA_UInt32 *monitoredItemId;
24
};
25
26
/*****************/
27
/* Subscriptions */
28
/*****************/
29
30
/* For ZIP_TREE we use clientHandle comparison */
31
static enum ZIP_CMP
32
0
UA_ClientHandle_cmp(const void *a, const void *b) {
33
0
    const UA_Client_MonitoredItem *aa = (const UA_Client_MonitoredItem *)a;
34
0
    const UA_Client_MonitoredItem *bb = (const UA_Client_MonitoredItem *)b;
35
0
    if(aa->parameters.clientHandle < bb->parameters.clientHandle)
36
0
        return ZIP_CMP_LESS;
37
0
    if(aa->parameters.clientHandle > bb->parameters.clientHandle)
38
0
        return ZIP_CMP_MORE;
39
0
    return ZIP_CMP_EQ;
40
0
}
41
42
0
ZIP_FUNCTIONS(MonitorItemsTree, UA_Client_MonitoredItem, zipfields,
Unexecuted instantiation: ua_client_subscriptions.c:MonitorItemsTree_ZIP_INSERT
Unexecuted instantiation: ua_client_subscriptions.c:MonitorItemsTree_ZIP_REMOVE
Unexecuted instantiation: ua_client_subscriptions.c:MonitorItemsTree_ZIP_ITER_INNER
43
              UA_Client_MonitoredItem, zipfields, UA_ClientHandle_cmp)
44
45
static UA_Client_MonitoredItem *
46
0
findMonitoredItemByHandle(UA_Client_Subscription *sub, UA_UInt32 clientHandle) {
47
0
    UA_Client_MonitoredItem dummy;
48
0
    memset(&dummy, 0, sizeof(dummy));
49
0
    dummy.parameters.clientHandle = clientHandle;
50
0
    return ZIP_FIND(MonitorItemsTree, &sub->monitoredItems, &dummy);
51
0
}
52
53
static void *
54
0
MonitoredItem_findPendingByHandle(void *data, UA_Client_MonitoredItem *mon) {
55
0
    UA_UInt32 clientHandle = *(UA_UInt32*)data;
56
0
    if(mon->pendingParameters.clientHandle == clientHandle)
57
0
        return mon;
58
0
    return NULL;
59
0
}
60
61
static void
62
MonitoredItem_delete(UA_Client *client, UA_Client_Subscription *sub,
63
                     UA_Client_MonitoredItem *mon);
64
65
static void
66
Subscription_create(UA_Client *client, UA_Client_Subscription *newSub,
67
0
                    UA_CreateSubscriptionResponse *response) {
68
0
    UA_LOCK_ASSERT(&client->clientMutex);
69
70
0
    UA_EventLoop *el = client->config.eventLoop;
71
72
0
    newSub->subscriptionId = response->subscriptionId;
73
0
    newSub->sequenceNumber = 0;
74
0
    newSub->lastActivity = el->dateTime_nowMonotonic(el);
75
0
    newSub->publishingInterval = response->revisedPublishingInterval;
76
0
    newSub->maxKeepAliveCount = response->revisedMaxKeepAliveCount;
77
0
    ZIP_INIT(&newSub->monitoredItems);
78
0
    LIST_INSERT_HEAD(&client->subscriptions, newSub, listEntry);
79
80
    /* Immediately send the first publish requests if there are none
81
     * outstanding */
82
0
    __Client_Subscriptions_backgroundPublish(client);
83
0
}
84
85
static void
86
Subscriptions_create_handler(UA_Client *client, void *data,
87
0
                             UA_UInt32 requestId, void *r) {
88
0
    UA_LOCK_ASSERT(&client->clientMutex);
89
90
0
    UA_CreateSubscriptionResponse *response = (UA_CreateSubscriptionResponse *)r;
91
0
    CustomCallback *cc = (CustomCallback *)data;
92
0
    UA_Client_Subscription *newSub = (UA_Client_Subscription *)cc->clientData;
93
0
    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
94
0
        UA_free(newSub);
95
0
        goto cleanup;
96
0
    }
97
98
    /* Prepare the internal representation */
99
0
    Subscription_create(client, newSub, response);
100
101
0
cleanup:
102
0
    if(cc->callback.createSubscription)
103
0
        cc->callback.createSubscription(client, cc->userData, requestId, response);
104
0
    UA_free(cc);
105
0
}
106
107
UA_CreateSubscriptionResponse
108
UA_Client_Subscriptions_create(UA_Client *client,
109
                               const UA_CreateSubscriptionRequest request,
110
                               void *subscriptionContext,
111
                               UA_Client_StatusChangeNotificationCallback statusChangeCallback,
112
0
                               UA_Client_DeleteSubscriptionCallback deleteCallback) {
113
0
    lockClient(client);
114
115
0
    UA_CreateSubscriptionResponse response;
116
0
    UA_Client_Subscription *sub = (UA_Client_Subscription *)
117
0
        UA_malloc(sizeof(UA_Client_Subscription));
118
0
    if(!sub) {
119
0
        UA_CreateSubscriptionResponse_init(&response);
120
0
        response.responseHeader.serviceResult = UA_STATUSCODE_BADOUTOFMEMORY;
121
0
        unlockClient(client);
122
0
        return response;
123
0
    }
124
0
    sub->context = subscriptionContext;
125
0
    sub->statusChangeCallback = statusChangeCallback;
126
0
    sub->deleteCallback = deleteCallback;
127
128
    /* Send the request as a synchronous service call */
129
0
    __Client_Service(client, &request, &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONREQUEST],
130
0
                     &response, &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONRESPONSE]);
131
0
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
132
0
        UA_free(sub);
133
0
        unlockClient(client);
134
0
        return response;
135
0
    }
136
137
0
    Subscription_create(client, sub, &response);
138
139
0
    unlockClient(client);
140
0
    return response;
141
0
}
142
143
UA_StatusCode
144
UA_Client_Subscriptions_create_async(UA_Client *client, const UA_CreateSubscriptionRequest request,
145
                                     void *subscriptionContext,
146
                                     UA_Client_StatusChangeNotificationCallback statusChangeCallback,
147
                                     UA_Client_DeleteSubscriptionCallback deleteCallback,
148
                                     UA_ClientAsyncCreateSubscriptionCallback createCallback,
149
0
                                     void *userdata, UA_UInt32 *requestId) {
150
0
    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
151
0
    if(!cc)
152
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
153
154
0
    UA_Client_Subscription *sub = (UA_Client_Subscription *)
155
0
        UA_malloc(sizeof(UA_Client_Subscription));
156
0
    if(!sub) {
157
0
        UA_free(cc);
158
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
159
0
    }
160
0
    sub->context = subscriptionContext;
161
0
    sub->statusChangeCallback = statusChangeCallback;
162
0
    sub->deleteCallback = deleteCallback;
163
164
0
    cc->callback.createSubscription = createCallback;
165
0
    cc->userData = userdata;
166
0
    cc->clientData = sub;
167
168
    /* Send the request as asynchronous service call */
169
0
    UA_StatusCode res =
170
0
        __UA_Client_AsyncService(client, &request,
171
0
                                 &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONREQUEST],
172
0
                                 Subscriptions_create_handler,
173
0
                                 &UA_TYPES[UA_TYPES_CREATESUBSCRIPTIONRESPONSE],
174
0
                                 cc, requestId);
175
0
    if(res != UA_STATUSCODE_GOOD) {
176
0
        UA_free(cc);
177
0
        UA_free(sub);
178
0
    }
179
0
    return res;
180
0
}
181
182
static UA_Client_Subscription *
183
0
findSubscriptionById(const UA_Client *client, UA_UInt32 subscriptionId) {
184
0
    UA_Client_Subscription *sub = NULL;
185
0
    LIST_FOREACH(sub, &client->subscriptions, listEntry) {
186
0
        if(sub->subscriptionId == subscriptionId)
187
0
            break;
188
0
    }
189
0
    return sub;
190
0
}
191
192
static void
193
Subscription_modify(UA_Client *client, UA_Client_Subscription *sub,
194
0
                    const UA_ModifySubscriptionResponse *response) {
195
0
    sub->publishingInterval = response->revisedPublishingInterval;
196
0
    sub->maxKeepAliveCount = response->revisedMaxKeepAliveCount;
197
0
}
198
199
static void
200
Subscription_modify_handler(UA_Client *client, void *data,
201
0
                            UA_UInt32 requestId, void *r) {
202
0
    UA_LOCK_ASSERT(&client->clientMutex);
203
204
0
    UA_ModifySubscriptionResponse *response = (UA_ModifySubscriptionResponse *)r;
205
0
    CustomCallback *cc = (CustomCallback *)data;
206
0
    UA_Client_Subscription *sub =
207
0
        findSubscriptionById(client, (UA_UInt32)(uintptr_t)cc->clientData);
208
0
    if(sub) {
209
0
        Subscription_modify(client, sub, response);
210
0
    } else {
211
0
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
212
0
                    "No internal representation of subscription %" PRIu32,
213
0
                    (UA_UInt32)(uintptr_t)cc->clientData);
214
0
    }
215
216
0
    if(cc->callback.modifySubscription)
217
0
        cc->callback.modifySubscription(client, cc->userData, requestId, response);
218
0
    UA_free(cc);
219
0
}
220
221
UA_StatusCode
222
UA_Client_Subscriptions_getContext(UA_Client *client, UA_UInt32 subscriptionId,
223
0
                                   void **subContext) {
224
0
    if(!client || !subContext)
225
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
226
227
0
    lockClient(client);
228
0
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
229
0
    if(!sub) {
230
0
        unlockClient(client);
231
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
232
0
    }
233
234
0
    *subContext = sub->context;
235
0
    unlockClient(client);
236
0
    return UA_STATUSCODE_GOOD;
237
0
}
238
239
UA_StatusCode
240
UA_Client_Subscriptions_setContext(UA_Client *client, UA_UInt32 subscriptionId,
241
0
                                   void *subContext) {
242
0
    if(!client)
243
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
244
245
0
    lockClient(client);
246
0
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
247
0
    if(!sub) {
248
0
        unlockClient(client);
249
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
250
0
    }
251
252
0
    sub->context = subContext;
253
0
    unlockClient(client);
254
0
    return UA_STATUSCODE_GOOD;
255
0
}
256
257
UA_ModifySubscriptionResponse
258
UA_Client_Subscriptions_modify(UA_Client *client,
259
0
                               const UA_ModifySubscriptionRequest request) {
260
0
    UA_ModifySubscriptionResponse response;
261
0
    UA_ModifySubscriptionResponse_init(&response);
262
263
    /* Find the internal representation */
264
0
    lockClient(client);
265
0
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
266
0
    if(!sub) {
267
0
        unlockClient(client);
268
0
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
269
0
        return response;
270
0
    }
271
272
    /* Call the service */
273
0
    __Client_Service(client,
274
0
                     &request, &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONREQUEST],
275
0
                     &response, &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONRESPONSE]);
276
277
    /* Adjust the internal representation. Lookup again for thread-safety. */
278
0
    sub = findSubscriptionById(client, request.subscriptionId);
279
0
    if(!sub) {
280
0
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
281
0
        unlockClient(client);
282
0
        return response;
283
0
    }
284
0
    Subscription_modify(client, sub, &response);
285
0
    unlockClient(client);
286
0
    return response;
287
0
}
288
289
UA_StatusCode
290
UA_Client_Subscriptions_modify_async(UA_Client *client,
291
                                     const UA_ModifySubscriptionRequest request,
292
                                     UA_ClientAsyncModifySubscriptionCallback callback,
293
0
                                     void *userdata, UA_UInt32 *requestId) {
294
0
    lockClient(client);
295
296
0
    UA_StatusCode res = __Client_AsyncServiceAdmission(client);
297
0
    if(res != UA_STATUSCODE_GOOD) {
298
0
        unlockClient(client);
299
0
        return res;
300
0
    }
301
302
0
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
303
0
    if(!sub) {
304
0
        unlockClient(client);
305
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
306
0
    }
307
308
0
    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
309
0
    if(!cc) {
310
0
        unlockClient(client);
311
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
312
0
    }
313
314
0
    cc->clientData = (void *)(uintptr_t)request.subscriptionId;
315
0
    cc->userData = userdata;
316
0
    cc->callback.modifySubscription = callback;
317
318
0
    res = __Client_AsyncServiceAdmitted(
319
0
        client, &request, &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONREQUEST],
320
0
        Subscription_modify_handler,
321
0
        &UA_TYPES[UA_TYPES_MODIFYSUBSCRIPTIONRESPONSE], cc, requestId);
322
0
    if(res != UA_STATUSCODE_GOOD)
323
0
        UA_free(cc);
324
325
0
    unlockClient(client);
326
0
    return res;
327
0
}
328
329
static void *
330
0
MonitoredItem_delete_wrapper(void *data, UA_Client_MonitoredItem *mon) {
331
0
    struct UA_Client_MonitoredItem_ForDelete *deleteMonitoredItem =
332
0
        (struct UA_Client_MonitoredItem_ForDelete *)data;
333
0
    if(deleteMonitoredItem != NULL) {
334
0
        if(deleteMonitoredItem->monitoredItemId != NULL &&
335
0
           (mon->monitoredItemId != *deleteMonitoredItem->monitoredItemId)) {
336
0
            return NULL;
337
0
        }
338
0
        MonitoredItem_delete(deleteMonitoredItem->client, deleteMonitoredItem->sub, mon);
339
0
    }
340
0
    return NULL;
341
0
}
342
343
static void
344
__Client_Subscription_deleteInternal(UA_Client *client,
345
0
                                     UA_Client_Subscription *sub) {
346
    /* Remove the MonitoredItems */
347
0
    struct UA_Client_MonitoredItem_ForDelete deleteMonitoredItem;
348
0
    memset(&deleteMonitoredItem, 0, sizeof(struct UA_Client_MonitoredItem_ForDelete));
349
0
    deleteMonitoredItem.client = client;
350
0
    deleteMonitoredItem.sub = sub;
351
0
    ZIP_ITER(MonitorItemsTree, &sub->monitoredItems,
352
0
             MonitoredItem_delete_wrapper, &deleteMonitoredItem);
353
354
    /* Call the delete callback */
355
0
    if(sub->deleteCallback) {
356
0
        void *subC = sub->context;
357
0
        UA_UInt32 subId = sub->subscriptionId;
358
0
        sub->deleteCallback(client, subId, subC);
359
0
    }
360
361
    /* Remove */
362
0
    LIST_REMOVE(sub, listEntry);
363
0
    UA_free(sub);
364
0
}
365
366
static void
367
__Client_Subscription_processDelete(UA_Client *client,
368
                                    const UA_DeleteSubscriptionsRequest *request,
369
0
                                    const UA_DeleteSubscriptionsResponse *response)  {
370
0
    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD)
371
0
        return;
372
373
    /* Check that the request and response size -- use the same index for both */
374
0
    if(request->subscriptionIdsSize != response->resultsSize)
375
0
        return;
376
377
0
    for(size_t i = 0; i < request->subscriptionIdsSize; i++) {
378
0
        if(response->results[i] != UA_STATUSCODE_GOOD &&
379
0
           response->results[i] != UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID)
380
0
            continue;
381
382
        /* Get the Subscription */
383
0
        UA_Client_Subscription *sub =
384
0
            findSubscriptionById(client, request->subscriptionIds[i]);
385
0
        if(!sub) {
386
0
            UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
387
0
                        "No internal representation of subscription %" PRIu32,
388
0
                        request->subscriptionIds[i]);
389
0
            continue;
390
0
        }
391
392
        /* Delete the Subscription */
393
0
        __Client_Subscription_deleteInternal(client, sub);
394
0
    }
395
0
}
396
397
typedef struct {
398
    UA_DeleteSubscriptionsRequest request;
399
    UA_ClientAsyncDeleteSubscriptionsCallback userCallback;
400
    void *userData;
401
} DeleteSubscriptionCallback;
402
403
static void
404
Subscriptions_delete_handler(UA_Client *client, void *data,
405
0
                             UA_UInt32 requestId, void *r) {
406
0
    UA_DeleteSubscriptionsResponse *response =
407
0
        (UA_DeleteSubscriptionsResponse *)r;
408
0
    DeleteSubscriptionCallback *dsc =
409
0
        (DeleteSubscriptionCallback*)data;
410
411
0
    lockClient(client);
412
413
    /* Delete */
414
0
    __Client_Subscription_processDelete(client, &dsc->request, response);
415
416
    /* Userland Callback */
417
0
    if(dsc->userCallback)
418
0
        dsc->userCallback(client, dsc->userData, requestId, response);
419
420
    /* Cleanup */
421
0
    UA_DeleteSubscriptionsRequest_clear(&dsc->request);
422
0
    UA_free(dsc);
423
424
0
    unlockClient(client);
425
0
}
426
427
UA_StatusCode
428
UA_Client_Subscriptions_delete_async(UA_Client *client,
429
                                     const UA_DeleteSubscriptionsRequest request,
430
                                     UA_ClientAsyncDeleteSubscriptionsCallback callback,
431
0
                                     void *userdata, UA_UInt32 *requestId) {
432
    /* Make a copy of the request that persists into the async callback */
433
0
    DeleteSubscriptionCallback *dsc = (DeleteSubscriptionCallback*)
434
0
        UA_malloc(sizeof(DeleteSubscriptionCallback));
435
0
    if(!dsc)
436
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
437
0
    dsc->userCallback = callback;
438
0
    dsc->userData = userdata;
439
0
    UA_StatusCode res = UA_DeleteSubscriptionsRequest_copy(&request, &dsc->request);
440
0
    if(res != UA_STATUSCODE_GOOD) {
441
0
        UA_free(dsc);
442
0
        return res;
443
0
    }
444
445
    /* Make the async call */
446
0
    res = __UA_Client_AsyncService(client, &request,
447
0
                                   &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSREQUEST],
448
0
                                   Subscriptions_delete_handler,
449
0
                                   &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSRESPONSE],
450
0
                                   dsc, requestId);
451
0
    if(res != UA_STATUSCODE_GOOD) {
452
0
        UA_DeleteSubscriptionsRequest_clear(&dsc->request);
453
0
        UA_free(dsc);
454
0
    }
455
0
    return res;
456
0
}
457
458
UA_DeleteSubscriptionsResponse
459
UA_Client_Subscriptions_delete(UA_Client *client,
460
0
                               const UA_DeleteSubscriptionsRequest request) {
461
0
    lockClient(client);
462
463
    /* Send the request */
464
0
    UA_DeleteSubscriptionsResponse response;
465
0
    __Client_Service(client, &request,
466
0
                     &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSREQUEST],
467
0
                     &response, &UA_TYPES[UA_TYPES_DELETESUBSCRIPTIONSRESPONSE]);
468
469
    /* Process */
470
0
    __Client_Subscription_processDelete(client, &request, &response);
471
472
0
    unlockClient(client);
473
0
    return response;
474
0
}
475
476
UA_StatusCode
477
0
UA_Client_Subscriptions_deleteSingle(UA_Client *client, UA_UInt32 subscriptionId) {
478
0
    UA_DeleteSubscriptionsRequest request;
479
0
    UA_DeleteSubscriptionsRequest_init(&request);
480
0
    request.subscriptionIds = &subscriptionId;
481
0
    request.subscriptionIdsSize = 1;
482
483
0
    UA_DeleteSubscriptionsResponse response =
484
0
        UA_Client_Subscriptions_delete(client, request);
485
486
0
    UA_StatusCode retval = response.responseHeader.serviceResult;
487
0
    if(retval != UA_STATUSCODE_GOOD) {
488
0
        UA_DeleteSubscriptionsResponse_clear(&response);
489
0
        return retval;
490
0
    }
491
492
0
    if(response.resultsSize != 1) {
493
0
        UA_DeleteSubscriptionsResponse_clear(&response);
494
0
        return UA_STATUSCODE_BADINTERNALERROR;
495
0
    }
496
497
0
    retval = response.results[0];
498
0
    UA_DeleteSubscriptionsResponse_clear(&response);
499
0
    return retval;
500
0
}
501
502
/******************/
503
/* MonitoredItems */
504
/******************/
505
506
static void
507
0
EventFields_clear(UA_KeyValueMap *eventFields) {
508
    /* The values are borrowed from the PublishResponse. Detach them before
509
     * clearing the map-owned field-name keys. */
510
0
    for(size_t i = 0; i < eventFields->mapSize; i++)
511
0
        UA_Variant_init(&eventFields->map[i].value);
512
0
    UA_KeyValueMap_clear(eventFields);
513
0
}
514
515
static void
516
MonitoredItem_delete(UA_Client *client, UA_Client_Subscription *sub,
517
0
                     UA_Client_MonitoredItem *mon) {
518
0
    UA_LOCK_ASSERT(&client->clientMutex);
519
520
0
    ZIP_REMOVE(MonitorItemsTree, &sub->monitoredItems, mon);
521
0
    if(mon->deleteCallback)
522
0
        mon->deleteCallback(client, sub->subscriptionId, sub->context,
523
0
                            mon->monitoredItemId, mon->context);
524
0
    EventFields_clear(&mon->eventFields);
525
0
    UA_MonitoringParameters_clear(&mon->parameters);
526
0
    UA_MonitoringParameters_clear(&mon->pendingParameters);
527
0
    UA_free(mon);
528
0
}
529
530
static UA_StatusCode
531
prepareEventFieldsMap(UA_KeyValueMap *eventFields,
532
0
                      UA_MonitoringParameters *params) {
533
    /* Get the EventFilter */
534
0
    UA_ExtensionObject *eo = &params->filter;
535
0
    if(eo->content.decoded.type != &UA_TYPES[UA_TYPES_EVENTFILTER])
536
0
        return UA_STATUSCODE_GOOD;
537
0
    UA_EventFilter *ef = (UA_EventFilter*)eo->content.decoded.data;
538
0
    UA_StatusCode res = UA_STATUSCODE_GOOD;
539
540
    /* Check whether there are fields */
541
0
    if(ef->selectClausesSize == 0)
542
0
        return UA_STATUSCODE_GOOD;
543
544
    /* Allocate the map */
545
0
    eventFields->map = (UA_KeyValuePair*)
546
0
        UA_calloc(ef->selectClausesSize, sizeof(UA_KeyValuePair));
547
0
    if(!eventFields->map)
548
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
549
0
    eventFields->mapSize = ef->selectClausesSize;
550
551
    /* Create the key-strings for the fields */
552
0
    for(size_t i = 0; i < eventFields->mapSize; i++) {
553
0
        res |= UA_SimpleAttributeOperand_print(&ef->selectClauses[i],
554
0
                                               &eventFields->map[i].key.name);
555
0
    }
556
557
0
    return res;
558
0
}
559
560
static UA_StatusCode
561
MonitoredItem_createBegin(UA_Client *client, UA_Client_Subscription *sub,
562
                          UA_MonitoredItemCreateRequest *item,
563
                          UA_Client_DeleteMonitoredItemCallback deleteCallback,
564
                          void *context, void *handlingCallback,
565
0
                          UA_Client_MonitoredItem **outMon) {
566
    /* Allocate MonitoredItem */
567
0
    UA_Client_MonitoredItem *mon = (UA_Client_MonitoredItem *)
568
0
        UA_calloc(1, sizeof(UA_Client_MonitoredItem));
569
0
    if(!mon)
570
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
571
572
    /* Set a unique ClientHandle and retain the active parameters. */
573
0
    item->requestedParameters.clientHandle = ++client->monitoredItemHandles;
574
0
    UA_StatusCode res =
575
0
        UA_MonitoringParameters_copy(&item->requestedParameters,
576
0
                                     &mon->parameters);
577
0
    if(res != UA_STATUSCODE_GOOD) {
578
0
        UA_free(mon);
579
0
        return res;
580
0
    }
581
582
0
    mon->isEventMonitoredItem =
583
0
        (item->itemToMonitor.attributeId == UA_ATTRIBUTEID_EVENTNOTIFIER);
584
585
    /* Fill in members and add to the client  */
586
0
    mon->context = context;
587
0
    mon->deleteCallback = deleteCallback;
588
0
    mon->handler.dataChangeCallback =
589
0
        (UA_Client_DataChangeNotificationCallback)(uintptr_t)handlingCallback;
590
0
    ZIP_INSERT(MonitorItemsTree, &sub->monitoredItems, mon);
591
592
0
    UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
593
0
                 "Subscription %" PRIu32 " | Added a MonitoredItem with handle %" PRIu32,
594
0
                 sub->subscriptionId, mon->parameters.clientHandle);
595
596
0
    *outMon = mon;
597
0
    return UA_STATUSCODE_GOOD;
598
0
}
599
600
static void
601
MonitoredItem_createFinish(UA_Client *client, UA_Client_Subscription *sub,
602
                           UA_Client_MonitoredItem *mon,
603
0
                           UA_MonitoredItemCreateResult *result) {
604
0
    UA_assert(result->statusCode == UA_STATUSCODE_GOOD);
605
606
0
    mon->monitoredItemId = result->monitoredItemId;
607
    /* revisedSamplingInterval; */
608
    /* revisedQueueSize; */
609
    /* filterResult; */
610
0
}
611
612
/************************************/
613
/* CreateMonitoredItems Synchronous */
614
/************************************/
615
616
static void
617
Client_MonitoredItems_create(UA_Client *client,
618
                             const UA_CreateMonitoredItemsRequest *constRequest,
619
                             void **contexts, void **handlingCallbacks,
620
                             UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
621
0
                             UA_CreateMonitoredItemsResponse *response) {
622
0
    UA_LOCK_ASSERT(&client->clientMutex);
623
0
    UA_CreateMonitoredItemsResponse_init(response);
624
625
    /* Any items? */
626
0
    if(constRequest->itemsToCreateSize == 0) {
627
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADNOTHINGTODO;
628
0
        return;
629
0
    }
630
631
    /* Get the subscription */
632
0
    UA_Client_Subscription *sub =
633
0
        findSubscriptionById(client, constRequest->subscriptionId);
634
0
    if(!sub) {
635
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
636
0
        return;
637
0
    }
638
639
    /* Make a mutable copy. We modify the request to set the internal
640
     * clientHandle. */
641
0
    UA_CreateMonitoredItemsRequest request;
642
0
    UA_StatusCode res =
643
0
        UA_CreateMonitoredItemsRequest_copy(constRequest, &request);
644
0
    if(res != UA_STATUSCODE_GOOD) {
645
0
        response->responseHeader.serviceResult = res;
646
0
        return;
647
0
    }
648
649
    /* Create the MonitoredItems */
650
0
    UA_STACKARRAY(UA_Client_MonitoredItem*, mons, request.itemsToCreateSize);
651
0
    memset(mons, 0, sizeof(UA_Client_MonitoredItem*) * request.itemsToCreateSize);
652
0
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
653
0
        void *context = (contexts) ? contexts[i] : NULL;
654
0
        void *handlingCallback = (handlingCallbacks) ? handlingCallbacks[i] : NULL;
655
0
        UA_Client_DeleteMonitoredItemCallback deleteCallback =
656
0
            (deleteCallbacks) ? deleteCallbacks[i] : NULL;
657
0
        res |= MonitoredItem_createBegin(client, sub, &request.itemsToCreate[i],
658
0
                                         deleteCallback, context,
659
0
                                         handlingCallback, &mons[i]);
660
0
    }
661
662
    /* Failure -> Delete created MonitoredItems. Directly call deleteCallback if
663
     * creation failed. The MonitoredItemId is not yet known, use zero. */
664
0
    if(res != UA_STATUSCODE_GOOD) {
665
0
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
666
0
            if(mons[i])
667
0
                MonitoredItem_delete(client, sub, mons[i]);
668
0
            else if(deleteCallbacks && contexts)
669
0
                deleteCallbacks[i](client, request.subscriptionId,
670
0
                                   sub->context, 0, contexts[i]);
671
0
        }
672
0
        UA_CreateMonitoredItemsRequest_clear(&request);
673
0
        response->responseHeader.serviceResult = res;
674
0
        return;
675
0
    }
676
677
    /* Call the service */
678
0
    __Client_Service(client, &request,
679
0
                     &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSREQUEST],
680
0
                     response, &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSRESPONSE]);
681
682
    /* Check that the response size is good */
683
0
    if(response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
684
0
       response->resultsSize != request.itemsToCreateSize)
685
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADINTERNALERROR;
686
687
    /* Update the MonitoredItems */
688
0
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
689
0
        UA_assert(mons[i]);
690
0
        UA_MonitoredItemCreateResult *item = &response->results[i];
691
0
        if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD ||
692
0
           item->statusCode != UA_STATUSCODE_GOOD) {
693
0
            MonitoredItem_delete(client, sub, mons[i]);
694
0
            continue;
695
0
        }
696
0
        MonitoredItem_createFinish(client, sub, mons[i], item);
697
0
    }
698
699
0
    UA_CreateMonitoredItemsRequest_clear(&request);
700
0
}
701
702
UA_CreateMonitoredItemsResponse
703
UA_Client_MonitoredItems_createDataChanges(UA_Client *client,
704
                                           const UA_CreateMonitoredItemsRequest request,
705
                                           void **contexts,
706
                                           UA_Client_DataChangeNotificationCallback *callbacks,
707
0
                                           UA_Client_DeleteMonitoredItemCallback *deleteCallbacks) {
708
0
    UA_CreateMonitoredItemsResponse response;
709
0
    lockClient(client);
710
0
    Client_MonitoredItems_create(client, &request, contexts, (void **)callbacks,
711
0
                                 deleteCallbacks, &response);
712
0
    unlockClient(client);
713
0
    return response;
714
0
}
715
716
UA_MonitoredItemCreateResult
717
UA_Client_MonitoredItems_createDataChange(UA_Client *client, UA_UInt32 subscriptionId,
718
                                          UA_TimestampsToReturn timestampsToReturn,
719
                                          const UA_MonitoredItemCreateRequest item,
720
                                          void *context,
721
                                          UA_Client_DataChangeNotificationCallback callback,
722
0
                                          UA_Client_DeleteMonitoredItemCallback deleteCallback) {
723
0
    UA_CreateMonitoredItemsRequest request;
724
0
    UA_CreateMonitoredItemsRequest_init(&request);
725
0
    request.subscriptionId = subscriptionId;
726
0
    request.timestampsToReturn = timestampsToReturn;
727
0
    request.itemsToCreate = (UA_MonitoredItemCreateRequest*)(uintptr_t)&item;
728
0
    request.itemsToCreateSize = 1;
729
0
    UA_CreateMonitoredItemsResponse response =
730
0
       UA_Client_MonitoredItems_createDataChanges(client, request, &context,
731
0
                                                   &callback, &deleteCallback);
732
0
    UA_MonitoredItemCreateResult result;
733
0
    UA_MonitoredItemCreateResult_init(&result);
734
0
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
735
0
        result.statusCode = response.responseHeader.serviceResult;
736
0
    if(result.statusCode == UA_STATUSCODE_GOOD &&
737
0
       response.resultsSize != 1)
738
0
        result.statusCode = UA_STATUSCODE_BADINTERNALERROR;
739
0
    if(result.statusCode == UA_STATUSCODE_GOOD) {
740
0
        result = response.results[0];
741
0
        UA_MonitoredItemCreateResult_init(&response.results[0]);
742
0
    }
743
0
    UA_CreateMonitoredItemsResponse_clear(&response);
744
0
    return result;
745
0
}
746
747
UA_CreateMonitoredItemsResponse
748
UA_Client_MonitoredItems_createEvents(UA_Client *client,
749
                                      const UA_CreateMonitoredItemsRequest request,
750
                                      void **contexts,
751
                                      UA_Client_EventNotificationCallback *callback,
752
0
                                      UA_Client_DeleteMonitoredItemCallback *deleteCallback) {
753
0
    UA_CreateMonitoredItemsResponse response;
754
0
    lockClient(client);
755
0
    Client_MonitoredItems_create(client, &request, contexts, (void **)callback,
756
0
                                 deleteCallback, &response);
757
0
    unlockClient(client);
758
0
    return response;
759
0
}
760
761
UA_MonitoredItemCreateResult
762
UA_Client_MonitoredItems_createEvent(UA_Client *client, UA_UInt32 subscriptionId,
763
                                     UA_TimestampsToReturn timestampsToReturn,
764
                                     const UA_MonitoredItemCreateRequest item, void *context,
765
                                     UA_Client_EventNotificationCallback callback,
766
0
                                     UA_Client_DeleteMonitoredItemCallback deleteCallback) {
767
0
    UA_CreateMonitoredItemsRequest request;
768
0
    UA_CreateMonitoredItemsRequest_init(&request);
769
0
    request.subscriptionId = subscriptionId;
770
0
    request.timestampsToReturn = timestampsToReturn;
771
0
    request.itemsToCreate = (UA_MonitoredItemCreateRequest*)(uintptr_t)&item;
772
0
    request.itemsToCreateSize = 1;
773
0
    UA_CreateMonitoredItemsResponse response =
774
0
       UA_Client_MonitoredItems_createEvents(client, request, &context,
775
0
                                             &callback, &deleteCallback);
776
0
    UA_MonitoredItemCreateResult result;
777
0
    UA_MonitoredItemCreateResult_init(&result);
778
0
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
779
0
        result.statusCode = response.responseHeader.serviceResult;
780
0
    if(result.statusCode == UA_STATUSCODE_GOOD &&
781
0
       response.resultsSize != 1)
782
0
        result.statusCode = UA_STATUSCODE_BADINTERNALERROR;
783
0
    if(result.statusCode == UA_STATUSCODE_GOOD) {
784
0
        result = response.results[0];
785
0
        UA_MonitoredItemCreateResult_init(&response.results[0]);
786
0
    }
787
0
    UA_CreateMonitoredItemsResponse_clear(&response);
788
0
    return result;
789
0
}
790
791
/*************************************/
792
/* CreateMonitoredItems Asynchronous */
793
/*************************************/
794
795
/* The handles are an array of [subId, monSize, monHandleId1, monHandle2, ...] */
796
static void
797
MonitoredItems_create_async_handler(UA_Client *client, void *data,
798
0
                                    UA_UInt32 requestId, void *resp) {
799
0
    CustomCallback *cc = (CustomCallback*)data;
800
0
    UA_UInt32 *handles = (UA_UInt32*)cc->clientData;
801
0
    UA_CreateMonitoredItemsResponse *response =
802
0
        (UA_CreateMonitoredItemsResponse *)resp;
803
804
0
    lockClient(client);
805
806
    /* Extract the first elements from the handles */
807
0
    UA_UInt32 subId = handles[0];
808
0
    UA_UInt32 monSize = handles[1];
809
0
    UA_UInt32 *monHandles = handles + 2;
810
811
    /* Check that the response size is good */
812
0
    if(response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
813
0
       response->resultsSize != monSize)
814
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADINTERNALERROR;
815
816
    /* Get the Subscription from the SubscriptionId */
817
0
    UA_Client_Subscription *sub = findSubscriptionById(client, subId);
818
0
    if(!sub)
819
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
820
821
    /* Update the MonitoredItems */
822
0
    for(size_t i = 0; sub && i < monSize; i++) {
823
        /* Get the MonitoredItem from the ClientHandle */
824
0
        UA_Client_MonitoredItem *mon =
825
0
            findMonitoredItemByHandle(sub, monHandles[i]);
826
0
        if(!mon)
827
0
            continue;
828
829
        /* Delete MonitoredItem if the creation failed */
830
0
        UA_MonitoredItemCreateResult *item = &response->results[i];
831
0
        if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD ||
832
0
           item->statusCode != UA_STATUSCODE_GOOD) {
833
0
            MonitoredItem_delete(client, sub, mon);
834
0
            continue;
835
0
        }
836
837
        /* Update the MonitoredItem with the server's response  */
838
0
        MonitoredItem_createFinish(client, sub, mon, item);
839
0
    }
840
841
    /* Notify the application */
842
0
    if(cc->callback.createMonitoredItems)
843
0
        cc->callback.createMonitoredItems(client, cc->userData, requestId, response);
844
845
    /* Clean up */
846
0
    UA_free(handles);
847
0
    UA_free(cc);
848
849
0
    unlockClient(client);
850
0
}
851
852
static UA_StatusCode
853
Client_MonitoredItems_createAsync(UA_Client *client,
854
                                  const UA_CreateMonitoredItemsRequest *constRequest,
855
                                  void **contexts, void **handlingCallbacks,
856
                                  UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
857
                                  UA_ClientAsyncCreateMonitoredItemsCallback createCallback,
858
0
                                  void *userdata, UA_UInt32 *requestId) {
859
0
    UA_LOCK_ASSERT(&client->clientMutex);
860
861
0
    UA_StatusCode res = __Client_AsyncServiceAdmission(client);
862
0
    if(res != UA_STATUSCODE_GOOD)
863
0
        return res;
864
865
    /* Get the Subscription */
866
0
    UA_Client_Subscription *sub =
867
0
        findSubscriptionById(client, constRequest->subscriptionId);
868
0
    if(!sub)
869
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
870
871
    /* Any items? */
872
0
    if(constRequest->itemsToCreateSize == 0)
873
0
        return UA_STATUSCODE_BADNOTHINGTODO;
874
875
    /* Make a mutable copy. We modify the request to set the internal
876
     * clientHandle. */
877
0
    UA_CreateMonitoredItemsRequest request;
878
0
    res = UA_CreateMonitoredItemsRequest_copy(constRequest, &request);
879
0
    if(res != UA_STATUSCODE_GOOD)
880
0
        return res;
881
882
    /* Allocate context for the async handling */
883
0
    CustomCallback *cc = (CustomCallback*)UA_calloc(1, sizeof(CustomCallback));
884
0
    if(!cc) {
885
0
        UA_CreateMonitoredItemsRequest_clear(&request);
886
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
887
0
    }
888
0
    UA_UInt32 *handles = (UA_UInt32*)
889
0
        UA_malloc(sizeof(UA_UInt32) * (request.itemsToCreateSize + 2));
890
0
    if(!handles) {
891
0
        UA_free(cc);
892
0
        UA_CreateMonitoredItemsRequest_clear(&request);
893
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
894
0
    }
895
896
    /* Create the MonitoredItems locally */
897
0
    UA_STACKARRAY(UA_Client_MonitoredItem*, mons, request.itemsToCreateSize);
898
0
    memset(mons, 0, sizeof(UA_Client_MonitoredItem*) * request.itemsToCreateSize);
899
0
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
900
0
        void *context = (contexts) ? contexts[i] : NULL;
901
0
        void *handlingCallback = (handlingCallbacks) ? handlingCallbacks[i] : NULL;
902
0
        UA_Client_DeleteMonitoredItemCallback deleteCallback =
903
0
            (deleteCallbacks) ? deleteCallbacks[i] : NULL;
904
0
        res |= MonitoredItem_createBegin(client, sub, &request.itemsToCreate[i],
905
0
                                         deleteCallback, context,
906
0
                                         handlingCallback, &mons[i]);
907
0
    }
908
909
    /* Failure -> Delete created MonitoredItems. Directly call deleteCallback if
910
     * creation failed. The MonitoredItemId is not yet known, use zero. */
911
0
    if(res != UA_STATUSCODE_GOOD) {
912
0
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
913
0
            if(mons[i])
914
0
                MonitoredItem_delete(client, sub, mons[i]);
915
0
            else if(deleteCallbacks && contexts)
916
0
                deleteCallbacks[i](client, request.subscriptionId,
917
0
                                   sub->context, 0, contexts[i]);
918
0
        }
919
0
        UA_free(handles);
920
0
        UA_free(cc);
921
0
        UA_CreateMonitoredItemsRequest_clear(&request);
922
0
        return res;
923
0
    }
924
925
    /* Set the async handler context */
926
0
    handles[0] = sub->subscriptionId;
927
0
    handles[1] = (UA_UInt32)request.itemsToCreateSize;
928
0
    for(size_t i = 0; i < request.itemsToCreateSize; i++) {
929
0
        handles[i+2] = mons[i]->parameters.clientHandle;
930
0
    }
931
0
    cc->clientData = handles;
932
0
    cc->callback.createMonitoredItems = createCallback;
933
0
    cc->userData = userdata;
934
935
    /* Call the service asynchronously */
936
0
    res = __Client_AsyncServiceAdmitted(
937
0
        client, &request, &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSREQUEST],
938
0
        MonitoredItems_create_async_handler,
939
0
        &UA_TYPES[UA_TYPES_CREATEMONITOREDITEMSRESPONSE], cc, requestId);
940
941
    /* Manually clean up the context in the failure case */
942
0
    if(res != UA_STATUSCODE_GOOD) {
943
0
        for(size_t i = 0; i < request.itemsToCreateSize; i++) {
944
0
            UA_assert(mons[i]);
945
0
            MonitoredItem_delete(client, sub, mons[i]);
946
0
        }
947
0
        UA_free(handles);
948
0
        UA_free(cc);
949
0
    }
950
951
    /* Clean up */
952
0
    UA_CreateMonitoredItemsRequest_clear(&request);
953
0
    return res;
954
0
}
955
956
UA_StatusCode
957
UA_Client_MonitoredItems_createDataChanges_async(UA_Client *client,
958
                                                 const UA_CreateMonitoredItemsRequest request,
959
                                                 void **contexts,
960
                                                 UA_Client_DataChangeNotificationCallback *callbacks,
961
                                                 UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
962
                                                 UA_ClientAsyncCreateMonitoredItemsCallback createCallback,
963
0
                                                 void *userdata, UA_UInt32 *requestId) {
964
0
    lockClient(client);
965
0
    UA_StatusCode res =
966
0
        Client_MonitoredItems_createAsync(client, &request, contexts,
967
0
                                          (void **)callbacks, deleteCallbacks,
968
0
                                          createCallback,
969
0
                                          userdata, requestId);
970
0
    unlockClient(client);
971
0
    return res;
972
0
}
973
974
UA_StatusCode
975
UA_Client_MonitoredItems_createEvents_async(UA_Client *client,
976
                                            const UA_CreateMonitoredItemsRequest request,
977
                                            void **contexts,
978
                                            UA_Client_EventNotificationCallback *callbacks,
979
                                            UA_Client_DeleteMonitoredItemCallback *deleteCallbacks,
980
                                            UA_ClientAsyncCreateMonitoredItemsCallback createCallback,
981
0
                                            void *userdata, UA_UInt32 *requestId) {
982
0
    lockClient(client);
983
0
    UA_StatusCode res =
984
0
        Client_MonitoredItems_createAsync(client, &request, contexts,
985
0
                                          (void **)callbacks, deleteCallbacks,
986
0
                                          createCallback,
987
0
                                          userdata, requestId);
988
0
    unlockClient(client);
989
0
    return res;
990
0
}
991
992
static void
993
MonitoredItems_delete(UA_Client *client, UA_Client_Subscription *sub,
994
                      const UA_DeleteMonitoredItemsRequest *request,
995
0
                      const UA_DeleteMonitoredItemsResponse *response) {
996
#ifdef __clang_analyzer__
997
    return;
998
#endif
999
1000
    /* Loop over deleted MonitoredItems */
1001
0
    struct UA_Client_MonitoredItem_ForDelete deleteMonitoredItem;
1002
0
    memset(&deleteMonitoredItem, 0, sizeof(struct UA_Client_MonitoredItem_ForDelete));
1003
0
    deleteMonitoredItem.client = client;
1004
0
    deleteMonitoredItem.sub = sub;
1005
1006
0
    for(size_t i = 0; i < response->resultsSize; i++) {
1007
0
        if(response->results[i] != UA_STATUSCODE_GOOD &&
1008
0
           response->results[i] != UA_STATUSCODE_BADMONITOREDITEMIDINVALID) {
1009
0
            continue;
1010
0
        }
1011
0
        deleteMonitoredItem.monitoredItemId = &request->monitoredItemIds[i];
1012
        /* Delete the internal representation */
1013
0
        ZIP_ITER(MonitorItemsTree,&sub->monitoredItems,
1014
0
                 MonitoredItem_delete_wrapper, &deleteMonitoredItem);
1015
0
    }
1016
0
}
1017
1018
static void
1019
0
MonitoredItems_delete_handler(UA_Client *client, void *d, UA_UInt32 requestId, void *r) {
1020
0
    UA_Client_Subscription *sub = NULL;
1021
0
    CustomCallback *cc = (CustomCallback *)d;
1022
0
    UA_DeleteMonitoredItemsResponse *response = (UA_DeleteMonitoredItemsResponse *)r;
1023
0
    UA_DeleteMonitoredItemsRequest *request =
1024
0
        (UA_DeleteMonitoredItemsRequest *)cc->clientData;
1025
1026
0
    lockClient(client);
1027
1028
0
    if(response->responseHeader.serviceResult != UA_STATUSCODE_GOOD)
1029
0
        goto cleanup;
1030
1031
0
    sub = findSubscriptionById(client, request->subscriptionId);
1032
0
    if(!sub) {
1033
0
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
1034
0
                    "No internal representation of subscription %" PRIu32,
1035
0
                    request->subscriptionId);
1036
0
        goto cleanup;
1037
0
    }
1038
1039
    /* Delete MonitoredItems from the internal representation */
1040
0
    MonitoredItems_delete(client, sub, request, response);
1041
1042
0
cleanup:
1043
0
    if(cc->callback.deleteMonitoredItems)
1044
0
        cc->callback.deleteMonitoredItems(client, cc->userData, requestId, response);
1045
0
    UA_DeleteMonitoredItemsRequest_delete(request);
1046
0
    UA_free(cc);
1047
1048
0
    unlockClient(client);
1049
0
}
1050
1051
UA_DeleteMonitoredItemsResponse
1052
UA_Client_MonitoredItems_delete(UA_Client *client,
1053
0
                                const UA_DeleteMonitoredItemsRequest request) {
1054
    /* Send the request */
1055
0
    UA_DeleteMonitoredItemsResponse response;
1056
0
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSREQUEST],
1057
0
                        &response, &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSRESPONSE]);
1058
1059
    /* A problem occured remote? */
1060
0
    if(response.responseHeader.serviceResult != UA_STATUSCODE_GOOD)
1061
0
        return response;
1062
1063
0
    lockClient(client);
1064
1065
    /* Find the internal subscription representation */
1066
0
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
1067
0
    if(!sub) {
1068
0
        UA_LOG_INFO(client->config.logging, UA_LOGCATEGORY_CLIENT,
1069
0
                    "No internal representation of subscription %" PRIu32,
1070
0
                    request.subscriptionId);
1071
0
        unlockClient(client);
1072
0
        return response;
1073
0
    }
1074
1075
    /* Remove MonitoredItems in the internal representation */
1076
0
    MonitoredItems_delete(client, sub, &request, &response);
1077
1078
0
    unlockClient(client);
1079
1080
0
    return response;
1081
0
}
1082
1083
UA_StatusCode
1084
UA_Client_MonitoredItems_delete_async(UA_Client *client,
1085
                                      const UA_DeleteMonitoredItemsRequest request,
1086
                                      UA_ClientAsyncDeleteMonitoredItemsCallback callback,
1087
0
                                      void *userdata, UA_UInt32 *requestId) {
1088
    /* Send the request */
1089
0
    CustomCallback *cc = (CustomCallback *)UA_calloc(1, sizeof(CustomCallback));
1090
0
    if(!cc)
1091
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
1092
1093
0
    UA_DeleteMonitoredItemsRequest *req_copy = UA_DeleteMonitoredItemsRequest_new();
1094
0
    if(!req_copy) {
1095
0
        UA_free(cc);
1096
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
1097
0
    }
1098
1099
0
    UA_DeleteMonitoredItemsRequest_copy(&request, req_copy);
1100
0
    cc->clientData = req_copy;
1101
0
    cc->callback.deleteMonitoredItems = callback;
1102
0
    cc->userData = userdata;
1103
1104
0
    UA_StatusCode res =
1105
0
        __UA_Client_AsyncService(client, &request,
1106
0
                                 &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSREQUEST],
1107
0
                                 MonitoredItems_delete_handler,
1108
0
                                 &UA_TYPES[UA_TYPES_DELETEMONITOREDITEMSRESPONSE],
1109
0
                                 cc, requestId);
1110
0
    if(res != UA_STATUSCODE_GOOD) {
1111
0
        UA_DeleteMonitoredItemsRequest_delete(req_copy);
1112
0
        UA_free(cc);
1113
0
    }
1114
0
    return res;
1115
0
}
1116
1117
UA_StatusCode
1118
UA_Client_MonitoredItems_deleteSingle(UA_Client *client, UA_UInt32 subscriptionId,
1119
0
                                      UA_UInt32 monitoredItemId) {
1120
0
    UA_DeleteMonitoredItemsRequest request;
1121
0
    UA_DeleteMonitoredItemsRequest_init(&request);
1122
0
    request.subscriptionId = subscriptionId;
1123
0
    request.monitoredItemIds = &monitoredItemId;
1124
0
    request.monitoredItemIdsSize = 1;
1125
1126
0
    UA_DeleteMonitoredItemsResponse response =
1127
0
        UA_Client_MonitoredItems_delete(client, request);
1128
1129
0
    UA_StatusCode retval = response.responseHeader.serviceResult;
1130
0
    if(retval != UA_STATUSCODE_GOOD) {
1131
0
        UA_DeleteMonitoredItemsResponse_clear(&response);
1132
0
        return retval;
1133
0
    }
1134
1135
0
    if(response.resultsSize != 1) {
1136
0
        UA_DeleteMonitoredItemsResponse_clear(&response);
1137
0
        return UA_STATUSCODE_BADINTERNALERROR;
1138
0
    }
1139
1140
0
    retval = response.results[0];
1141
0
    UA_DeleteMonitoredItemsResponse_clear(&response);
1142
0
    return retval;
1143
0
}
1144
1145
static void *
1146
0
MonitoredItem_findByID(void *data, UA_Client_MonitoredItem *mon) {
1147
0
    UA_UInt32 monitorId = *(UA_UInt32*)data;
1148
0
    if(monitorId && (mon->monitoredItemId == monitorId))
1149
0
        return mon;
1150
0
    return NULL;
1151
0
}
1152
1153
static UA_Client_MonitoredItem *
1154
0
findMonitoredItemById(UA_Client_Subscription *sub, UA_UInt32 monitoredItemId) {
1155
0
    return (UA_Client_MonitoredItem *)
1156
0
        ZIP_ITER(MonitorItemsTree, &sub->monitoredItems,
1157
0
                 MonitoredItem_findByID, &monitoredItemId);
1158
0
}
1159
1160
static UA_StatusCode
1161
MonitoredItems_prepareModify(UA_Client *client, UA_Client_Subscription *sub,
1162
0
                             UA_ModifyMonitoredItemsRequest *request) {
1163
0
    UA_STACKARRAY(UA_Client_MonitoredItem*, mons, request->itemsToModifySize);
1164
0
    UA_STACKARRAY(UA_MonitoringParameters, preparedParameters,
1165
0
                  request->itemsToModifySize);
1166
0
    memset(mons, 0, sizeof(UA_Client_MonitoredItem*) *
1167
0
           request->itemsToModifySize);
1168
0
    memset(preparedParameters, 0, sizeof(UA_MonitoringParameters) *
1169
0
           request->itemsToModifySize);
1170
1171
    /* Prepare every settings slot before changing the MonitoredItems. So an
1172
     * allocation failure leaves all active and pending settings untouched. */
1173
0
    for(size_t i = 0; i < request->itemsToModifySize; ++i) {
1174
0
        UA_MonitoredItemModifyRequest *mimr = &request->itemsToModify[i];
1175
0
        mimr->requestedParameters.clientHandle = 0;
1176
0
        mons[i] = findMonitoredItemById(sub, mimr->monitoredItemId);
1177
0
        if(!mons[i])
1178
0
            continue;
1179
1180
0
        mimr->requestedParameters.clientHandle = ++client->monitoredItemHandles;
1181
0
        UA_StatusCode res =
1182
0
            UA_MonitoringParameters_copy(&mimr->requestedParameters,
1183
0
                                         &preparedParameters[i]);
1184
0
        if(res != UA_STATUSCODE_GOOD) {
1185
0
            for(size_t j = 0; j <= i; j++)
1186
0
                UA_MonitoringParameters_clear(&preparedParameters[j]);
1187
0
            return res;
1188
0
        }
1189
0
    }
1190
1191
    /* Commit the prepared settings. A newer modification supersedes a pending
1192
     * generation that has not produced a notification yet. */
1193
0
    for(size_t i = 0; i < request->itemsToModifySize; ++i) {
1194
0
        UA_Client_MonitoredItem *mon = mons[i];
1195
0
        if(!mon)
1196
0
            continue;
1197
0
        UA_MonitoringParameters_clear(&mon->pendingParameters);
1198
0
        mon->pendingParameters = preparedParameters[i];
1199
0
        memset(&preparedParameters[i], 0, sizeof(UA_MonitoringParameters));
1200
0
    }
1201
0
    return UA_STATUSCODE_GOOD;
1202
0
}
1203
1204
static void
1205
MonitoredItems_reconcileModify(UA_Client_Subscription *sub,
1206
                               const UA_ModifyMonitoredItemsRequest *request,
1207
0
                               UA_ModifyMonitoredItemsResponse *response) {
1208
0
    UA_Boolean validResponse = response &&
1209
0
        response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
1210
0
        response->resultsSize == request->itemsToModifySize;
1211
0
    if(response && response->responseHeader.serviceResult == UA_STATUSCODE_GOOD &&
1212
0
       !validResponse)
1213
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADINTERNALERROR;
1214
1215
0
    for(size_t i = 0; i < request->itemsToModifySize; ++i) {
1216
0
        if(validResponse && response->results[i].statusCode == UA_STATUSCODE_GOOD)
1217
0
            continue;
1218
0
        const UA_MonitoredItemModifyRequest *mimr = &request->itemsToModify[i];
1219
0
        UA_Client_MonitoredItem *mon =
1220
0
            findMonitoredItemById(sub, mimr->monitoredItemId);
1221
0
        if(mon && mon->pendingParameters.clientHandle ==
1222
0
                  mimr->requestedParameters.clientHandle)
1223
0
            UA_MonitoringParameters_clear(&mon->pendingParameters);
1224
0
    }
1225
0
}
1226
1227
UA_ModifyMonitoredItemsResponse
1228
UA_Client_MonitoredItems_modify(UA_Client *client,
1229
0
                                const UA_ModifyMonitoredItemsRequest request) {
1230
0
    UA_ModifyMonitoredItemsResponse response;
1231
0
    UA_ModifyMonitoredItemsResponse_init(&response);
1232
1233
    /* Make a modifiable copy of the request */
1234
0
    UA_ModifyMonitoredItemsRequest modifiedRequest;
1235
0
    UA_StatusCode res = UA_ModifyMonitoredItemsRequest_copy(&request, &modifiedRequest);
1236
0
    if(res != UA_STATUSCODE_GOOD) {
1237
0
        response.responseHeader.serviceResult = res;
1238
0
        return response;
1239
0
    }
1240
1241
0
    lockClient(client);
1242
1243
    /* Get the subscription */
1244
0
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
1245
0
    if(!sub) {
1246
0
        unlockClient(client);
1247
0
        UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
1248
0
        response.responseHeader.serviceResult = UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
1249
0
        return response;
1250
0
    }
1251
1252
0
    res = MonitoredItems_prepareModify(client, sub, &modifiedRequest);
1253
0
    if(res != UA_STATUSCODE_GOOD) {
1254
0
        response.responseHeader.serviceResult = res;
1255
0
        unlockClient(client);
1256
0
        UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
1257
0
        return response;
1258
0
    }
1259
1260
    /* Call the service */
1261
0
    __Client_Service(client, &modifiedRequest,
1262
0
                     &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSREQUEST], &response,
1263
0
                     &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSRESPONSE]);
1264
1265
0
    MonitoredItems_reconcileModify(sub, &modifiedRequest, &response);
1266
1267
0
    unlockClient(client);
1268
1269
    /* Clean up */
1270
0
    UA_ModifyMonitoredItemsRequest_clear(&modifiedRequest);
1271
0
    return response;
1272
0
}
1273
1274
static void
1275
MonitoredItems_modify_async_handler(UA_Client *client, void *data,
1276
0
                                    UA_UInt32 requestId, void *resp) {
1277
0
    CustomCallback *cc = (CustomCallback*)data;
1278
0
    UA_ModifyMonitoredItemsRequest *request =
1279
0
        (UA_ModifyMonitoredItemsRequest*)cc->clientData;
1280
0
    UA_ModifyMonitoredItemsResponse *response =
1281
0
        (UA_ModifyMonitoredItemsResponse*)resp;
1282
1283
0
    lockClient(client);
1284
0
    UA_Client_Subscription *sub =
1285
0
        findSubscriptionById(client, request->subscriptionId);
1286
0
    if(sub)
1287
0
        MonitoredItems_reconcileModify(sub, request, response);
1288
1289
0
    if(cc->callback.modifyMonitoredItems)
1290
0
        cc->callback.modifyMonitoredItems(client, cc->userData,
1291
0
                                          requestId, response);
1292
1293
0
    UA_ModifyMonitoredItemsRequest_delete(request);
1294
0
    UA_free(cc);
1295
0
    unlockClient(client);
1296
0
}
1297
1298
UA_StatusCode
1299
UA_Client_MonitoredItems_modify_async(UA_Client *client,
1300
                                      const UA_ModifyMonitoredItemsRequest request,
1301
                                      UA_ClientAsyncModifyMonitoredItemsCallback callback,
1302
0
                                      void *userdata, UA_UInt32 *requestId) {
1303
0
    UA_ModifyMonitoredItemsRequest *requestCopy =
1304
0
        UA_ModifyMonitoredItemsRequest_new();
1305
0
    if(!requestCopy)
1306
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
1307
0
    UA_StatusCode res =
1308
0
        UA_ModifyMonitoredItemsRequest_copy(&request, requestCopy);
1309
0
    if(res != UA_STATUSCODE_GOOD) {
1310
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1311
0
        return res;
1312
0
    }
1313
1314
0
    lockClient(client);
1315
1316
0
    res = __Client_AsyncServiceAdmission(client);
1317
0
    if(res != UA_STATUSCODE_GOOD) {
1318
0
        unlockClient(client);
1319
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1320
0
        return res;
1321
0
    }
1322
1323
    /* Get the subscription */
1324
0
    UA_Client_Subscription *sub = findSubscriptionById(client, request.subscriptionId);
1325
0
    if(!sub) {
1326
0
        unlockClient(client);
1327
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1328
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
1329
0
    }
1330
1331
0
    res = MonitoredItems_prepareModify(client, sub, requestCopy);
1332
0
    if(res != UA_STATUSCODE_GOOD) {
1333
0
        unlockClient(client);
1334
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1335
0
        return res;
1336
0
    }
1337
1338
0
    CustomCallback *cc = (CustomCallback*)UA_calloc(1, sizeof(CustomCallback));
1339
0
    if(!cc) {
1340
0
        MonitoredItems_reconcileModify(sub, requestCopy, NULL);
1341
0
        unlockClient(client);
1342
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1343
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
1344
0
    }
1345
0
    cc->clientData = requestCopy;
1346
0
    cc->callback.modifyMonitoredItems = callback;
1347
0
    cc->userData = userdata;
1348
1349
    /* Call the service */
1350
0
    UA_StatusCode statusCode = __Client_AsyncServiceAdmitted(
1351
0
        client, requestCopy, &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSREQUEST],
1352
0
        MonitoredItems_modify_async_handler,
1353
0
        &UA_TYPES[UA_TYPES_MODIFYMONITOREDITEMSRESPONSE], cc, requestId);
1354
0
    if(statusCode != UA_STATUSCODE_GOOD) {
1355
0
        MonitoredItems_reconcileModify(sub, requestCopy, NULL);
1356
0
        UA_ModifyMonitoredItemsRequest_delete(requestCopy);
1357
0
        UA_free(cc);
1358
0
    }
1359
1360
0
    unlockClient(client);
1361
0
    return statusCode;
1362
0
}
1363
1364
UA_StatusCode
1365
UA_Client_MonitoredItem_getContext(UA_Client *client, UA_UInt32 subscriptionId,
1366
0
                                   UA_UInt32 monitoredItemId, void **monContext) {
1367
0
    if(!client || !monContext)
1368
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
1369
1370
0
    *monContext = NULL;
1371
1372
0
    lockClient(client);
1373
0
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
1374
0
    if(!sub) {
1375
0
        unlockClient(client);
1376
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
1377
0
    }
1378
1379
0
    UA_StatusCode itemstatus = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
1380
0
    UA_Client_MonitoredItem *monItem = findMonitoredItemById(sub, monitoredItemId);
1381
0
    if(monItem) {
1382
0
        *monContext = monItem->context;
1383
0
        itemstatus = UA_STATUSCODE_GOOD;
1384
0
    }
1385
0
    unlockClient(client);
1386
0
    return itemstatus;
1387
0
}
1388
1389
UA_StatusCode
1390
UA_Client_MonitoredItem_setContext(UA_Client *client, UA_UInt32 subscriptionId,
1391
0
                                   UA_UInt32 monitoredItemId, void *monContext) {
1392
0
    if(!client)
1393
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
1394
1395
0
    lockClient(client);
1396
0
    UA_Client_Subscription *sub = findSubscriptionById(client, subscriptionId);
1397
0
    if(!sub) {
1398
0
        unlockClient(client);
1399
0
        return UA_STATUSCODE_BADSUBSCRIPTIONIDINVALID;
1400
0
    }
1401
1402
0
    UA_StatusCode itemstatus = UA_STATUSCODE_BADMONITOREDITEMIDINVALID;
1403
0
    UA_Client_MonitoredItem *monItem = findMonitoredItemById(sub, monitoredItemId);
1404
0
    if(monItem) {
1405
0
        monItem->context = monContext;
1406
0
        itemstatus = UA_STATUSCODE_GOOD;
1407
0
    }
1408
0
    unlockClient(client);
1409
0
    return itemstatus;
1410
0
}
1411
1412
/*************************************/
1413
/* Async Processing of Notifications */
1414
/*************************************/
1415
1416
/* Assume the request is already initialized */
1417
UA_StatusCode
1418
0
__Client_preparePublishRequest(UA_Client *client, UA_PublishRequest *request) {
1419
0
    UA_LOCK_ASSERT(&client->clientMutex);
1420
1421
    /* Count acks */
1422
0
    UA_Client_NotificationsAckNumber *ack;
1423
0
    LIST_FOREACH(ack, &client->pendingNotificationsAcks, listEntry)
1424
0
        ++request->subscriptionAcknowledgementsSize;
1425
1426
    /* Create the array. Returns a sentinel pointer if the length is zero. */
1427
0
    request->subscriptionAcknowledgements = (UA_SubscriptionAcknowledgement*)
1428
0
        UA_Array_new(request->subscriptionAcknowledgementsSize,
1429
0
                     &UA_TYPES[UA_TYPES_SUBSCRIPTIONACKNOWLEDGEMENT]);
1430
0
    if(!request->subscriptionAcknowledgements) {
1431
0
        request->subscriptionAcknowledgementsSize = 0;
1432
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
1433
0
    }
1434
1435
0
    size_t i = 0;
1436
0
    UA_Client_NotificationsAckNumber *ack_tmp;
1437
0
    LIST_FOREACH_SAFE(ack, &client->pendingNotificationsAcks, listEntry, ack_tmp) {
1438
0
        LIST_REMOVE(ack, listEntry);
1439
0
        UA_SubscriptionAcknowledgement *reqAck = &request->subscriptionAcknowledgements[i];
1440
0
        reqAck->sequenceNumber = ack->subAck.sequenceNumber;
1441
0
        reqAck->subscriptionId = ack->subAck.subscriptionId;
1442
0
        UA_free(ack);
1443
0
        i++;
1444
0
    }
1445
0
    return UA_STATUSCODE_GOOD;
1446
0
}
1447
1448
/* According to specification, Part 4 5.13.1, the value 0 is never used for the
1449
 * sequence number */
1450
static UA_UInt32
1451
0
__nextSequenceNumber(UA_UInt32 sequenceNumber) {
1452
0
    UA_UInt32 nextSequenceNumber = sequenceNumber + 1;
1453
0
    if(nextSequenceNumber == 0)
1454
0
        nextSequenceNumber = 1;
1455
0
    return nextSequenceNumber;
1456
0
}
1457
1458
static UA_Client_MonitoredItem *
1459
findMonitoredItemForNotification(UA_Client_Subscription *sub,
1460
0
                                 UA_UInt32 clientHandle) {
1461
0
    UA_Client_MonitoredItem *mon =
1462
0
        findMonitoredItemByHandle(sub, clientHandle);
1463
0
    if(mon)
1464
0
        return mon;
1465
1466
    /* A notification is the authoritative signal that the server has started
1467
     * to use the modified settings. Promote the embedded pending slot and
1468
     * re-key the existing tree node. */
1469
0
    mon = (UA_Client_MonitoredItem*)
1470
0
        ZIP_ITER(MonitorItemsTree, &sub->monitoredItems,
1471
0
                 MonitoredItem_findPendingByHandle, &clientHandle);
1472
0
    if(!mon)
1473
0
        return NULL;
1474
1475
0
    ZIP_REMOVE(MonitorItemsTree, &sub->monitoredItems, mon);
1476
0
    UA_MonitoringParameters_clear(&mon->parameters);
1477
0
    mon->parameters = mon->pendingParameters;
1478
0
    UA_MonitoringParameters_init(&mon->pendingParameters);
1479
0
    EventFields_clear(&mon->eventFields);
1480
0
    ZIP_INSERT(MonitorItemsTree, &sub->monitoredItems, mon);
1481
0
    return mon;
1482
0
}
1483
1484
static void
1485
processDataChangeNotification(UA_Client *client, UA_Client_Subscription *sub,
1486
0
                              UA_DataChangeNotification *dataChangeNotification) {
1487
0
    UA_LOCK_ASSERT(&client->clientMutex);
1488
1489
0
    for(size_t j = 0; j < dataChangeNotification->monitoredItemsSize; ++j) {
1490
0
        UA_MonitoredItemNotification *min = &dataChangeNotification->monitoredItems[j];
1491
1492
        /* Find the MonitoredItem */
1493
0
        UA_Client_MonitoredItem *mon =
1494
0
            findMonitoredItemForNotification(sub, min->clientHandle);
1495
1496
0
        if(!mon) {
1497
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1498
0
                           "Could not process a notification with clienthandle %" PRIu32
1499
0
                           " on subscription %" PRIu32, min->clientHandle, sub->subscriptionId);
1500
0
            continue;
1501
0
        }
1502
1503
0
        if(mon->isEventMonitoredItem) {
1504
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1505
0
                           "MonitoredItem is configured for Events. But received a "
1506
0
                           "DataChangeNotification.");
1507
0
            continue;
1508
0
        }
1509
1510
0
        if(mon->handler.dataChangeCallback) {
1511
0
            void *subC = sub->context;
1512
0
            void *monC = mon->context;
1513
0
            UA_UInt32 subId = sub->subscriptionId;
1514
0
            UA_UInt32 monId = mon->monitoredItemId;
1515
0
            mon->handler.dataChangeCallback(client, subId, subC, monId, monC, &min->value);
1516
0
        }
1517
0
    }
1518
0
}
1519
1520
static void
1521
processEventNotification(UA_Client *client, UA_Client_Subscription *sub,
1522
0
                         UA_EventNotificationList *eventNotificationList) {
1523
0
    UA_LOCK_ASSERT(&client->clientMutex);
1524
1525
0
    for(size_t j = 0; j < eventNotificationList->eventsSize; ++j) {
1526
0
        UA_EventFieldList *efl = &eventNotificationList->events[j];
1527
1528
        /* Find the MonitoredItem */
1529
0
        UA_Client_MonitoredItem *mon =
1530
0
            findMonitoredItemForNotification(sub, efl->clientHandle);
1531
1532
0
        if(!mon) {
1533
0
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
1534
0
                         "Could not process a notification with clienthandle %" PRIu32
1535
0
                         " on subscription %" PRIu32, efl->clientHandle,
1536
0
                         sub->subscriptionId);
1537
0
            continue;
1538
0
        }
1539
1540
0
        if(!mon->isEventMonitoredItem) {
1541
0
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
1542
0
                         "MonitoredItem is configured for DataChanges. But received a "
1543
0
                         "EventNotification");
1544
0
            continue;
1545
0
        }
1546
1547
        /* Build the callback metadata from the active filter only when it is
1548
         * first needed. After a promotion the cache is empty and rebuilt from
1549
         * the newly active parameters. */
1550
0
        if(mon->eventFields.mapSize == 0) {
1551
0
            UA_StatusCode res =
1552
0
                prepareEventFieldsMap(&mon->eventFields, &mon->parameters);
1553
0
            if(res != UA_STATUSCODE_GOOD) {
1554
0
                EventFields_clear(&mon->eventFields);
1555
0
                UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1556
0
                               "Could not prepare event fields: %s",
1557
0
                               UA_StatusCode_name(res));
1558
0
                continue;
1559
0
            }
1560
0
        }
1561
1562
0
        if(mon->eventFields.mapSize != efl->eventFieldsSize) {
1563
0
            UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
1564
0
                         "MonitoredItem received a EventNotification with the "
1565
0
                          "wrong number of event fields");
1566
0
            continue;
1567
0
        }
1568
1569
        /* Add the borrowed notification values to the cached field names. */
1570
0
        for(size_t i = 0; i < mon->eventFields.mapSize; i++)
1571
0
            mon->eventFields.map[i].value = efl->eventFields[i];
1572
0
        mon->handler.eventCallback(client, sub->subscriptionId, sub->context,
1573
0
                                   mon->monitoredItemId, mon->context,
1574
0
                                   mon->eventFields);
1575
0
    }
1576
0
}
1577
1578
static void
1579
processNotificationMessage(UA_Client *client, UA_Client_Subscription *sub,
1580
0
                           UA_ExtensionObject *msg) {
1581
0
    UA_LOCK_ASSERT(&client->clientMutex);
1582
1583
0
    if(msg->encoding != UA_EXTENSIONOBJECT_DECODED)
1584
0
        return;
1585
1586
    /* Handle DataChangeNotification */
1587
0
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_DATACHANGENOTIFICATION]) {
1588
0
        UA_DataChangeNotification *dataChangeNotification =
1589
0
            (UA_DataChangeNotification *)msg->content.decoded.data;
1590
0
        processDataChangeNotification(client, sub, dataChangeNotification);
1591
0
        return;
1592
0
    }
1593
1594
    /* Handle EventNotification */
1595
0
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_EVENTNOTIFICATIONLIST]) {
1596
0
        UA_EventNotificationList *eventNotificationList =
1597
0
            (UA_EventNotificationList *)msg->content.decoded.data;
1598
0
        processEventNotification(client, sub, eventNotificationList);
1599
0
        return;
1600
0
    }
1601
1602
    /* Handle StatusChangeNotification */
1603
0
    if(msg->content.decoded.type == &UA_TYPES[UA_TYPES_STATUSCHANGENOTIFICATION]) {
1604
0
        if(sub->statusChangeCallback) {
1605
0
            void *subC = sub->context;
1606
0
            UA_UInt32 subId = sub->subscriptionId;
1607
0
            sub->statusChangeCallback(client, subId, subC,
1608
0
                                      (UA_StatusChangeNotification*)msg->content.decoded.data);
1609
0
        } else {
1610
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1611
0
                           "Dropped a StatusChangeNotification since no "
1612
0
                           "callback is registered");
1613
0
        }
1614
0
        return;
1615
0
    }
1616
1617
0
    UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1618
0
                   "Unknown notification message type");
1619
0
}
1620
1621
void
1622
__Client_Subscriptions_processPublishResponse(UA_Client *client, UA_PublishRequest *request,
1623
0
                                              UA_PublishResponse *response) {
1624
0
    UA_LOCK_ASSERT(&client->clientMutex);
1625
1626
    /* Reduce the number of "in-flight" PublishRequests */
1627
0
    client->currentlyOutStandingPublishRequests--;
1628
1629
    /* Process ServiceResult for bad StatusCodes without referring to a
1630
     * SubscriptionId */
1631
0
    switch(response->responseHeader.serviceResult) {
1632
0
    case UA_STATUSCODE_BADTOOMANYPUBLISHREQUESTS:
1633
        /* Correct the assumed number of max outstanding requests */
1634
0
        if(client->config.outStandingPublishRequests > 1) {
1635
0
            client->config.outStandingPublishRequests--;
1636
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1637
0
                           "PublishResponse: Too many PublishRequest, reduce "
1638
0
                           "outStandingPublishRequests to %" PRId16,
1639
0
                           client->config.outStandingPublishRequests);
1640
0
        } else {
1641
0
            UA_LOG_ERROR(client->config.logging, UA_LOGCATEGORY_CLIENT,
1642
0
                         "PublishResponse: Too many PublishRequests when "
1643
0
                         "outStandingPublishRequests = 1");
1644
0
            UA_Client_Subscriptions_deleteSingle(client, response->subscriptionId);
1645
0
        }
1646
0
        return;
1647
1648
0
    case UA_STATUSCODE_BADSHUTDOWN:
1649
        /* If the remote server shuts down, DEBUG-log to avoid a warning-storm
1650
         * for normal operations */
1651
0
        UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
1652
0
                     "PublishResponse: Received BadShutdown status");
1653
0
        return;
1654
1655
0
    case UA_STATUSCODE_BADNOSUBSCRIPTION:
1656
        /* There is no Subscription configured, the server expects no
1657
         * PublishRequests. We demote this to debug-logging, as it can occur
1658
         * during regular shutdown when the Subscriptions are removed. */
1659
0
        UA_LOG_DEBUG(client->config.logging, UA_LOGCATEGORY_CLIENT,
1660
0
                     "PublishResponse: Received BadNoSubscription status");
1661
0
        return;
1662
1663
0
    default:
1664
0
        break;
1665
0
    }
1666
1667
    /* Get the Subscription */
1668
0
    UA_Client_Subscription *sub = findSubscriptionById(client, response->subscriptionId);
1669
0
    if(!sub) {
1670
0
        response->responseHeader.serviceResult = UA_STATUSCODE_BADNOSUBSCRIPTION;
1671
0
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1672
0
                       "PublishResponse: Received response for an unknown Subscription");
1673
0
        return;
1674
0
    }
1675
1676
    /* Process ServiceResult */
1677
0
    switch(response->responseHeader.serviceResult) {
1678
0
    case UA_STATUSCODE_BADSESSIONCLOSED:
1679
        /* The Session no longer exists on the server - remove the Subscription */
1680
0
        __Client_Subscription_deleteInternal(client, sub);
1681
0
        return;
1682
1683
0
    case UA_STATUSCODE_BADTIMEOUT:
1684
0
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1685
0
                       "PublishResponse: Aborted with BadTimeout status");
1686
0
        if(client->config.subscriptionInactivityCallback) {
1687
0
            void *subC = sub->context;
1688
0
            UA_UInt32 subId = sub->subscriptionId;
1689
0
            client->config.subscriptionInactivityCallback(client, subId, subC);
1690
0
        }
1691
0
        return;
1692
1693
0
    case UA_STATUSCODE_GOOD:
1694
0
        break; /* Continue below */
1695
1696
0
    default:
1697
        /* Catch-all for other bad StatusCodes */
1698
0
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1699
0
                       "PublishResponse: Received %s status",
1700
0
                       UA_StatusCode_name(response->responseHeader.serviceResult));
1701
0
        return;
1702
0
    }
1703
1704
    /* Update the LastActivity for the Subscription */
1705
0
    UA_EventLoop *el = client->config.eventLoop;
1706
0
    sub->lastActivity = el->dateTime_nowMonotonic(el);
1707
1708
    /* Detect missing message - OPC Unified Architecture, Part 4 5.13.1.1 e) */
1709
0
    UA_NotificationMessage *msg = &response->notificationMessage;
1710
0
    if(__nextSequenceNumber(sub->sequenceNumber) != msg->sequenceNumber) {
1711
0
        UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1712
0
                       "PublishResponse: Invalid subscription sequence number: "
1713
0
                       "Expected %" PRIu32 " but got %" PRIu32,
1714
0
                       __nextSequenceNumber(sub->sequenceNumber), msg->sequenceNumber);
1715
        /* This is an error. But we do not abort the connection. Some server
1716
         * SDKs misbehave from time to time and send out-of-order sequence
1717
         * numbers. (Probably some multi-threading synchronization issue.) */
1718
        /* UA_Client_disconnect(client);
1719
           return; */
1720
0
    }
1721
1722
    /* According to f), a keep-alive message contains no notifications and has
1723
     * the sequence number of the next NotificationMessage that is to be sent =>
1724
     * More than one consecutive keep-alive message or a NotificationMessage
1725
     * following a keep-alive message will share the same sequence number. */
1726
0
    if(msg->notificationDataSize)
1727
0
        sub->sequenceNumber = msg->sequenceNumber;
1728
1729
    /* Process the notification messages */
1730
0
    for(size_t k = 0; k < msg->notificationDataSize; ++k)
1731
0
        processNotificationMessage(client, sub, &msg->notificationData[k]);
1732
1733
    /* Add the current NotificationMessage (SequenceNumber) to the list of
1734
     * pending acks to be acknowledged. But only if it is in the list of
1735
     * sequence numbers the server has available. */
1736
0
    for(size_t i = 0; i < response->availableSequenceNumbersSize; i++) {
1737
0
        if(response->availableSequenceNumbers[i] != msg->sequenceNumber)
1738
0
            continue;
1739
0
        UA_Client_NotificationsAckNumber *tmpAck = (UA_Client_NotificationsAckNumber*)
1740
0
            UA_malloc(sizeof(UA_Client_NotificationsAckNumber));
1741
0
        if(!tmpAck) {
1742
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1743
0
                           "PublishResponse: Not enough memory to store the pending "
1744
0
                           "acknowledgement for Subscription %" PRIu32, sub->subscriptionId);
1745
0
            break;
1746
0
        }
1747
0
        tmpAck->subAck.sequenceNumber = msg->sequenceNumber;
1748
0
        tmpAck->subAck.subscriptionId = sub->subscriptionId;
1749
0
        LIST_INSERT_HEAD(&client->pendingNotificationsAcks, tmpAck, listEntry);
1750
0
        break;
1751
0
    }
1752
0
}
1753
1754
static void
1755
processPublishResponseAsync(UA_Client *client, void *userdata,
1756
0
                            UA_UInt32 requestId, void *response) {
1757
0
    UA_PublishRequest *req = (UA_PublishRequest*)userdata;
1758
0
    UA_PublishResponse *res = (UA_PublishResponse*)response;
1759
1760
0
    lockClient(client);
1761
1762
    /* Process the response */
1763
0
    __Client_Subscriptions_processPublishResponse(client, req, res);
1764
1765
    /* Delete the cached request */
1766
0
    UA_PublishRequest_delete(req);
1767
1768
    /* Fill up the outstanding publish requests */
1769
0
    __Client_Subscriptions_backgroundPublish(client);
1770
1771
0
    unlockClient(client);
1772
0
}
1773
1774
void
1775
120
__Client_Subscriptions_clear(UA_Client *client) {
1776
120
    UA_Client_NotificationsAckNumber *n;
1777
120
    UA_Client_NotificationsAckNumber *tmp;
1778
120
    LIST_FOREACH_SAFE(n, &client->pendingNotificationsAcks, listEntry, tmp) {
1779
0
        LIST_REMOVE(n, listEntry);
1780
0
        UA_free(n);
1781
0
    }
1782
1783
120
    UA_Client_Subscription *sub;
1784
120
    UA_Client_Subscription *tmps;
1785
120
    LIST_FOREACH_SAFE(sub, &client->subscriptions, listEntry, tmps)
1786
0
        __Client_Subscription_deleteInternal(client, sub); /* force local removal */
1787
1788
120
    client->monitoredItemHandles = 0;
1789
120
}
1790
1791
void
1792
0
__Client_Subscriptions_backgroundPublishInactivityCheck(UA_Client *client) {
1793
0
    UA_LOCK_ASSERT(&client->clientMutex);
1794
1795
0
    UA_EventLoop *el = client->config.eventLoop;
1796
0
    UA_DateTime nowm = el->dateTime_nowMonotonic(el);
1797
1798
0
    UA_Client_Subscription *sub;
1799
0
    LIST_FOREACH(sub, &client->subscriptions, listEntry) {
1800
0
        UA_DateTime maxSilence = (UA_DateTime)
1801
0
            ((sub->publishingInterval * sub->maxKeepAliveCount) +
1802
0
             client->config.timeout) * UA_DATETIME_MSEC;
1803
0
        if(maxSilence + sub->lastActivity < nowm) {
1804
            /* Reset activity */
1805
0
            sub->lastActivity = nowm;
1806
1807
0
            if(client->config.subscriptionInactivityCallback) {
1808
0
                void *subC = sub->context;
1809
0
                UA_UInt32 subId = sub->subscriptionId;
1810
0
                client->config.subscriptionInactivityCallback(client, subId, subC);
1811
0
            }
1812
0
            UA_LOG_WARNING(client->config.logging, UA_LOGCATEGORY_CLIENT,
1813
0
                           "Inactivity for Subscription %" PRIu32 ".",
1814
0
                           sub->subscriptionId);
1815
0
        }
1816
0
    }
1817
0
}
1818
1819
void
1820
0
__Client_Subscriptions_backgroundPublish(UA_Client *client) {
1821
0
    UA_LOCK_ASSERT(&client->clientMutex);
1822
1823
0
    if(client->sessionState != UA_SESSIONSTATE_ACTIVATED)
1824
0
        return;
1825
1826
    /* The session must have at least one subscription */
1827
0
    if(!LIST_FIRST(&client->subscriptions))
1828
0
        return;
1829
1830
0
    while(client->currentlyOutStandingPublishRequests < client->config.outStandingPublishRequests) {
1831
0
        UA_PublishRequest *request = UA_PublishRequest_new();
1832
0
        if(!request)
1833
0
            return;
1834
1835
        /* Publish requests are valid for 10 minutes */
1836
0
        request->requestHeader.timeoutHint = 10 * 60 * 1000;
1837
1838
0
        UA_StatusCode retval = __Client_preparePublishRequest(client, request);
1839
0
        if(retval != UA_STATUSCODE_GOOD) {
1840
0
            UA_PublishRequest_delete(request);
1841
0
            return;
1842
0
        }
1843
1844
0
        retval = __Client_AsyncServiceInternal(client, request,
1845
0
                                         &UA_TYPES[UA_TYPES_PUBLISHREQUEST],
1846
0
                                         processPublishResponseAsync,
1847
0
                                         &UA_TYPES[UA_TYPES_PUBLISHRESPONSE],
1848
0
                                         (void*)request, NULL);
1849
0
        if(retval != UA_STATUSCODE_GOOD) {
1850
0
            UA_PublishRequest_delete(request);
1851
0
            return;
1852
0
        }
1853
1854
0
        client->currentlyOutStandingPublishRequests++;
1855
0
    }
1856
0
}
1857
1858
UA_SetPublishingModeResponse
1859
UA_Client_Subscriptions_setPublishingMode(UA_Client *client,
1860
0
                                          const UA_SetPublishingModeRequest request) {
1861
0
    UA_SetPublishingModeResponse response;
1862
0
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETPUBLISHINGMODEREQUEST],
1863
0
                        &response, &UA_TYPES[UA_TYPES_SETPUBLISHINGMODERESPONSE]);
1864
0
    return response;
1865
0
}
1866
1867
UA_SetMonitoringModeResponse
1868
UA_Client_MonitoredItems_setMonitoringMode(UA_Client *client,
1869
0
                                           const UA_SetMonitoringModeRequest request) {
1870
0
    UA_SetMonitoringModeResponse response;
1871
0
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETMONITORINGMODEREQUEST],
1872
0
                        &response, &UA_TYPES[UA_TYPES_SETMONITORINGMODERESPONSE]);
1873
0
    return response;
1874
0
}
1875
1876
static void
1877
MonitoredItems_setMonitoringMode_async_handler(UA_Client *client, void *userdata,
1878
0
                                               UA_UInt32 requestId, void *response) {
1879
0
    UA_AsyncCallbackContext *ctx = (UA_AsyncCallbackContext*)userdata;
1880
0
    if(ctx->callback.setMonitoringMode)
1881
0
        ctx->callback.setMonitoringMode(
1882
0
            client, ctx->userdata, requestId,
1883
0
            (UA_SetMonitoringModeResponse*)response);
1884
0
}
1885
1886
UA_StatusCode
1887
UA_Client_MonitoredItems_setMonitoringMode_async(UA_Client *client,
1888
                                                 const UA_SetMonitoringModeRequest request,
1889
                                                 UA_ClientAsyncSetMonitoringModeCallback callback,
1890
0
                                                 void *userdata, UA_UInt32 *requestId) {
1891
0
    UA_AsyncCallbackContext ctx;
1892
0
    UA_StatusCode res;
1893
0
    ctx.callback.setMonitoringMode = callback;
1894
0
    ctx.userdata = userdata;
1895
0
    ctx.resultType = NULL;
1896
0
    ctx.attributeId = UA_ATTRIBUTEID_INVALID;
1897
0
    lockClient(client);
1898
0
    res = __Client_AsyncServiceWithContext(
1899
0
        client, &request, &UA_TYPES[UA_TYPES_SETMONITORINGMODEREQUEST],
1900
0
        MonitoredItems_setMonitoringMode_async_handler,
1901
0
        &UA_TYPES[UA_TYPES_SETMONITORINGMODERESPONSE], NULL, &ctx, requestId);
1902
0
    unlockClient(client);
1903
0
    return res;
1904
0
}
1905
1906
UA_SetTriggeringResponse
1907
UA_Client_MonitoredItems_setTriggering(UA_Client *client,
1908
0
                                       const UA_SetTriggeringRequest request) {
1909
0
    UA_SetTriggeringResponse response;
1910
0
    __UA_Client_Service(client, &request, &UA_TYPES[UA_TYPES_SETTRIGGERINGREQUEST],
1911
0
                        &response, &UA_TYPES[UA_TYPES_SETTRIGGERINGRESPONSE]);
1912
0
    return response;
1913
0
}
1914
1915
static void
1916
MonitoredItems_setTriggering_async_handler(UA_Client *client, void *userdata,
1917
0
                                           UA_UInt32 requestId, void *response) {
1918
0
    UA_AsyncCallbackContext *ctx = (UA_AsyncCallbackContext*)userdata;
1919
0
    if(ctx->callback.setTriggering)
1920
0
        ctx->callback.setTriggering(
1921
0
            client, ctx->userdata, requestId,
1922
0
            (UA_SetTriggeringResponse*)response);
1923
0
}
1924
1925
UA_StatusCode
1926
UA_Client_MonitoredItems_setTriggering_async(UA_Client *client,
1927
                                             const UA_SetTriggeringRequest request,
1928
                                             UA_ClientAsyncSetTriggeringCallback callback,
1929
0
                                             void *userdata, UA_UInt32 *requestId) {
1930
0
    UA_AsyncCallbackContext ctx;
1931
0
    UA_StatusCode res;
1932
0
    ctx.callback.setTriggering = callback;
1933
0
    ctx.userdata = userdata;
1934
0
    ctx.resultType = NULL;
1935
0
    ctx.attributeId = UA_ATTRIBUTEID_INVALID;
1936
0
    lockClient(client);
1937
0
    res = __Client_AsyncServiceWithContext(
1938
0
        client, &request, &UA_TYPES[UA_TYPES_SETTRIGGERINGREQUEST],
1939
0
        MonitoredItems_setTriggering_async_handler,
1940
0
        &UA_TYPES[UA_TYPES_SETTRIGGERINGRESPONSE], NULL, &ctx, requestId);
1941
0
    unlockClient(client);
1942
0
    return res;
1943
0
}