Coverage Report

Created: 2026-11-05 07:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/src/open62541_15/src/pubsub/ua_pubsub_reader.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 (c) 2017-2025 Fraunhofer IOSB (Author: Andreas Ebner)
6
 * Copyright (c) 2019 Fraunhofer IOSB (Author: Julius Pfrommer)
7
 * Copyright (c) 2019 Kalycito Infotech Private Limited
8
 * Copyright (c) 2021 Fraunhofer IOSB (Author: Jan Hermes)
9
 * Copyright (c) 2022 Siemens AG (Author: Thomas Fischer)
10
 * Copyright (c) 2022 Fraunhofer IOSB (Author: Noel Graf)
11
 */
12
13
#include "ua_pubsub_internal.h"
14
15
#ifdef UA_ENABLE_PUBSUB /* conditional compilation */
16
17
#include "ua_pubsub_networkmessage.h"
18
19
static UA_Boolean
20
0
publisherIdIsMatching(UA_NetworkMessage *msg, UA_PublisherId *idB) {
21
0
    if(!msg->publisherIdEnabled)
22
0
        return true;
23
0
    UA_PublisherId *idA = &msg->publisherId;
24
0
    if(idA->idType != idB->idType)
25
0
        return false;
26
0
    switch(idA->idType) {
27
0
        case UA_PUBLISHERIDTYPE_BYTE:   return idA->id.byte == idB->id.byte;
28
0
        case UA_PUBLISHERIDTYPE_UINT16: return idA->id.uint16 == idB->id.uint16;
29
0
        case UA_PUBLISHERIDTYPE_UINT32: return idA->id.uint32 == idB->id.uint32;
30
0
        case UA_PUBLISHERIDTYPE_UINT64: return idA->id.uint64 == idB->id.uint64;
31
0
        case UA_PUBLISHERIDTYPE_STRING: return UA_String_equal(&idA->id.string, &idB->id.string);
32
0
        default: break;
33
0
    }
34
0
    return false;
35
0
}
36
37
#if UA_LOGLEVEL <= 200
38
static void
39
printPublisherId(char *out, size_t size, UA_PublisherId *id) {
40
    switch(id->idType) {
41
    case UA_PUBLISHERIDTYPE_BYTE:   mp_snprintf(out, size, "(b)%u", (unsigned)id->id.byte); break;
42
    case UA_PUBLISHERIDTYPE_UINT16: mp_snprintf(out, size, "(u16)%u", (unsigned)id->id.uint16); break;
43
    case UA_PUBLISHERIDTYPE_UINT32: mp_snprintf(out, size, "(u32)%u", (unsigned)id->id.uint32); break;
44
    case UA_PUBLISHERIDTYPE_UINT64: mp_snprintf(out, size, "(u64)%lu", id->id.uint64); break;
45
    case UA_PUBLISHERIDTYPE_STRING: mp_snprintf(out, size, "\"%S\"", id->id.string); break;
46
    default: out[0] = 0; break;
47
    }
48
}
49
#endif
50
51
UA_StatusCode
52
UA_DataSetReader_checkIdentifier(UA_PubSubManager *psm, UA_DataSetReader *dsr,
53
0
                                 UA_NetworkMessage *msg) {
54
0
    if(!publisherIdIsMatching(msg, &dsr->config.publisherId)) {
55
#if UA_LOGLEVEL <= 200
56
        char idAStr[512];
57
        char idBStr[512];
58
        printPublisherId(idAStr, 512, &dsr->config.publisherId);
59
        printPublisherId(idBStr, 512, &msg->publisherId);
60
        UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "PublisherId does not match. "
61
                            "Expected %s, received %s", idAStr, idBStr);
62
#endif
63
0
        return UA_STATUSCODE_BADNOTFOUND;
64
0
    }
65
66
0
    UA_ReaderGroup *rg = dsr->linkedReaderGroup;
67
0
    if(rg->config.encodingMimeType == UA_PUBSUB_ENCODING_JSON) {
68
        // TODO
69
        /* if(dsr->config.dataSetWriterId == */
70
        /*    *msg->payloadHeader.dataSetPayloadHeader.dataSetWriterIds) { */
71
        /*     return UA_STATUSCODE_GOOD; */
72
        /* } */
73
74
        /* UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "DataSetWriterId does not match. " */
75
        /*                     "Expected %u, received %u", dsr->config.dataSetWriterId, */
76
        /*                     *msg->payloadHeader.dataSetPayloadHeader.dataSetWriterIds); */
77
0
        return UA_STATUSCODE_BADNOTFOUND;
78
0
    }
79
80
0
    if(msg->groupHeaderEnabled && msg->groupHeader.writerGroupIdEnabled) {
81
0
        if(dsr->config.writerGroupId != msg->groupHeader.writerGroupId) {
82
0
            UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "WriterGroupId does not match. "
83
0
                                "Expected %u, received %u", dsr->config.writerGroupId,
84
0
                                msg->groupHeader.writerGroupId);
85
0
            return UA_STATUSCODE_BADNOTFOUND;
86
0
        }
87
0
    }
88
89
0
    if(msg->payloadHeaderEnabled) {
90
0
        for(size_t i = 0; i < msg->messageCount; i++) {
91
0
            if(dsr->config.dataSetWriterId == msg->dataSetWriterIds[i])
92
0
                return UA_STATUSCODE_GOOD;
93
0
        }
94
0
        UA_LOG_DEBUG_PUBSUB(psm->logging, dsr,
95
0
                            "DataSetWriterIds in the payload do not match");
96
0
        return UA_STATUSCODE_BADNOTFOUND;
97
0
    }
98
99
0
    return UA_STATUSCODE_GOOD;
100
0
}
101
102
UA_DataSetReader *
103
0
UA_DataSetReader_find(UA_PubSubManager *psm, const UA_NodeId id) {
104
0
    if(!psm)
105
0
        return NULL;
106
0
    UA_PubSubConnection *psc;
107
0
    TAILQ_FOREACH(psc, &psm->connections, listEntry) {
108
0
        UA_ReaderGroup *rg;
109
0
        LIST_FOREACH(rg, &psc->readerGroups, listEntry) {
110
0
            UA_DataSetReader *dsr;
111
0
            LIST_FOREACH(dsr, &rg->readers, listEntry) {
112
0
                if(UA_NodeId_equal(&id, &dsr->head.identifier))
113
0
                    return dsr;
114
0
            }
115
0
        }
116
0
    }
117
0
    return NULL;
118
0
}
119
120
static UA_StatusCode
121
0
validateDSRConfig(UA_PubSubManager *psm, UA_DataSetReader *dsr) {
122
    /* Check if used dataSet metaData is valid in context of the rest of the config */
123
0
    if(dsr->config.dataSetFieldContentMask & UA_DATASETFIELDCONTENTMASK_RAWDATA) {
124
0
        for(size_t i = 0; i < dsr->config.dataSetMetaData.fieldsSize; i++) {
125
0
            const UA_FieldMetaData *field = &dsr->config.dataSetMetaData.fields[i];
126
0
            if((field->builtInType == UA_NS0ID_STRING || field->builtInType == UA_NS0ID_BYTESTRING) &&
127
0
               field->maxStringLength == 0) {
128
                /* Fields of type String or ByteString need to have defined
129
                 * MaxStringLength*/
130
0
                UA_LOG_ERROR_PUBSUB(psm->logging, dsr,
131
0
                                    "Add DataSetReader failed. MaxStringLength must be "
132
0
                                    "set in MetaData when using RawData field encoding.");
133
0
                return UA_STATUSCODE_BADCONFIGURATIONERROR;
134
0
            }
135
136
            /* Warn if the field is a structure type that contains
137
             * String/ByteString members. MaxStringLength padding is not
138
             * yet applied for strings nested inside structures. */
139
0
            const UA_DataType *type =
140
0
                UA_findDataTypeWithCustom(&field->dataType,
141
0
                                          psm->sc.server->config.customDataTypes);
142
0
            if(type &&
143
0
               (type->typeKind == UA_DATATYPEKIND_STRUCTURE ||
144
0
                type->typeKind == UA_DATATYPEKIND_UNION ||
145
0
                type->typeKind == UA_DATATYPEKIND_OPTSTRUCT) &&
146
0
               typeContainsString(type, 0)) {
147
0
                UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
148
0
                                      "Field %.*s uses a structure type that "
149
0
                                      "contains String/ByteString members. "
150
0
                                      "MaxStringLength padding is not applied for "
151
0
                                      "strings inside structures with RawData encoding.",
152
0
                                      (int)field->name.length, field->name.data);
153
0
            }
154
0
        }
155
0
    }
156
0
    return UA_STATUSCODE_GOOD;
157
0
}
158
159
static void
160
0
disconnectDSR2Standalone(UA_PubSubManager *psm, UA_DataSetReader *dsr) {
161
    /* Check if a sds name is defined */
162
0
    const UA_String sdsName = dsr->config.linkedStandaloneSubscribedDataSetName;
163
0
    if(UA_String_isEmpty(&sdsName))
164
0
        return;
165
166
0
    UA_SubscribedDataSet *sds = UA_SubscribedDataSet_findByName(psm, sdsName);
167
0
    if(!sds)
168
0
        return;
169
170
    /* Remove the backpointer from the sds */
171
0
    sds->connectedReader = NULL;
172
173
    /* Remove the references in the information model */
174
0
#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
175
0
    disconnectDataSetReaderToDataSet(psm->sc.server, dsr->head.identifier);
176
0
#endif
177
0
}
178
179
/* Connect to StandaloneSubscribedDataSet if a name is defined */
180
static UA_StatusCode
181
0
connectDSR2Standalone(UA_PubSubManager *psm, UA_DataSetReader *dsr) {
182
    /* Check if a sds name is defined */
183
0
    const UA_String sdsName = dsr->config.linkedStandaloneSubscribedDataSetName;
184
0
    if(UA_String_isEmpty(&sdsName))
185
0
        return UA_STATUSCODE_GOOD;
186
187
0
    UA_SubscribedDataSet *sds = UA_SubscribedDataSet_findByName(psm, sdsName);
188
0
    if(!sds)
189
0
        return UA_STATUSCODE_BADNOTFOUND;
190
191
    /* Already connected? */
192
0
    if(sds->connectedReader) {
193
0
        if(sds->connectedReader != dsr)
194
0
            UA_LOG_ERROR_PUBSUB(psm->logging, dsr,
195
0
                                "Configured StandaloneSubscribedDataSet already "
196
0
                                "connected to a different DataSetReader");
197
0
        return UA_STATUSCODE_BADINTERNALERROR;
198
0
    }
199
200
    /* Check supported type */
201
0
    if(sds->config.subscribedDataSetType != UA_PUBSUB_SDS_TARGET) {
202
0
        UA_LOG_ERROR_PUBSUB(psm->logging, dsr,
203
0
                            "Not implemented! Currently only SubscribedDataSet as "
204
0
                            "TargetVariables is implemented");
205
0
        return UA_STATUSCODE_BADNOTIMPLEMENTED;
206
0
    }
207
208
0
    UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "Connecting SubscribedDataSet");
209
210
    /* Copy the metadata from the sds */
211
0
    UA_DataSetMetaDataType metaData;
212
0
    UA_StatusCode res = UA_DataSetMetaDataType_copy(&sds->config.dataSetMetaData, &metaData);
213
0
    if(res != UA_STATUSCODE_GOOD)
214
0
        return res;
215
216
    /* Prepare the input for _createTargetVariables and call it */
217
0
    UA_TargetVariablesDataType *tvs = &sds->config.subscribedDataSet.target;
218
0
    res = DataSetReader_createTargetVariables(psm, dsr, tvs->targetVariablesSize,
219
0
                                              tvs->targetVariables);
220
0
    if(res != UA_STATUSCODE_GOOD) {
221
0
        UA_DataSetMetaDataType_clear(&metaData);
222
0
        return res;
223
0
    }
224
225
    /* Use the metadata from the sds */
226
0
    UA_DataSetMetaDataType_clear(&dsr->config.dataSetMetaData);
227
0
    dsr->config.dataSetMetaData = metaData;
228
229
    /* Set the backpointer from the sds */
230
0
    sds->connectedReader = dsr;
231
232
    /* Make the connection visible in the information model */
233
0
#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
234
0
    return connectDataSetReaderToDataSet(psm->sc.server, dsr->head.identifier,
235
0
                                         sds->head.identifier);
236
#else
237
    return UA_STATUSCODE_GOOD;
238
#endif
239
0
}
240
241
UA_StatusCode
242
UA_DataSetReader_create(UA_PubSubManager *psm, UA_NodeId readerGroupIdentifier,
243
                        const UA_DataSetReaderConfig *dataSetReaderConfig,
244
0
                        UA_NodeId *readerIdentifier) {
245
0
    if(!psm || !dataSetReaderConfig)
246
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
247
248
0
    UA_LOCK_ASSERT(&psm->sc.server->serviceMutex);
249
250
    /* Search the reader group by the given readerGroupIdentifier */
251
0
    UA_ReaderGroup *rg = UA_ReaderGroup_find(psm, readerGroupIdentifier);
252
0
    if(!rg)
253
0
        return UA_STATUSCODE_BADNOTFOUND;
254
255
0
    if(UA_PubSubState_isEnabled(rg->head.state)) {
256
0
        UA_LOG_WARNING_PUBSUB(psm->logging, rg,
257
0
                              "Cannot add a DataSetReader while the "
258
0
                              "ReaderGroup is enabled");
259
0
        return UA_STATUSCODE_BADCONFIGURATIONERROR;
260
0
    }
261
262
    /* Allocate memory for new DataSetReader */
263
0
    UA_DataSetReader *dsr = (UA_DataSetReader *)
264
0
        UA_calloc(1, sizeof(UA_DataSetReader));
265
0
    if(!dsr)
266
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
267
268
0
    dsr->head.componentType = UA_PUBSUBCOMPONENT_DATASETREADER;
269
0
    dsr->linkedReaderGroup = rg;
270
271
    /* Add the new reader to the group. Add to the end of the linked list to
272
     * ensure the order for the realtime offsets is as expected. The received
273
     * DataSetMessages are matched via UA_DataSetReader_checkIdentifier for the
274
     * non-RT path. */
275
0
    UA_DataSetReader *after = LIST_FIRST(&rg->readers);
276
0
    if(!after) {
277
0
        LIST_INSERT_HEAD(&rg->readers, dsr, listEntry);
278
0
    } else {
279
0
        while(LIST_NEXT(after, listEntry))
280
0
            after = LIST_NEXT(after, listEntry);
281
0
        LIST_INSERT_AFTER(after, dsr, listEntry);
282
0
    }
283
0
    rg->readersCount++;
284
285
    /* Copy the config into the new dataSetReader */
286
0
    UA_StatusCode retVal =
287
0
        UA_DataSetReaderConfig_copy(dataSetReaderConfig, &dsr->config);
288
0
    if(retVal != UA_STATUSCODE_GOOD) {
289
0
        UA_PubSubComponent_freeWithoutLifecycleCallback(
290
0
            psm, dsr, UA_PUBSUBCOMPONENT_DATASETREADER);
291
0
        return retVal;
292
0
    }
293
294
0
#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
295
0
    retVal = addDataSetReaderRepresentation(psm->sc.server, dsr);
296
0
    if(retVal != UA_STATUSCODE_GOOD) {
297
0
        UA_LOG_ERROR_PUBSUB(psm->logging, rg,
298
0
                            "Adding the DataSetReader to the information model failed");
299
0
        UA_PubSubComponent_freeWithoutLifecycleCallback(
300
0
            psm, dsr, UA_PUBSUBCOMPONENT_DATASETREADER);
301
0
        return retVal;
302
0
    }
303
#else
304
    UA_PubSubManager_generateUniqueNodeId(psm, &dsr->head.identifier);
305
#endif
306
307
    /* Cache the log string */
308
0
    char tmpLogIdStr[128];
309
0
    mp_snprintf(tmpLogIdStr, 128, "%SDataSetReader %N\t| ",
310
0
                dsr->linkedReaderGroup->head.logIdString, dsr->head.identifier);
311
0
    dsr->head.logIdString = UA_STRING_ALLOC(tmpLogIdStr);
312
313
    /* Connect to StandaloneSubscribedDataSet if a name is defined. Needs to be
314
     * added in the information model first, as this adds references to the
315
     * StandaloneSubscribedDataSet. */
316
0
    retVal = connectDSR2Standalone(psm, dsr);
317
0
    if(retVal != UA_STATUSCODE_GOOD) {
318
0
        UA_PubSubComponent_freeWithoutLifecycleCallback(
319
0
            psm, dsr, UA_PUBSUBCOMPONENT_DATASETREADER);
320
0
        return retVal;
321
0
    }
322
323
    /* Validate the config */
324
0
    retVal = validateDSRConfig(psm, dsr);
325
0
    if(retVal != UA_STATUSCODE_GOOD) {
326
0
        UA_PubSubComponent_freeWithoutLifecycleCallback(
327
0
            psm, dsr, UA_PUBSUBCOMPONENT_DATASETREADER);
328
0
        return retVal;
329
0
    }
330
331
    /* Notify the application that a new Reader was created.
332
     * This may internally adjust the config */
333
0
    UA_Server *server = psm->sc.server;
334
0
    if(server->config.pubSubConfig.componentLifecycleCallback) {
335
0
        UA_StatusCode res = server->config.pubSubConfig.
336
0
            componentLifecycleCallback(server, dsr->head.identifier,
337
0
                                       UA_PUBSUBCOMPONENT_DATASETREADER, false);
338
0
        if(res != UA_STATUSCODE_GOOD) {
339
            /* The app refused the component; free without re-asking the
340
             * lifecycle callback (it would re-reject and leak the reader). */
341
0
            UA_PubSubComponent_freeWithoutLifecycleCallback(
342
0
                psm, dsr, UA_PUBSUBCOMPONENT_DATASETREADER);
343
0
            return res;
344
0
        }
345
0
    }
346
347
0
    UA_LOG_INFO_PUBSUB(psm->logging, dsr, "DataSetReader created (State: %s)",
348
0
                       UA_PubSubState_name(dsr->head.state));
349
350
0
    if(readerIdentifier)
351
0
        UA_NodeId_copy(&dsr->head.identifier, readerIdentifier);
352
353
    /* Enable the DataSetReader immediately if the enabled flag is set */
354
0
    if(dataSetReaderConfig->enabled)
355
0
        UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_OPERATIONAL,
356
0
                                        UA_STATUSCODE_GOOD);
357
358
0
    return UA_STATUSCODE_GOOD;
359
0
}
360
361
UA_StatusCode
362
0
UA_DataSetReader_remove(UA_PubSubManager *psm, UA_DataSetReader *dsr) {
363
0
    UA_LOCK_ASSERT(&psm->sc.server->serviceMutex);
364
365
0
    UA_ReaderGroup *rg = dsr->linkedReaderGroup;
366
0
    UA_assert(rg);
367
368
    /* Check if the ReaderGroup is enabled */
369
0
    if(UA_PubSubState_isEnabled(rg->head.state)) {
370
0
        UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
371
0
                              "Removal of the DataSetReader not possible while "
372
0
                              "the ReaderGroup is enabled");
373
0
        return UA_STATUSCODE_BADINTERNALERROR;
374
0
    }
375
376
    /* Check with the application if we can remove */
377
0
    UA_Server *server = psm->sc.server;
378
0
    if(server->config.pubSubConfig.componentLifecycleCallback) {
379
0
        UA_StatusCode res = server->config.pubSubConfig.
380
0
            componentLifecycleCallback(server, dsr->head.identifier,
381
0
                                       UA_PUBSUBCOMPONENT_DATASETREADER, true);
382
0
        if(res != UA_STATUSCODE_GOOD)
383
0
            return res;
384
0
    }
385
386
    /* Disable and signal to the application */
387
0
    UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_DISABLED,
388
0
                                    UA_STATUSCODE_BADSHUTDOWN);
389
390
    /* Remove from information model */
391
0
#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
392
0
    deleteNode(psm->sc.server, dsr->head.identifier, true);
393
0
#endif
394
395
    /* Check if a Standalone-SubscribedDataSet is associated with this reader
396
     * and disconnect it*/
397
0
    const UA_String sdsName = dsr->config.linkedStandaloneSubscribedDataSetName;
398
0
    UA_SubscribedDataSet *sds = (UA_String_isEmpty(&sdsName)) ?
399
0
        NULL : UA_SubscribedDataSet_findByName(psm, sdsName);
400
0
    if(sds && sds->connectedReader == dsr)
401
0
        sds->connectedReader = NULL;
402
403
    /* Remove DataSetReader from group */
404
0
    LIST_REMOVE(dsr, listEntry);
405
0
    rg->readersCount--;
406
407
0
    UA_LOG_INFO_PUBSUB(psm->logging, dsr, "DataSetReader deleted");
408
409
0
    UA_DataSetReaderConfig_clear(&dsr->config);
410
0
    UA_PubSubComponentHead_clear(&dsr->head);
411
0
    UA_free(dsr);
412
413
0
    return UA_STATUSCODE_GOOD;
414
0
}
415
416
UA_StatusCode
417
UA_DataSetReaderConfig_copy(const UA_DataSetReaderConfig *src,
418
0
                            UA_DataSetReaderConfig *dst) {
419
0
    memcpy(dst, src, sizeof(UA_DataSetReaderConfig));
420
0
    dst->writerGroupId = src->writerGroupId;
421
0
    dst->dataSetWriterId = src->dataSetWriterId;
422
0
    dst->dataSetFieldContentMask = src->dataSetFieldContentMask;
423
0
    dst->messageReceiveTimeout = src->messageReceiveTimeout;
424
425
0
    UA_StatusCode ret = UA_String_copy(&src->name, &dst->name);
426
0
    ret |= UA_PublisherId_copy(&src->publisherId, &dst->publisherId);
427
0
    ret |= UA_DataSetMetaDataType_copy(&src->dataSetMetaData, &dst->dataSetMetaData);
428
0
    ret |= UA_ExtensionObject_copy(&src->messageSettings, &dst->messageSettings);
429
0
    ret |= UA_ExtensionObject_copy(&src->transportSettings, &dst->transportSettings);
430
0
    ret |= UA_String_copy(&src->linkedStandaloneSubscribedDataSetName,
431
0
                             &dst->linkedStandaloneSubscribedDataSetName);
432
433
0
    if(src->subscribedDataSetType == UA_PUBSUB_SDS_TARGET) {
434
0
        ret |= UA_TargetVariablesDataType_copy(&src->subscribedDataSet.target,
435
0
                                               &dst->subscribedDataSet.target);
436
0
    }
437
438
0
    if(ret != UA_STATUSCODE_GOOD)
439
0
        UA_DataSetReaderConfig_clear(dst);
440
441
0
    return ret;
442
0
}
443
444
void
445
0
UA_DataSetReaderConfig_clear(UA_DataSetReaderConfig *cfg) {
446
0
    UA_String_clear(&cfg->name);
447
0
    UA_String_clear(&cfg->linkedStandaloneSubscribedDataSetName);
448
0
    UA_PublisherId_clear(&cfg->publisherId);
449
0
    UA_DataSetMetaDataType_clear(&cfg->dataSetMetaData);
450
0
    UA_ExtensionObject_clear(&cfg->messageSettings);
451
0
    UA_ExtensionObject_clear(&cfg->transportSettings);
452
0
    if(cfg->subscribedDataSetType == UA_PUBSUB_SDS_TARGET) {
453
0
        UA_TargetVariablesDataType_clear(&cfg->subscribedDataSet.target);
454
0
    }
455
0
}
456
457
UA_StatusCode
458
UA_DataSetReader_setPubSubState(UA_PubSubManager *psm, UA_DataSetReader *dsr,
459
0
                                UA_PubSubState targetState, UA_StatusCode errorReason) {
460
0
    UA_ReaderGroup *rg = dsr->linkedReaderGroup;
461
0
    UA_assert(rg);
462
463
    /* Callback to modify the WriterGroup config and change the targetState
464
     * before the state machine executes */
465
0
    UA_Server *server = psm->sc.server;
466
0
    if(server->config.pubSubConfig.beforeStateChangeCallback) {
467
0
        server->config.pubSubConfig.
468
0
            beforeStateChangeCallback(server, dsr->head.identifier, &targetState);
469
0
    }
470
471
0
    UA_StatusCode res = UA_STATUSCODE_GOOD;
472
0
    UA_PubSubState oldState = dsr->head.state;
473
474
    /* Custom state machine */
475
0
    if(dsr->config.customStateMachine) {
476
0
        res = dsr->config.customStateMachine(server, dsr->head.identifier,
477
0
                                             dsr->config.context,
478
0
                                             &dsr->head.state, targetState);
479
0
        if(res != UA_STATUSCODE_GOOD)
480
0
            errorReason = res;
481
0
        goto finalize_state_machine;
482
0
    }
483
484
    /* Internal state machine */
485
0
    switch(targetState) {
486
        /* Disabled */
487
0
    case UA_PUBSUBSTATE_DISABLED:
488
0
    case UA_PUBSUBSTATE_ERROR:
489
0
        dsr->head.state = targetState;
490
0
        break;
491
492
        /* Enabled */
493
0
    case UA_PUBSUBSTATE_PAUSED:
494
0
    case UA_PUBSUBSTATE_PREOPERATIONAL:
495
0
    case UA_PUBSUBSTATE_OPERATIONAL:
496
0
        if(rg->head.state == UA_PUBSUBSTATE_DISABLED ||
497
0
           rg->head.state == UA_PUBSUBSTATE_ERROR ||
498
0
           rg->head.state == UA_PUBSUBSTATE_PAUSED) {
499
0
            dsr->head.state = UA_PUBSUBSTATE_PAUSED; /* RG is disabled -> paused */
500
0
        } else {
501
0
            dsr->head.state = rg->head.state; /* RG is enabled -> same state */
502
0
        }
503
0
        break;
504
505
0
    default:
506
0
        dsr->head.state = UA_PUBSUBSTATE_ERROR;
507
0
        res = UA_STATUSCODE_BADINTERNALERROR;
508
0
        errorReason = res;
509
0
        break;
510
0
    }
511
512
    /* Only keep the timeout callback if the reader is operational */
513
0
    if(dsr->head.state != UA_PUBSUBSTATE_OPERATIONAL &&
514
0
       dsr->msgRcvTimeoutTimerId != 0) {
515
0
        UA_EventLoop *el = psm->sc.server->config.eventLoop;
516
0
        el->removeTimer(el, dsr->msgRcvTimeoutTimerId);
517
0
        dsr->msgRcvTimeoutTimerId = 0;
518
0
    }
519
520
0
 finalize_state_machine:
521
522
    /* No state change has happened */
523
0
    if(dsr->head.state == oldState)
524
0
        return res;
525
526
0
    UA_LOG_INFO_PUBSUB(psm->logging, dsr, "%s -> %s",
527
0
                       UA_PubSubState_name(oldState),
528
0
                       UA_PubSubState_name(dsr->head.state));
529
530
    /* Inform application about state change */
531
0
    if(server->config.pubSubConfig.stateChangeCallback)
532
0
        server->config.pubSubConfig.
533
0
            stateChangeCallback(server, dsr->head.identifier,
534
0
                                dsr->head.state, errorReason);
535
                                
536
0
    return res;
537
0
}
538
539
/* This Method is used to initially set the SubscribedDataSet to
540
 * TargetVariablesType and to create the list of target Variables of a
541
 * SubscribedDataSetType. */
542
UA_StatusCode
543
DataSetReader_createTargetVariables(UA_PubSubManager *psm, UA_DataSetReader *dsr,
544
0
                                    size_t tvsSize, const UA_FieldTargetDataType *tvs) {
545
0
    UA_LOCK_ASSERT(&psm->sc.server->serviceMutex);
546
547
0
    if(UA_PubSubState_isEnabled(dsr->head.state)) {
548
0
        UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
549
0
                              "Cannot create Target Variables failed while "
550
0
                              "the DataSetReader is enabled");
551
0
        return UA_STATUSCODE_BADCONFIGURATIONERROR;
552
0
    }
553
554
0
    UA_TargetVariablesDataType newVars;
555
0
    UA_TargetVariablesDataType tmp = {tvsSize, (UA_FieldTargetDataType*)(uintptr_t)tvs};
556
0
    UA_StatusCode res = UA_TargetVariablesDataType_copy(&tmp, &newVars);   
557
0
    if(res != UA_STATUSCODE_GOOD)
558
0
        return res;
559
560
0
    UA_TargetVariablesDataType_clear(&dsr->config.subscribedDataSet.target);
561
0
    dsr->config.subscribedDataSet.target = newVars;
562
0
    return UA_STATUSCODE_GOOD;
563
0
}
564
565
static void
566
UA_DataSetReader_handleMessageReceiveTimeout(UA_PubSubManager *psm,
567
0
                                             UA_DataSetReader *dsr) {
568
0
    UA_assert(dsr->head.componentType == UA_PUBSUBCOMPONENT_DATASETREADER);
569
570
    /* Don't signal an error if we don't expect messages to arrive */
571
0
    if(dsr->head.state != UA_PUBSUBSTATE_OPERATIONAL &&
572
0
       dsr->head.state != UA_PUBSUBSTATE_PREOPERATIONAL)
573
0
        return;
574
575
0
    UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "Message receive timeout occurred");
576
577
0
    lockServer(psm->sc.server);
578
0
    UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_ERROR,
579
0
                                    UA_STATUSCODE_BADTIMEOUT);
580
0
    unlockServer(psm->sc.server);
581
0
}
582
583
void
584
UA_DataSetReader_process(UA_PubSubManager *psm, UA_DataSetReader *dsr,
585
0
                         UA_DataSetMessage *msg) {
586
0
    if(!dsr || !msg || !psm)
587
0
        return;
588
589
0
    UA_LOG_DEBUG_PUBSUB(psm->logging, dsr, "Received a network message");
590
591
    /* Received a (first) message for the Reader.
592
     * Transition from PreOperational to Operational. */
593
0
    if(dsr->head.state == UA_PUBSUBSTATE_PREOPERATIONAL)
594
0
        UA_DataSetReader_setPubSubState(psm, dsr, dsr->head.state, UA_STATUSCODE_GOOD);
595
596
0
    if(dsr->head.state != UA_PUBSUBSTATE_OPERATIONAL &&
597
0
       dsr->head.state != UA_PUBSUBSTATE_PREOPERATIONAL) {
598
0
        UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
599
0
                              "Received a network message but not operational");
600
0
        return;
601
0
    }
602
603
0
    if(!msg->header.dataSetMessageValid) {
604
0
        UA_LOG_INFO_PUBSUB(psm->logging, dsr,
605
0
                           "DataSetMessage is discarded: message is not valid");
606
0
        return;
607
0
    }
608
609
    /* TODO: Check ConfigurationVersion */
610
    /* if(msg->header.configVersionMajorVersionEnabled) {
611
     *     if(msg->header.configVersionMajorVersion !=
612
     *            dsr->config.dataSetMetaData.configurationVersion.majorVersion) {
613
     *         UA_LOG_WARNING(psm->logging, UA_LOGCATEGORY_SERVER,
614
     *                        "DataSetMessage is discarded: ConfigurationVersion "
615
     *                        "MajorVersion does not match");
616
     *         return;
617
     *     }
618
     * } */
619
620
0
    if(msg->header.dataSetMessageType != UA_DATASETMESSAGE_DATAKEYFRAME) {
621
0
        UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
622
0
                              "DataSetMessage is discarded: Only keyframes are supported");
623
0
        return;
624
0
    }
625
626
    /* Configure / Update the timeout callback */
627
0
    if(dsr->config.messageReceiveTimeout > 0.0) {
628
0
        UA_EventLoop *el = psm->sc.server->config.eventLoop;
629
0
        if(dsr->msgRcvTimeoutTimerId == 0) {
630
0
            el->addTimer(el, (UA_Callback)UA_DataSetReader_handleMessageReceiveTimeout,
631
0
                         psm, dsr, dsr->config.messageReceiveTimeout, NULL,
632
0
                         UA_TIMERPOLICY_CURRENTTIME, &dsr->msgRcvTimeoutTimerId);
633
0
        } else {
634
            /* Reset the next execution time to now + interval */
635
0
            el->modifyTimer(el, dsr->msgRcvTimeoutTimerId,
636
0
                            dsr->config.messageReceiveTimeout, NULL,
637
0
                            UA_TIMERPOLICY_CURRENTTIME);
638
0
        }
639
0
    }
640
641
    /* Received a heartbeat with no fields */
642
0
    if(msg->fieldCount == 0)
643
0
        return;
644
645
    /* Check whether the field count matches the configuration */
646
0
    UA_TargetVariablesDataType *tvs = &dsr->config.subscribedDataSet.target;
647
0
    if(tvs->targetVariablesSize != msg->fieldCount) {
648
0
        UA_LOG_WARNING_PUBSUB(psm->logging, dsr,
649
0
                              "Number of fields does not match the "
650
0
                              "TargetVariables configuration");
651
0
        return;
652
0
    }
653
654
    /* Write the message fields. RT has the external data value configured. */
655
0
    UA_StatusCode res = UA_STATUSCODE_GOOD;
656
0
    for(size_t i = 0; i < msg->fieldCount; i++) {
657
0
        UA_FieldTargetDataType *tv = &tvs->targetVariables[i];
658
0
        UA_DataValue *field = &msg->data.keyFrameFields[i];
659
0
        if(!field->hasValue)
660
0
            continue;
661
662
        /* Write via the Write-Service */
663
0
        UA_WriteValue writeVal;
664
0
        UA_WriteValue_init(&writeVal);
665
0
        writeVal.attributeId = tv->attributeId;
666
0
        writeVal.indexRange = tv->receiverIndexRange;
667
0
        writeVal.nodeId = tv->targetNodeId;
668
0
        writeVal.value = *field;
669
0
        Operation_Write(psm->sc.server, &psm->sc.server->adminSession, &writeVal, &res);
670
0
        if(res != UA_STATUSCODE_GOOD)
671
0
            UA_LOG_INFO_PUBSUB(psm->logging, dsr,
672
0
                               "Error writing KeyFrame field %u: %s",
673
0
                               (unsigned)i, UA_StatusCode_name(res));
674
0
    }
675
0
}
676
677
/**************/
678
/* Server API */
679
/**************/
680
681
UA_StatusCode
682
UA_Server_addDataSetReader(UA_Server *server, UA_NodeId readerGroupId,
683
                           const UA_DataSetReaderConfig *config,
684
0
                           UA_NodeId *dsrId) {
685
0
    if(!server || !config)
686
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
687
0
    lockServer(server);
688
0
    UA_StatusCode res =
689
0
        UA_DataSetReader_create(getPSM(server), readerGroupId, config, dsrId);
690
0
    unlockServer(server);
691
0
    return res;
692
0
}
693
694
UA_StatusCode
695
0
UA_Server_removeDataSetReader(UA_Server *server, const UA_NodeId readerId) {
696
0
    if(!server)
697
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
698
0
    lockServer(server);
699
0
    UA_PubSubManager *psm = getPSM(server);
700
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, readerId);
701
0
    UA_StatusCode res = (dsr) ?
702
0
        UA_DataSetReader_remove(psm, dsr) : UA_STATUSCODE_BADNOTFOUND;
703
0
    unlockServer(server);
704
0
    return res;
705
0
}
706
707
UA_StatusCode
708
UA_Server_getDataSetReaderConfig(UA_Server *server, const UA_NodeId dsrId,
709
0
                                 UA_DataSetReaderConfig *config) {
710
0
    if(!server || !config)
711
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
712
0
    lockServer(server);
713
0
    UA_PubSubManager *psm = getPSM(server);
714
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dsrId);
715
0
    UA_StatusCode res = (dsr) ?
716
0
        UA_DataSetReaderConfig_copy(&dsr->config, config) : UA_STATUSCODE_BADNOTFOUND;
717
0
    unlockServer(server);
718
0
    return res;
719
0
}
720
721
UA_StatusCode
722
UA_Server_getDataSetReaderState(UA_Server *server, const UA_NodeId dsrId,
723
0
                                UA_PubSubState *state) {
724
0
    if(!server || !state)
725
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
726
0
    lockServer(server);
727
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(getPSM(server), dsrId);
728
0
    UA_StatusCode res = UA_STATUSCODE_BADNOTFOUND;
729
0
    if(dsr) {
730
0
        res = UA_STATUSCODE_GOOD;
731
0
        *state = dsr->head.state;
732
0
    }
733
0
    unlockServer(server);
734
0
    return res;
735
0
}
736
737
UA_StatusCode
738
0
UA_Server_enableDataSetReader(UA_Server *server, const UA_NodeId dsrId) {
739
0
    if(!server)
740
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
741
0
    lockServer(server);
742
0
    UA_StatusCode ret = UA_STATUSCODE_GOOD;
743
0
    UA_PubSubManager *psm = getPSM(server);
744
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dsrId);
745
0
    if(dsr)
746
0
        UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_OPERATIONAL,
747
0
                                        UA_STATUSCODE_GOOD);
748
0
    else
749
0
        ret = UA_STATUSCODE_BADNOTFOUND;
750
0
    unlockServer(server);
751
0
    return ret;
752
0
}
753
754
UA_StatusCode
755
0
UA_Server_disableDataSetReader(UA_Server *server, const UA_NodeId dsrId) {
756
0
    if(!server)
757
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
758
0
    lockServer(server);
759
0
    UA_StatusCode ret = UA_STATUSCODE_GOOD;
760
0
    UA_PubSubManager *psm = getPSM(server);
761
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dsrId);
762
0
    if(dsr)
763
0
        UA_DataSetReader_setPubSubState(psm, dsr, UA_PUBSUBSTATE_DISABLED,
764
0
                                        UA_STATUSCODE_GOOD);
765
0
    else
766
0
        ret = UA_STATUSCODE_BADNOTFOUND;
767
0
    unlockServer(server);
768
0
    return ret;
769
0
}
770
771
UA_StatusCode
772
UA_Server_setDataSetReaderTargetVariables(UA_Server *server, const UA_NodeId dsrId,
773
                                          size_t targetVariablesSize,
774
0
                                          const UA_FieldTargetDataType *targetVariables) {
775
0
    if(!server)
776
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
777
0
    lockServer(server);
778
0
    UA_PubSubManager *psm = getPSM(server);
779
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dsrId);
780
0
    UA_StatusCode res = (dsr) ?
781
0
        DataSetReader_createTargetVariables(psm, dsr, targetVariablesSize,
782
0
                                            targetVariables) : UA_STATUSCODE_BADNOTFOUND;
783
0
    unlockServer(server);
784
0
    return res;
785
0
}
786
787
UA_StatusCode
788
UA_Server_updateDataSetReaderConfig(UA_Server *server, const UA_NodeId dsrId,
789
0
                                    const UA_DataSetReaderConfig *config) {
790
0
    if(!server || !config)
791
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
792
793
0
    lockServer(server);
794
0
    UA_PubSubManager *psm = getPSM(server);
795
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dsrId);
796
0
    if(!dsr) {
797
0
        unlockServer(server);
798
0
        return UA_STATUSCODE_BADNOTFOUND;
799
0
    }
800
801
0
    if(UA_PubSubState_isEnabled(dsr->head.state)) {
802
0
        UA_LOG_ERROR_PUBSUB(psm->logging, dsr,
803
0
                            "The DataSetReader must be disabled to update the config");
804
0
        unlockServer(server);
805
0
        return UA_STATUSCODE_BADINTERNALERROR;
806
0
    }
807
808
    /* Store the old config */
809
0
    UA_DataSetReaderConfig oldConfig = dsr->config;
810
811
    /* Copy the config into the new dataSetReader */
812
0
    UA_StatusCode retVal = UA_DataSetReaderConfig_copy(config, &dsr->config);
813
0
    if(retVal != UA_STATUSCODE_GOOD)
814
0
        goto errout;
815
816
    /* Change the connection to a StandaloneSubscribedDataSet */
817
0
    if(!UA_String_equal(&dsr->config.linkedStandaloneSubscribedDataSetName,
818
0
                        &oldConfig.linkedStandaloneSubscribedDataSetName)) {
819
0
        disconnectDSR2Standalone(psm, dsr);
820
0
        retVal = connectDSR2Standalone(psm, dsr);
821
0
        if(retVal != UA_STATUSCODE_GOOD)
822
0
            goto errout;
823
0
    }
824
825
    /* Validate the new config */
826
0
    retVal = validateDSRConfig(psm, dsr);
827
0
    if(retVal != UA_STATUSCODE_GOOD)
828
0
        goto errout;
829
830
    /* Clean up and return */
831
0
    UA_DataSetReaderConfig_clear(&oldConfig);
832
0
    unlockServer(server);
833
0
    return UA_STATUSCODE_GOOD;
834
835
    /* Fall back to the old config */
836
0
 errout:
837
0
    UA_DataSetReaderConfig_clear(&dsr->config);
838
0
    dsr->config = oldConfig;
839
0
    unlockServer(server);
840
0
    return retVal;
841
0
}
842
843
/**********************/
844
/* Offset Computation */
845
/**********************/
846
847
static UA_StatusCode
848
UA_PubSubDataSetReader_generateKeyFrameMessage(UA_Server *server,
849
                                               UA_DataSetMessage *dsm,
850
0
                                               UA_DataSetReader *dsr) {
851
    /* Prepare DataSetMessageContent */
852
0
    UA_TargetVariablesDataType *tv = &dsr->config.subscribedDataSet.target;
853
0
    UA_DataSetMetaDataType *metaData = &dsr->config.dataSetMetaData;
854
0
    if(tv->targetVariablesSize != metaData->fieldsSize)
855
0
        metaData = NULL;
856
0
    dsm->header.dataSetMessageValid = true;
857
0
    dsm->header.dataSetMessageType = UA_DATASETMESSAGE_DATAKEYFRAME;
858
0
    dsm->fieldCount = (UA_UInt16) tv->targetVariablesSize;
859
0
    dsm->data.keyFrameFields = (UA_DataValue *)
860
0
            UA_Array_new(tv->targetVariablesSize, &UA_TYPES[UA_TYPES_DATAVALUE]);
861
0
    if(!dsm->data.keyFrameFields)
862
0
        return UA_STATUSCODE_BADOUTOFMEMORY;
863
864
0
     for(size_t counter = 0; counter < tv->targetVariablesSize; counter++) {
865
        /* Read the value and set the source in the reader config */
866
0
        UA_DataValue *dfv = &dsm->data.keyFrameFields[counter];
867
0
        UA_FieldTargetDataType *ftv = &tv->targetVariables[counter];
868
869
        /* Synthesize the field value from the FieldMetaData. This allows us to
870
         * prevent a read from the information model during startup. */
871
0
        UA_FieldMetaData *fieldMetaData = (metaData) ? &metaData->fields[counter] : NULL;
872
0
        if(fieldMetaData && fieldMetaData->valueRank == UA_VALUERANK_SCALAR) {
873
0
            const UA_DataType *type =
874
0
                UA_findDataTypeWithCustom(&fieldMetaData->dataType,
875
0
                                          server->config.customDataTypes);
876
0
            if(type == &UA_TYPES[UA_TYPES_STRING] && fieldMetaData->maxStringLength > 0) {
877
0
                UA_String *s = UA_String_new();
878
0
                if(!s) {
879
0
                    UA_DataSetMessage_clear(dsm);
880
0
                    return UA_STATUSCODE_BADOUTOFMEMORY;
881
0
                }
882
0
                s->data = (UA_Byte*)
883
0
                    UA_calloc(fieldMetaData->maxStringLength, sizeof(UA_Byte));
884
0
                if(!s->data) {
885
0
                    UA_free(s);
886
0
                    UA_DataSetMessage_clear(dsm);
887
0
                    return UA_STATUSCODE_BADOUTOFMEMORY;
888
0
                }
889
0
                s->length = fieldMetaData->maxStringLength;
890
0
                UA_Variant_setScalar(&dfv->value, s, type);
891
0
                dfv->hasValue = true;
892
0
            } else if(type && type->memSize < 512) {
893
0
                char buf[512];
894
0
                UA_init(buf, type);
895
0
                UA_StatusCode res = UA_Variant_setScalarCopy(&dfv->value, buf, type);
896
0
                if(res != UA_STATUSCODE_GOOD) {
897
0
                    UA_DataSetMessage_clear(dsm);
898
0
                    return res;
899
0
                }
900
0
                dfv->hasValue = true;
901
0
            }
902
0
        }
903
904
        /* Read the value from the information model */
905
0
        if(!dfv->hasValue) {
906
0
            UA_ReadValueId rvi;
907
0
            UA_ReadValueId_init(&rvi);
908
0
            rvi.nodeId = ftv->targetNodeId;
909
0
            rvi.attributeId = ftv->attributeId;
910
0
            rvi.indexRange = ftv->writeIndexRange;
911
0
            *dfv = readWithSession(server, &server->adminSession, &rvi,
912
0
                                   UA_TIMESTAMPSTORETURN_NEITHER);
913
0
        }
914
915
        /* Deactivate statuscode? */
916
0
        if(((u64)dsr->config.dataSetFieldContentMask &
917
0
            (u64)UA_DATASETFIELDCONTENTMASK_STATUSCODE) == 0)
918
0
            dfv->hasStatus = false;
919
920
        /* Deactivate timestamps */
921
0
        if(((u64)dsr->config.dataSetFieldContentMask &
922
0
            (u64)UA_DATASETFIELDCONTENTMASK_SOURCETIMESTAMP) == 0)
923
0
            dfv->hasSourceTimestamp = false;
924
0
        if(((u64)dsr->config.dataSetFieldContentMask &
925
0
            (u64)UA_DATASETFIELDCONTENTMASK_SOURCEPICOSECONDS) == 0)
926
0
            dfv->hasSourcePicoseconds = false;
927
0
        if(((u64)dsr->config.dataSetFieldContentMask &
928
0
            (u64)UA_DATASETFIELDCONTENTMASK_SERVERTIMESTAMP) == 0)
929
0
            dfv->hasServerTimestamp = false;
930
0
        if(((u64)dsr->config.dataSetFieldContentMask &
931
0
            (u64)UA_DATASETFIELDCONTENTMASK_SERVERPICOSECONDS) == 0)
932
0
            dfv->hasServerPicoseconds = false;
933
0
    }
934
935
0
    return UA_STATUSCODE_GOOD;
936
0
}
937
938
/* Generate a DataSetMessage for the given reader. */
939
UA_StatusCode
940
UA_DataSetReader_generateDataSetMessage(UA_Server *server,
941
                                        UA_DataSetMessage *dsm,
942
0
                                        UA_DataSetReader *dsr) {
943
    /* Support only for UADP configuration
944
     * TODO: JSON encoding if UA_DataSetReader_generateDataSetMessage used other
945
     * that RT configuration */
946
947
0
    UA_ExtensionObject *settings = &dsr->config.messageSettings;
948
0
    if(settings->content.decoded.type != &UA_TYPES[UA_TYPES_UADPDATASETREADERMESSAGEDATATYPE])
949
0
        return UA_STATUSCODE_BADNOTSUPPORTED;
950
951
    /* The configuration Flags are included inside the std. defined
952
     * UA_UadpDataSetReaderMessageDataType */
953
0
    UA_UadpDataSetReaderMessageDataType defaultUadpConfiguration;
954
0
    UA_UadpDataSetReaderMessageDataType *dsrMessageDataType =
955
0
        (UA_UadpDataSetReaderMessageDataType*) settings->content.decoded.data;
956
957
0
    if(!(settings->encoding == UA_EXTENSIONOBJECT_DECODED ||
958
0
         settings->encoding == UA_EXTENSIONOBJECT_DECODED_NODELETE) ||
959
0
       !dsrMessageDataType->dataSetMessageContentMask) {
960
        /* Create default flag configuration if no dataSetMessageContentMask or
961
         * even messageSettings in UadpDataSetWriterMessageDataType was
962
         * passed. */
963
0
        memset(&defaultUadpConfiguration, 0, sizeof(UA_UadpDataSetReaderMessageDataType));
964
0
        defaultUadpConfiguration.dataSetMessageContentMask = (UA_UadpDataSetMessageContentMask)
965
0
            ((u64)UA_UADPDATASETMESSAGECONTENTMASK_TIMESTAMP |
966
0
             (u64)UA_UADPDATASETMESSAGECONTENTMASK_MAJORVERSION |
967
0
             (u64)UA_UADPDATASETMESSAGECONTENTMASK_MINORVERSION);
968
0
        dsrMessageDataType = &defaultUadpConfiguration;
969
0
    }
970
971
    /* The field encoding depends on the flags inside the reader config. */
972
0
    if(dsr->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_RAWDATA) {
973
0
        dsm->header.fieldEncoding = UA_FIELDENCODING_RAWDATA;
974
0
    } else if((u64)dsr->config.dataSetFieldContentMask &
975
0
              ((u64)UA_DATASETFIELDCONTENTMASK_SOURCETIMESTAMP |
976
0
               (u64)UA_DATASETFIELDCONTENTMASK_SERVERPICOSECONDS |
977
0
               (u64)UA_DATASETFIELDCONTENTMASK_SOURCEPICOSECONDS |
978
0
               (u64)UA_DATASETFIELDCONTENTMASK_STATUSCODE)) {
979
0
        dsm->header.fieldEncoding = UA_FIELDENCODING_DATAVALUE;
980
0
    } else {
981
0
        dsm->header.fieldEncoding = UA_FIELDENCODING_VARIANT;
982
0
    }
983
984
    /* Std: 'The DataSetMessageContentMask defines the flags for the content
985
     * of the DataSetMessage header.' */
986
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
987
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_MAJORVERSION) {
988
0
        dsm->header.configVersionMajorVersionEnabled = true;
989
0
        dsm->header.configVersionMajorVersion =
990
0
            dsr->config.dataSetMetaData.configurationVersion.majorVersion;
991
0
    }
992
993
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
994
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_MINORVERSION) {
995
0
        dsm->header.configVersionMinorVersionEnabled = true;
996
0
        dsm->header.configVersionMinorVersion =
997
0
            dsr->config.dataSetMetaData.configurationVersion.minorVersion;
998
0
    }
999
1000
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
1001
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_SEQUENCENUMBER) {
1002
        /* Will be modified when subscriber receives new nw msg */
1003
0
        dsm->header.dataSetMessageSequenceNrEnabled = true;
1004
0
        dsm->header.dataSetMessageSequenceNr = 1;
1005
0
    }
1006
1007
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
1008
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_TIMESTAMP) {
1009
0
        dsm->header.timestampEnabled = true;
1010
0
        dsm->header.timestamp = UA_DateTime_now();
1011
0
    }
1012
1013
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
1014
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_PICOSECONDS)
1015
0
        dsm->header.picoSecondsIncluded = false;
1016
1017
0
    if((u64)dsrMessageDataType->dataSetMessageContentMask &
1018
0
       (u64)UA_UADPDATASETMESSAGECONTENTMASK_STATUS)
1019
0
        dsm->header.statusEnabled = true;
1020
1021
    /* Not supported for Delta frames atm */
1022
0
    return UA_PubSubDataSetReader_generateKeyFrameMessage(server, dsm, dsr);
1023
0
}
1024
1025
1026
UA_StatusCode
1027
UA_Server_computeDataSetReaderOffsetTable(UA_Server *server,
1028
                                          const UA_NodeId dataSetReaderId,
1029
0
                                          UA_PubSubOffsetTable *ot) {
1030
    /* Validate the arguments */
1031
0
    if(!server || !ot)
1032
0
        return UA_STATUSCODE_BADINVALIDARGUMENT;
1033
1034
0
    lockServer(server);
1035
1036
    /* Get the DataSetReader */
1037
0
    UA_PubSubManager *psm = getPSM(server);
1038
0
    UA_DataSetReader *dsr = UA_DataSetReader_find(psm, dataSetReaderId);
1039
0
    if(!dsr) {
1040
0
        unlockServer(server);
1041
0
        return UA_STATUSCODE_BADNOTFOUND;
1042
0
    }
1043
1044
    /* Generate the DataSetMessage */
1045
0
    UA_DataSetMessage dsm;
1046
0
    memset(&dsm, 0, sizeof(UA_DataSetMessage));
1047
0
    UA_StatusCode res = UA_DataSetReader_generateDataSetMessage(server, &dsm, dsr);
1048
0
    if(res != UA_STATUSCODE_GOOD) {
1049
0
        unlockServer(server);
1050
0
        return res;
1051
0
    }
1052
1053
    /* Reset the OffsetTable */
1054
0
    memset(ot, 0, sizeof(UA_PubSubOffsetTable));
1055
1056
    /* Prepare the encoding context */
1057
0
    UA_DataSetMessage_EncodingMetaData emd;
1058
0
    memset(&emd, 0, sizeof(UA_DataSetMessage_EncodingMetaData));
1059
0
    emd.dataSetWriterId = dsr->config.dataSetWriterId;
1060
0
    emd.fields = dsr->config.dataSetMetaData.fields;
1061
0
    emd.fieldsSize = dsr->config.dataSetMetaData.fieldsSize;
1062
1063
0
    PubSubEncodeCtx ctx;
1064
0
    memset(&ctx, 0, sizeof(PubSubEncodeCtx));
1065
0
    ctx.ot = ot;
1066
0
    ctx.eo.metaData = &emd;
1067
0
    ctx.eo.metaDataSize = 1;
1068
1069
    /* Compute the offset */
1070
0
    size_t fieldindex = 0;
1071
0
    UA_FieldTargetDataType *tv = NULL;
1072
0
    size_t msgSize = UA_DataSetMessage_calcSizeBinary(&ctx, &emd, &dsm, 0);
1073
0
    if(msgSize == 0) {
1074
0
        res = UA_STATUSCODE_BADINTERNALERROR;
1075
0
        goto errout;
1076
0
    }
1077
1078
    /* Allocate the message */
1079
0
    res = UA_ByteString_allocBuffer(&ot->networkMessage, msgSize);
1080
0
    if(res != UA_STATUSCODE_GOOD)
1081
0
        goto errout;
1082
1083
    /* Create the ByteString of the encoded DataSetMessage */
1084
0
    ctx.ctx.pos = ot->networkMessage.data;
1085
0
    ctx.ctx.end = ot->networkMessage.data + ot->networkMessage.length;
1086
0
    res = UA_DataSetMessage_encodeBinary(&ctx, &emd, &dsm);
1087
0
    if(res != UA_STATUSCODE_GOOD)
1088
0
        goto errout;
1089
1090
    /* Pick up the component NodeIds */
1091
0
    for(size_t i = 0; i < ot->offsetsSize; i++) {
1092
0
        UA_PubSubOffset *o = &ot->offsets[i];
1093
0
        switch(o->offsetType) {
1094
0
        case UA_PUBSUBOFFSETTYPE_DATASETMESSAGE_SEQUENCENUMBER:
1095
0
        case UA_PUBSUBOFFSETTYPE_DATASETMESSAGE_STATUS:
1096
0
        case UA_PUBSUBOFFSETTYPE_DATASETMESSAGE_TIMESTAMP:
1097
0
        case UA_PUBSUBOFFSETTYPE_DATASETMESSAGE_PICOSECONDS:
1098
0
            res |= UA_NodeId_copy(&dsr->head.identifier, &o->component);
1099
0
            break;
1100
0
        case UA_PUBSUBOFFSETTYPE_DATASETFIELD_DATAVALUE:
1101
0
            tv = &dsr->config.subscribedDataSet.target.targetVariables[fieldindex];
1102
0
            res |= UA_NodeId_copy(&tv->targetNodeId, &o->component);
1103
0
            fieldindex++;
1104
0
            break;
1105
0
        case UA_PUBSUBOFFSETTYPE_DATASETFIELD_VARIANT:
1106
0
            tv = &dsr->config.subscribedDataSet.target.targetVariables[fieldindex];
1107
0
            res |= UA_NodeId_copy(&tv->targetNodeId, &o->component);
1108
0
            fieldindex++;
1109
0
            break;
1110
0
        case UA_PUBSUBOFFSETTYPE_DATASETFIELD_RAW:
1111
0
            tv = &dsr->config.subscribedDataSet.target.targetVariables[fieldindex];
1112
0
            res |= UA_NodeId_copy(&tv->targetNodeId, &o->component);
1113
0
            fieldindex++;
1114
0
            break;
1115
0
        default:
1116
0
            res = UA_STATUSCODE_BADINTERNALERROR;
1117
0
            break;
1118
0
        }
1119
0
    }
1120
1121
    /* Clean up */
1122
0
 errout:
1123
0
    UA_DataSetMessage_clear(&dsm);
1124
0
    if(res != UA_STATUSCODE_GOOD)
1125
0
        UA_PubSubOffsetTable_clear(ot);
1126
0
    unlockServer(server);
1127
0
    return res;
1128
0
}
1129
1130
#endif /* UA_ENABLE_PUBSUB */