12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426 |
- /* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at http://mozilla.org/MPL/2.0/.
- *
- * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
- * Copyright (c) 2019 Fraunhofer IOSB (Author: Julius Pfrommer)
- * Copyright (c) 2019 Kalycito Infotech Private Limited
- */
- #include <open62541/server_config_default.h>
- #include "server/ua_server_internal.h"
- #ifdef UA_ENABLE_PUBSUB /* conditional compilation */
- #include "ua_pubsub.h"
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- #include "ua_pubsub_ns0.h"
- #endif
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- #include "ua_types_encoding_binary.h"
- #endif
- #define UA_MAX_STACKBUF 512 /* Max size of network messages on the stack */
- #define UA_MAX_SIZENAME 64 /* Max size of Qualified Name of Subscribed Variable */
- /* Forward declaration */
- static size_t
- iterateVariableAttrDimension(UA_VariableAttributes varAttr);
- static void
- UA_WriterGroup_deleteMembers(UA_Server *server, UA_WriterGroup *writerGroup);
- static void
- UA_DataSetField_deleteMembers(UA_DataSetField *field);
- /**********************************************/
- /* Connection */
- /**********************************************/
- UA_StatusCode
- UA_PubSubConnectionConfig_copy(const UA_PubSubConnectionConfig *src,
- UA_PubSubConnectionConfig *dst) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- memcpy(dst, src, sizeof(UA_PubSubConnectionConfig));
- retVal |= UA_String_copy(&src->name, &dst->name);
- retVal |= UA_Variant_copy(&src->address, &dst->address);
- retVal |= UA_String_copy(&src->transportProfileUri, &dst->transportProfileUri);
- retVal |= UA_Variant_copy(&src->connectionTransportSettings, &dst->connectionTransportSettings);
- if(src->connectionPropertiesSize > 0){
- dst->connectionProperties = (UA_KeyValuePair *)
- UA_calloc(src->connectionPropertiesSize, sizeof(UA_KeyValuePair));
- if(!dst->connectionProperties){
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- for(size_t i = 0; i < src->connectionPropertiesSize; i++){
- retVal |= UA_QualifiedName_copy(&src->connectionProperties[i].key,
- &dst->connectionProperties[i].key);
- retVal |= UA_Variant_copy(&src->connectionProperties[i].value,
- &dst->connectionProperties[i].value);
- }
- }
- return retVal;
- }
- UA_StatusCode
- UA_Server_getPubSubConnectionConfig(UA_Server *server, const UA_NodeId connection,
- UA_PubSubConnectionConfig *config) {
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_PubSubConnection *currentPubSubConnection =
- UA_PubSubConnection_findConnectionbyId(server, connection);
- if(!currentPubSubConnection)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_PubSubConnectionConfig tmpPubSubConnectionConfig;
- //deep copy of the actual config
- UA_PubSubConnectionConfig_copy(currentPubSubConnection->config, &tmpPubSubConnectionConfig);
- *config = tmpPubSubConnectionConfig;
- return UA_STATUSCODE_GOOD;
- }
- UA_PubSubConnection *
- UA_PubSubConnection_findConnectionbyId(UA_Server *server, UA_NodeId connectionIdentifier) {
- for(size_t i = 0; i < server->pubSubManager.connectionsSize; i++){
- if(UA_NodeId_equal(&connectionIdentifier, &server->pubSubManager.connections[i].identifier)){
- return &server->pubSubManager.connections[i];
- }
- }
- return NULL;
- }
- void
- UA_PubSubConnectionConfig_deleteMembers(UA_PubSubConnectionConfig *connectionConfig) {
- UA_String_deleteMembers(&connectionConfig->name);
- UA_String_deleteMembers(&connectionConfig->transportProfileUri);
- UA_Variant_deleteMembers(&connectionConfig->connectionTransportSettings);
- UA_Variant_deleteMembers(&connectionConfig->address);
- for(size_t i = 0; i < connectionConfig->connectionPropertiesSize; i++){
- UA_QualifiedName_deleteMembers(&connectionConfig->connectionProperties[i].key);
- UA_Variant_deleteMembers(&connectionConfig->connectionProperties[i].value);
- }
- UA_free(connectionConfig->connectionProperties);
- }
- void
- UA_PubSubConnection_deleteMembers(UA_Server *server, UA_PubSubConnection *connection) {
- //delete connection config
- UA_PubSubConnectionConfig_deleteMembers(connection->config);
- //remove contained WriterGroups
- UA_WriterGroup *writerGroup, *tmpWriterGroup;
- LIST_FOREACH_SAFE(writerGroup, &connection->writerGroups, listEntry, tmpWriterGroup){
- UA_Server_removeWriterGroup(server, writerGroup->identifier);
- }
- /* remove contained ReaderGroups */
- UA_ReaderGroup *readerGroups, *tmpReaderGroup;
- LIST_FOREACH_SAFE(readerGroups, &connection->readerGroups, listEntry, tmpReaderGroup){
- UA_Server_removeReaderGroup(server, readerGroups->identifier);
- }
- UA_NodeId_deleteMembers(&connection->identifier);
- if(connection->channel){
- connection->channel->close(connection->channel);
- }
- UA_free(connection->config);
- }
- UA_StatusCode
- UA_Server_addWriterGroup(UA_Server *server, const UA_NodeId connection,
- const UA_WriterGroupConfig *writerGroupConfig,
- UA_NodeId *writerGroupIdentifier) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- if(!writerGroupConfig)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- //search the connection by the given connectionIdentifier
- UA_PubSubConnection *currentConnectionContext =
- UA_PubSubConnection_findConnectionbyId(server, connection);
- if(!currentConnectionContext)
- return UA_STATUSCODE_BADNOTFOUND;
- //allocate memory for new WriterGroup
- UA_WriterGroup *newWriterGroup = (UA_WriterGroup *) UA_calloc(1, sizeof(UA_WriterGroup));
- if(!newWriterGroup)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- newWriterGroup->linkedConnection = currentConnectionContext->identifier;
- UA_PubSubManager_generateUniqueNodeId(server, &newWriterGroup->identifier);
- if(writerGroupIdentifier){
- UA_NodeId_copy(&newWriterGroup->identifier, writerGroupIdentifier);
- }
- //deep copy of the config
- UA_WriterGroupConfig tmpWriterGroupConfig;
- retVal |= UA_WriterGroupConfig_copy(writerGroupConfig, &tmpWriterGroupConfig);
- if(!tmpWriterGroupConfig.messageSettings.content.decoded.type) {
- UA_UadpWriterGroupMessageDataType *wgm = UA_UadpWriterGroupMessageDataType_new();
- tmpWriterGroupConfig.messageSettings.content.decoded.data = wgm;
- tmpWriterGroupConfig.messageSettings.content.decoded.type =
- &UA_TYPES[UA_TYPES_UADPWRITERGROUPMESSAGEDATATYPE];
- tmpWriterGroupConfig.messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED;
- }
- newWriterGroup->config = tmpWriterGroupConfig;
- retVal |= UA_WriterGroup_addPublishCallback(server, newWriterGroup);
- LIST_INSERT_HEAD(¤tConnectionContext->writerGroups, newWriterGroup, listEntry);
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addWriterGroupRepresentation(server, newWriterGroup);
- #endif
- return retVal;
- }
- UA_StatusCode
- UA_Server_removeWriterGroup(UA_Server *server, const UA_NodeId writerGroup){
- UA_WriterGroup *wg = UA_WriterGroup_findWGbyId(server, writerGroup);
- if(!wg)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_PubSubConnection *connection =
- UA_PubSubConnection_findConnectionbyId(server, wg->linkedConnection);
- if(!connection)
- return UA_STATUSCODE_BADNOTFOUND;
- //unregister the publish callback
- UA_PubSubManager_removeRepeatedPubSubCallback(server, wg->publishCallbackId);
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- removeGroupRepresentation(server, wg);
- #endif
- UA_WriterGroup_deleteMembers(server, wg);
- LIST_REMOVE(wg, listEntry);
- UA_free(wg);
- return UA_STATUSCODE_GOOD;
- }
- /**********************************************/
- /* ReaderGroup */
- /**********************************************/
- /**
- * Add ReaderGroup to connection.
- *
- * @param server
- * @param connectionIdentifier
- * @param readerGroupConfiguration
- * @param readerGroupIdentifier
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_addReaderGroup(UA_Server *server, UA_NodeId connectionIdentifier,
- const UA_ReaderGroupConfig *readerGroupConfig,
- UA_NodeId *readerGroupIdentifier) {
- UA_StatusCode retval = UA_STATUSCODE_GOOD;
- UA_ReaderGroupConfig tmpReaderGroupConfig;
- /* Search the connection by the given connectionIdentifier */
- if(!readerGroupConfig) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- /* Search the connection by the given connectionIdentifier */
- UA_PubSubConnection *currentConnectionContext = UA_PubSubConnection_findConnectionbyId(server, connectionIdentifier);
- if(!currentConnectionContext) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- /* Allocate memory for new reader group */
- UA_ReaderGroup *newGroup = (UA_ReaderGroup *)UA_calloc(1, sizeof(UA_ReaderGroup));
- if(!newGroup) {
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- /* Generate nodeid for the readergroup identifier */
- newGroup->linkedConnection = currentConnectionContext->identifier;
- UA_PubSubManager_generateUniqueNodeId(server, &newGroup->identifier);
- if(readerGroupIdentifier) {
- UA_NodeId_copy(&newGroup->identifier, readerGroupIdentifier);
- }
- /* Deep copy of the config */
- retval |= UA_ReaderGroupConfig_copy(readerGroupConfig, &tmpReaderGroupConfig);
- newGroup->config = tmpReaderGroupConfig;
- retval |= UA_ReaderGroup_addSubscribeCallback(server, newGroup);
- LIST_INSERT_HEAD(¤tConnectionContext->readerGroups, newGroup, listEntry);
- currentConnectionContext->readerGroupsSize++;
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addReaderGroupRepresentation(server, newGroup);
- #endif
- return retval;
- }
- /**
- * Remove ReaderGroup from connection and delete contained readers.
- *
- * @param server
- * @param groupIdentifier
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_removeReaderGroup(UA_Server *server, UA_NodeId groupIdentifier) {
- UA_ReaderGroup* readerGroup = UA_ReaderGroup_findRGbyId(server, groupIdentifier);
- if(readerGroup == NULL) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- /* Search the connection to which the given readergroup is connected to */
- UA_PubSubConnection *connection = UA_PubSubConnection_findConnectionbyId(server, readerGroup->linkedConnection);
- if(connection == NULL) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- /* To Do:RemoveGroupRepresentation(server, &readerGroup->identifier) */
- #endif
- /* UA_Server_ReaderGroup_delete also removes itself from the list */
- UA_Server_ReaderGroup_delete(server, readerGroup);
- /* Remove readerGroup from Connection */
- LIST_REMOVE(readerGroup, listEntry);
- UA_free(readerGroup);
- return UA_STATUSCODE_GOOD;
- }
- /**
- * To Do:
- * Update ReaderGroup configuration.
- *
- * @param server
- * @param readerGroupIdentifier
- * @param readerGroupConfiguration
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_ReaderGroup_updateConfig(UA_Server *server, UA_NodeId readerGroupIdentifier,
- const UA_ReaderGroupConfig *config) {
- return UA_STATUSCODE_BADNOTIMPLEMENTED;
- }
- /**
- * Get ReaderGroup configuration.
- *
- * @param server
- * @param groupIdentifier
- * @param readerGroupConfiguration
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_ReaderGroup_getConfig(UA_Server *server, UA_NodeId readerGroupIdentifier,
- UA_ReaderGroupConfig *config) {
- if(!config) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- /* Identify the readergroup through the readerGroupIdentifier */
- UA_ReaderGroup *currentReaderGroup = UA_ReaderGroup_findRGbyId(server, readerGroupIdentifier);
- if(!currentReaderGroup) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- UA_ReaderGroupConfig tmpReaderGroupConfig;
- /* deep copy of the actual config */
- UA_ReaderGroupConfig_copy(¤tReaderGroup->config, &tmpReaderGroupConfig);
- *config = tmpReaderGroupConfig;
- return UA_STATUSCODE_GOOD;
- }
- /* To Do UA_ReaderGroupConfig delete */
- /**
- * Delete ReaderGroup.
- *
- * @param server
- * @param groupIdentifier
- */
- void UA_Server_ReaderGroup_delete(UA_Server* server, UA_ReaderGroup *readerGroup) {
- /* To Do Call UA_ReaderGroupConfig_delete */
- UA_DataSetReader *dataSetReader, *tmpDataSetReader;
- LIST_FOREACH_SAFE(dataSetReader, &readerGroup->readers, listEntry, tmpDataSetReader) {
- UA_DataSetReader_delete(server, dataSetReader);
- }
- UA_PubSubConnection* pConn = UA_PubSubConnection_findConnectionbyId(server, readerGroup->linkedConnection);
- if(pConn != NULL) {
- pConn->readerGroupsSize--;
- }
- /* Delete ReaderGroup and its members */
- UA_String_deleteMembers(&readerGroup->config.name);
- UA_NodeId_deleteMembers(&readerGroup->linkedConnection);
- UA_NodeId_deleteMembers(&readerGroup->identifier);
- }
- /**
- * Copy ReaderGroup configuration.
- *
- * @param source
- * @param destination
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_ReaderGroupConfig_copy(const UA_ReaderGroupConfig *src,
- UA_ReaderGroupConfig *dst) {
- UA_String_copy(&src->name, &dst->name);
- /* Currently simple memcpy only */
- memcpy(&dst->securityParameters, &src->securityParameters, sizeof(UA_PubSubSecurityParameters));
- return UA_STATUSCODE_GOOD;
- }
- static UA_DataSetReader *
- getReaderFromIdentifier(UA_Server *server, UA_NetworkMessage *pMsg, UA_PubSubConnection *pConnection) {
- if(pConnection->readerGroupsSize == 1) {
- if(LIST_FIRST(&pConnection->readerGroups)->readersCount == 1) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "only 1 DataSetReader available. This one will be used.");
- return LIST_FIRST(&LIST_FIRST(&pConnection->readerGroups)->readers);
- }
- }
- if(!pMsg->publisherIdEnabled)
- return NULL;
- UA_ReaderGroup* readerGroup;
- LIST_FOREACH(readerGroup, &pConnection->readerGroups, listEntry) {
- UA_DataSetReader *tmpReader;
- LIST_FOREACH(tmpReader, &readerGroup->readers, listEntry) {
- switch (pMsg->publisherIdType) {
- case UA_PUBLISHERDATATYPE_BYTE:
- if(tmpReader->config.publisherId.type == &UA_TYPES[UA_TYPES_BYTE] &&
- pMsg->publisherIdType == UA_PUBLISHERDATATYPE_BYTE &&
- pMsg->publisherId.publisherIdByte == *(UA_Byte*)tmpReader->config.publisherId.data) {
- return tmpReader;
- }
- break;
- case UA_PUBLISHERDATATYPE_UINT16:
- if(tmpReader->config.publisherId.type == &UA_TYPES[UA_TYPES_UINT16] &&
- pMsg->publisherIdType == UA_PUBLISHERDATATYPE_UINT16 &&
- pMsg->publisherId.publisherIdUInt16 == *(UA_UInt16*)tmpReader->config.publisherId.data) {
- return tmpReader;
- }
- break;
- case UA_PUBLISHERDATATYPE_UINT32:
- if(tmpReader->config.publisherId.type == &UA_TYPES[UA_TYPES_UINT32] &&
- pMsg->publisherIdType == UA_PUBLISHERDATATYPE_UINT32 &&
- pMsg->publisherId.publisherIdUInt32 == *(UA_UInt32*)tmpReader->config.publisherId.data) {
- return tmpReader;
- }
- break;
- case UA_PUBLISHERDATATYPE_UINT64:
- if(tmpReader->config.publisherId.type == &UA_TYPES[UA_TYPES_UINT64] &&
- pMsg->publisherIdType == UA_PUBLISHERDATATYPE_UINT64 &&
- pMsg->publisherId.publisherIdUInt64 == *(UA_UInt64*)tmpReader->config.publisherId.data) {
- return tmpReader;
- }
- break;
- case UA_PUBLISHERDATATYPE_STRING:
- if(tmpReader->config.publisherId.type == &UA_TYPES[UA_TYPES_STRING] &&
- pMsg->publisherIdType == UA_PUBLISHERDATATYPE_STRING &&
- UA_String_equal(&pMsg->publisherId.publisherIdString, (UA_String*)tmpReader->config.publisherId.data)) {
- return tmpReader;
- }
- break;
- default:
- return NULL;
- }
- }
- }
- return NULL;
- }
- /**
- * Process NetworkMessage.
- *
- * @param server
- * @param networkmessage
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_processNetworkMessage(UA_Server *server, UA_NetworkMessage *pMsg,
- UA_PubSubConnection *pConnection) {
- if(!pMsg || !pConnection)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- /* To Do The condition with dataSetWriterIdAvailable and WriterGroupIdAvailable to be handled
- * when pMsg->groupHeaderEnabled, pMsg->dataSetClassIdEnabled, pMsg->payloadHeaderEnabled
- * Here some filtering is possible */
- UA_DataSetReader* dataSetReaderErg = getReaderFromIdentifier(server, pMsg, pConnection);
- /* No Reader with the specified id found */
- if(!dataSetReaderErg) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "No DataSetReader found with PublisherId");
- return UA_STATUSCODE_BADNOTFOUND; /* TODO: Check the return code */
- }
- UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER, "DataSetReader found with PublisherId");
- UA_Byte anzDataSets = 1;
- if(pMsg->payloadHeaderEnabled)
- anzDataSets = pMsg->payloadHeader.dataSetPayloadHeader.count;
- for(UA_Byte iterator = 0; iterator < anzDataSets; iterator++) {
- UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER, "Process Msg with DataSetReader!");
- UA_Server_DataSetReader_process(server, dataSetReaderErg, &pMsg->payload.dataSetPayload.dataSetMessages[iterator]);
- }
- /* To Do the condition with dataSetWriterId and WriterGroupId
- * else condition for dataSetWriterIdAvailable and writerGroupIdAvailable) */
- return UA_STATUSCODE_GOOD;
- }
- /**
- * Find size iterator - array dimension for VariableAttributes
- *
- * @param VariableAttributes
- * @return global variable size iterator
- */
- static size_t
- iterateVariableAttrDimension (UA_VariableAttributes varAttr) {
- size_t sizeIterator = 0;
- for(size_t indexIterator = 0; indexIterator < varAttr.arrayDimensionsSize; indexIterator++) {
- sizeIterator += varAttr.arrayDimensions[indexIterator];
- }
- return sizeIterator;
- }
- /**
- * Find ReaderGroup with its identifier.
- *
- * @param server
- * @param groupIdentifier
- * @return the ReaderGroup or NULL if not found
- */
- UA_ReaderGroup * UA_ReaderGroup_findRGbyId(UA_Server *server, UA_NodeId identifier) {
- for (size_t iteratorConn = 0; iteratorConn < server->pubSubManager.connectionsSize; iteratorConn++) {
- UA_ReaderGroup* readerGroup = NULL;
- LIST_FOREACH(readerGroup, &server->pubSubManager.connections[iteratorConn].readerGroups, listEntry) {
- if(UA_NodeId_equal(&identifier, &readerGroup->identifier)) {
- return readerGroup;
- }
- }
- }
- return NULL;
- }
- /**
- * Find a DataSetReader with its identifier
- *
- * @param server
- * @param identifier
- * @return the DataSetReader or NULL if not found
- */
- UA_DataSetReader *UA_ReaderGroup_findDSRbyId(UA_Server *server, UA_NodeId identifier) {
- for (size_t iteratorConn = 0; iteratorConn < server->pubSubManager.connectionsSize; iteratorConn++) {
- UA_ReaderGroup* readerGroup = NULL;
- LIST_FOREACH(readerGroup, &server->pubSubManager.connections[iteratorConn].readerGroups, listEntry) {
- UA_DataSetReader *tmpReader;
- LIST_FOREACH(tmpReader, &readerGroup->readers, listEntry) {
- if(UA_NodeId_equal(&tmpReader->identifier, &identifier)) {
- return tmpReader;
- }
- }
- }
- }
- return NULL;
- }
- /**********************************************/
- /* DataSetReader */
- /**********************************************/
- /**
- * Add a DataSetReader to ReaderGroup
- *
- * @param server
- * @param readerGroupIdentifier
- * @param dataSetReaderConfig
- * @param readerIdentifier
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_addDataSetReader(UA_Server *server, UA_NodeId readerGroupIdentifier,
- const UA_DataSetReaderConfig *dataSetReaderConfig,
- UA_NodeId *readerIdentifier) {
- /* Search the reader group by the given readerGroupIdentifier */
- UA_ReaderGroup *readerGroup = UA_ReaderGroup_findRGbyId(server, readerGroupIdentifier);
- if(!dataSetReaderConfig) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- if(readerGroup == NULL) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- /* Allocate memory for new DataSetReader */
- UA_DataSetReader *newDataSetReader = (UA_DataSetReader *)UA_calloc(1, sizeof(UA_DataSetReader));
- /* Copy the config into the new dataSetReader */
- UA_DataSetReaderConfig_copy(dataSetReaderConfig, &newDataSetReader->config);
- newDataSetReader->linkedReaderGroup = readerGroup->identifier;
- UA_PubSubManager_generateUniqueNodeId(server, &newDataSetReader->identifier);
- if(readerIdentifier != NULL) {
- UA_NodeId_copy(&newDataSetReader->identifier, readerIdentifier);
- }
- /* Add the new reader to the group */
- LIST_INSERT_HEAD(&readerGroup->readers, newDataSetReader, listEntry);
- readerGroup->readersCount++;
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addDataSetReaderRepresentation(server, newDataSetReader);
- #endif
- return UA_STATUSCODE_GOOD;
- }
- /**
- * Remove a DataSetReader from ReaderGroup
- *
- * @param server
- * @param readerGroupIdentifier
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_removeDataSetReader(UA_Server *server, UA_NodeId readerIdentifier) {
- /* Remove datasetreader given by the identifier */
- UA_DataSetReader *dataSetReader = UA_ReaderGroup_findDSRbyId(server, readerIdentifier);
- if(!dataSetReader) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- removeDataSetReaderRepresentation(server, dataSetReader);
- #endif
- UA_DataSetReader_delete(server, dataSetReader);
- return UA_STATUSCODE_GOOD;
- }
- /**
- * Update the config of the DataSetReader.
- *
- * @param server
- * @param dataSetReaderIdentifier
- * @param readerGroupIdentifier
- * @param config
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_DataSetReader_updateConfig(UA_Server *server, UA_NodeId dataSetReaderIdentifier, UA_NodeId readerGroupIdentifier,
- const UA_DataSetReaderConfig *config) {
- if(config == NULL) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- UA_DataSetReader *currentDataSetReader = UA_ReaderGroup_findDSRbyId(server, dataSetReaderIdentifier);
- UA_ReaderGroup *currentReaderGroup = UA_ReaderGroup_findRGbyId(server, readerGroupIdentifier);
- if(!currentDataSetReader) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- /* The update functionality will be extended during the next PubSub batches.
- * Currently is only a change of the publishing interval possible. */
- if(currentDataSetReader->config.writerGroupId != config->writerGroupId) {
- UA_PubSubManager_removeRepeatedPubSubCallback(server, currentReaderGroup->subscribeCallbackId);
- currentDataSetReader->config.writerGroupId = config->writerGroupId;
- UA_ReaderGroup_subscribeCallback(server, currentReaderGroup);
- }
- else {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "No or unsupported ReaderGroup update.");
- }
- return UA_STATUSCODE_GOOD;
- }
- /**
- * Get the current config of the UA_DataSetReader.
- *
- * @param server
- * @param dataSetReaderIdentifier
- * @param config
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_DataSetReader_getConfig(UA_Server *server, UA_NodeId dataSetReaderIdentifier,
- UA_DataSetReaderConfig *config) {
- if(!config) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- UA_DataSetReader *currentDataSetReader = UA_ReaderGroup_findDSRbyId(server, dataSetReaderIdentifier);
- if(!currentDataSetReader) {
- return UA_STATUSCODE_BADNOTFOUND;
- }
- UA_DataSetReaderConfig tmpReaderConfig;
- /* Deep copy of the actual config */
- UA_DataSetReaderConfig_copy(¤tDataSetReader->config, &tmpReaderConfig);
- *config = tmpReaderConfig;
- return UA_STATUSCODE_GOOD;
- }
- /**
- * This Method is used to initially set the SubscribedDataSet to TargetVariablesType and to create the list of target Variables of a SubscribedDataSetType.
- *
- * @param server
- * @param dataSetReaderIdentifier
- * @param targetVariables
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_Server_DataSetReader_createTargetVariables(UA_Server *server, UA_NodeId dataSetReaderIdentifier, UA_TargetVariablesDataType *targetVariables) {
- UA_StatusCode retval = UA_STATUSCODE_BADUNEXPECTEDERROR;
- UA_DataSetReader* pDS = UA_ReaderGroup_findDSRbyId(server, dataSetReaderIdentifier);
- if(pDS == NULL) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- if(pDS->subscribedDataSetTarget.targetVariablesSize > 0) {
- UA_TargetVariablesDataType_deleteMembers(&pDS->subscribedDataSetTarget);
- pDS->subscribedDataSetTarget.targetVariablesSize = 0;
- pDS->subscribedDataSetTarget.targetVariables = NULL;
- }
- /* Set subscribed dataset to TargetVariableType */
- pDS->subscribedDataSetType = UA_PUBSUB_SDS_TARGET;
- retval = UA_TargetVariablesDataType_copy(targetVariables, &pDS->subscribedDataSetTarget);
- return retval;
- }
- /**
- * Adds Subscribed Variables from the DataSetMetaData for the given DataSet into the given parent node
- * and creates the corresponding data in the targetVariables of the DataSetReader
- *
- * @param server
- * @param parentNode
- * @param dataSetReaderIdentifier
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode UA_Server_DataSetReader_addTargetVariables(UA_Server *server, UA_NodeId *parentNode, UA_NodeId dataSetReaderIdentifier, UA_SubscribedDataSetEnumType sdsType) {
- if((server == NULL) || (parentNode == NULL)) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- UA_StatusCode retval = UA_STATUSCODE_GOOD;
- UA_DataSetReader* pDataSetReader = UA_ReaderGroup_findDSRbyId(server, dataSetReaderIdentifier);
- if(pDataSetReader == NULL) {
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- /* To Do Declare for all other type */
- UA_DateTime dateTimeTypeVal = 0;
- UA_TargetVariablesDataType targetVars;
- targetVars.targetVariablesSize = pDataSetReader->config.dataSetMetaData.fieldsSize;
- targetVars.targetVariables = (UA_FieldTargetDataType *)UA_calloc(targetVars.targetVariablesSize, sizeof(UA_FieldTargetDataType));
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- for (size_t iteratorField = 0; iteratorField < pDataSetReader->config.dataSetMetaData.fieldsSize; iteratorField++) {
- UA_VariableAttributes vAttr = UA_VariableAttributes_default;
- vAttr.valueRank = pDataSetReader->config.dataSetMetaData.fields[iteratorField].valueRank;
- if(pDataSetReader->config.dataSetMetaData.fields[iteratorField].arrayDimensionsSize > 0) {
- retVal = UA_Array_copy(pDataSetReader->config.dataSetMetaData.fields[iteratorField].arrayDimensions, pDataSetReader->config.dataSetMetaData.fields[iteratorField].arrayDimensionsSize, (void**)&vAttr.arrayDimensions, &UA_TYPES[UA_TYPES_UINT32]);
- if(retVal == UA_STATUSCODE_GOOD) {
- vAttr.arrayDimensionsSize = pDataSetReader->config.dataSetMetaData.fields[iteratorField].arrayDimensionsSize;
- }
- }
- switch (pDataSetReader->config.dataSetMetaData.fields[iteratorField].builtInType) {
- case UA_NS0ID_BOOLEAN:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Boolean *boolArray = (UA_Boolean*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_BOOLEAN]);
- UA_Variant_setArrayCopy(&vAttr.value, boolArray, sizeIterator, &UA_TYPES[UA_TYPES_BOOLEAN]);
- UA_Array_delete(boolArray, sizeIterator, &UA_TYPES[UA_TYPES_BOOLEAN]);
- }
- else {
- UA_Boolean boolTypeVal = false;
- UA_Variant_setScalar(&vAttr.value, &boolTypeVal, &UA_TYPES[UA_TYPES_BOOLEAN]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_SBYTE:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_SByte *sbyteArray = (UA_SByte*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_SBYTE]);
- UA_Variant_setArrayCopy(&vAttr.value, sbyteArray, sizeIterator, &UA_TYPES[UA_TYPES_SBYTE]);
- UA_Array_delete(sbyteArray, sizeIterator, &UA_TYPES[UA_TYPES_SBYTE]);
- }
- else {
- UA_SByte sbyteTypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &sbyteTypeVal, &UA_TYPES[UA_TYPES_SBYTE]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_BYTE:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Byte *byteArray = (UA_Byte*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_BYTE]);
- UA_Variant_setArrayCopy(&vAttr.value, byteArray, sizeIterator, &UA_TYPES[UA_TYPES_BYTE]);
- UA_Array_delete(byteArray, sizeIterator, &UA_TYPES[UA_TYPES_BYTE]);
- }
- else {
- UA_Byte byteTypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &byteTypeVal, &UA_TYPES[UA_TYPES_BYTE]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_INT16:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Int16 *int16Array = (UA_Int16*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_INT16]);
- UA_Variant_setArrayCopy(&vAttr.value, int16Array, sizeIterator, &UA_TYPES[UA_TYPES_INT16]);
- UA_Array_delete(int16Array, sizeIterator, &UA_TYPES[UA_TYPES_INT16]);
- }
- else {
- UA_Int16 int16TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &int16TypeVal, &UA_TYPES[UA_TYPES_INT16]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_UINT16:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_UInt16 *uint16Array = (UA_UInt16*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_UINT16]);
- UA_Variant_setArrayCopy(&vAttr.value, uint16Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT16]);
- UA_Array_delete(uint16Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT16]);
- }
- else {
- UA_UInt16 uint16TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &uint16TypeVal, &UA_TYPES[UA_TYPES_UINT16]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_INT32:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Int32 *int32Array = (UA_Int32*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_INT32]);
- UA_Variant_setArrayCopy(&vAttr.value, int32Array, sizeIterator, &UA_TYPES[UA_TYPES_INT32]);
- UA_Array_delete(int32Array, sizeIterator, &UA_TYPES[UA_TYPES_INT32]);
- }
- else {
- UA_Int32 int32TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &int32TypeVal, &UA_TYPES[UA_TYPES_INT32]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_UINT32:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_UInt32 *uint32Array = (UA_UInt32*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_UINT32]);
- UA_Variant_setArrayCopy(&vAttr.value, uint32Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT32]);
- UA_Array_delete(uint32Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT32]);
- }
- else {
- UA_UInt32 uint32TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &uint32TypeVal, &UA_TYPES[UA_TYPES_UINT32]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_INT64:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Int64 *int64Array = (UA_Int64*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_INT64]);
- UA_Variant_setArrayCopy(&vAttr.value, int64Array, sizeIterator, &UA_TYPES[UA_TYPES_INT64]);
- UA_Array_delete(int64Array, sizeIterator, &UA_TYPES[UA_TYPES_INT64]);
- }
- else {
- UA_Int64 int64TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &int64TypeVal, &UA_TYPES[UA_TYPES_INT64]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_UINT64:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_UInt64 *uint64Array = (UA_UInt64*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_UINT64]);
- UA_Variant_setArrayCopy(&vAttr.value, uint64Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT64]);
- UA_Array_delete(uint64Array, sizeIterator, &UA_TYPES[UA_TYPES_UINT64]);
- }
- else {
- UA_UInt64 uint64TypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &uint64TypeVal, &UA_TYPES[UA_TYPES_UINT64]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_FLOAT:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Float *floatArray = (UA_Float*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_FLOAT]);
- UA_Variant_setArrayCopy(&vAttr.value, floatArray, sizeIterator, &UA_TYPES[UA_TYPES_FLOAT]);
- UA_Array_delete(floatArray, sizeIterator, &UA_TYPES[UA_TYPES_FLOAT]);
- }
- else {
- UA_Float floatTypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &floatTypeVal, &UA_TYPES[UA_TYPES_FLOAT]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_DOUBLE:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Double *doubleArray = (UA_Double*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_DOUBLE]);
- UA_Variant_setArrayCopy(&vAttr.value, doubleArray, sizeIterator, &UA_TYPES[UA_TYPES_DOUBLE]);
- UA_Array_delete(doubleArray, sizeIterator, &UA_TYPES[UA_TYPES_DOUBLE]);
- }
- else {
- UA_Double doubleTypeVal = 0.0;
- UA_Variant_setScalar(&vAttr.value, &doubleTypeVal, &UA_TYPES[UA_TYPES_DOUBLE]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_STRING:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_String *stringArray = (UA_String*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_STRING]);
- UA_Variant_setArrayCopy(&vAttr.value, stringArray, sizeIterator, &UA_TYPES[UA_TYPES_STRING]);
- UA_Array_delete(stringArray, sizeIterator, &UA_TYPES[UA_TYPES_STRING]);
- }
- else {
- UA_String stringTypeVal = UA_STRING_NULL;
- UA_Variant_setScalar(&vAttr.value, &stringTypeVal, &UA_TYPES[UA_TYPES_STRING]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_DATETIME:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_DateTime *dateTimeArray = (UA_DateTime*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_DATETIME]);
- UA_Variant_setArrayCopy(&vAttr.value, dateTimeArray, sizeIterator, &UA_TYPES[UA_TYPES_DATETIME]);
- UA_Array_delete(dateTimeArray, sizeIterator, &UA_TYPES[UA_TYPES_DATETIME]);
- }
- else {
- dateTimeTypeVal = 0;
- UA_Variant_setScalar(&vAttr.value, &dateTimeTypeVal, &UA_TYPES[UA_TYPES_DATETIME]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_GUID:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_Guid *guidArray = (UA_Guid*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_GUID]);
- UA_Variant_setArrayCopy(&vAttr.value, guidArray, sizeIterator, &UA_TYPES[UA_TYPES_GUID]);
- UA_Array_delete(guidArray, sizeIterator, &UA_TYPES[UA_TYPES_GUID]);
- }
- else {
- UA_Guid guidTypeVal = UA_GUID_NULL;
- UA_Variant_setScalar(&vAttr.value, &guidTypeVal, &UA_TYPES[UA_TYPES_GUID]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_NODEID:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_NodeId *nodeidArray = (UA_NodeId*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_NODEID]);
- UA_Variant_setArrayCopy(&vAttr.value, nodeidArray, sizeIterator, &UA_TYPES[UA_TYPES_NODEID]);
- UA_Array_delete(nodeidArray, sizeIterator, &UA_TYPES[UA_TYPES_NODEID]);
- }
- else {
- UA_NodeId nodeidTypeVal = UA_NODEID_NULL;
- UA_Variant_setScalar(&vAttr.value, &nodeidTypeVal, &UA_TYPES[UA_TYPES_NODEID]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- case UA_NS0ID_BYTESTRING:
- if(vAttr.arrayDimensionsSize > 0) {
- size_t sizeIterator = iterateVariableAttrDimension(vAttr);
- UA_ByteString *byteStringArray = (UA_ByteString*)UA_Array_new(sizeIterator, &UA_TYPES[UA_TYPES_BYTESTRING]);
- UA_Variant_setArrayCopy(&vAttr.value, byteStringArray, sizeIterator, &UA_TYPES[UA_TYPES_BYTESTRING]);
- UA_Array_delete(byteStringArray, sizeIterator, &UA_TYPES[UA_TYPES_BYTESTRING]);
- }
- else {
- UA_ByteString byteStringTypeVal = UA_BYTESTRING_NULL;
- UA_Variant_setScalar(&vAttr.value, &byteStringTypeVal, &UA_TYPES[UA_TYPES_BYTESTRING]);
- }
- UA_NodeId_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].dataType, &vAttr.dataType);
- break;
- default:
- /* Type not supported */
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "Type %u not supported, create empty variable.", pDataSetReader->config.dataSetMetaData.fields[iteratorField].builtInType);
- break;
- }
- vAttr.accessLevel = UA_ACCESSLEVELMASK_READ;
- UA_LocalizedText_copy(&pDataSetReader->config.dataSetMetaData.fields[iteratorField].description, &vAttr.description);
- UA_QualifiedName qn;
- UA_QualifiedName_init(&qn);
- char szTmpName[UA_MAX_SIZENAME];
- if(pDataSetReader->config.dataSetMetaData.fields[iteratorField].name.length > 0) {
- UA_UInt16 slen = UA_MAX_SIZENAME -1;
- vAttr.displayName.locale = UA_STRING("en-US");
- vAttr.displayName.text = pDataSetReader->config.dataSetMetaData.fields[iteratorField].name;
- if(pDataSetReader->config.dataSetMetaData.fields[iteratorField].name.length < slen) {
- slen = (UA_UInt16)pDataSetReader->config.dataSetMetaData.fields[iteratorField].name.length;
- UA_snprintf(szTmpName, sizeof(szTmpName), "%s", (const char*)pDataSetReader->config.dataSetMetaData.fields[iteratorField].name.data);
- }
- szTmpName[slen] = '\0';
- qn = UA_QUALIFIEDNAME(1, szTmpName);
- }
- else {
- strcpy(szTmpName, "SubscribedVariable");
- vAttr.displayName = UA_LOCALIZEDTEXT("en-US", szTmpName);
- qn = UA_QUALIFIEDNAME(1, "SubscribedVariable");
- }
- /* Add variable to the given parent node */
- UA_NodeId newNode;
- retVal = UA_Server_addVariableNode(server, UA_NODEID_NULL, *parentNode,
- UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), qn,
- UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), vAttr, NULL, &newNode);
- if(retVal == UA_STATUSCODE_GOOD) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "addVariableNode %s succeeded", szTmpName);
- }
- else {
- retval = retVal;
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_USERLAND, "addVariableNode: error 0x%x", retVal);
- }
- UA_FieldTargetDataType_init(&targetVars.targetVariables[iteratorField]);
- targetVars.targetVariables[iteratorField].attributeId = UA_ATTRIBUTEID_VALUE;
- UA_NodeId_copy(&newNode, &targetVars.targetVariables[iteratorField].targetNodeId);
- UA_NodeId_deleteMembers(&newNode);
- if(vAttr.arrayDimensionsSize > 0) {
- UA_Variant_deleteMembers(&vAttr.value);
- UA_Array_delete(vAttr.arrayDimensions, vAttr.arrayDimensionsSize, &UA_TYPES[UA_TYPES_UINT32]);
- }
- }
- if(sdsType == UA_PUBSUB_SDS_TARGET) {
- retval = UA_Server_DataSetReader_createTargetVariables(server, pDataSetReader->identifier, &targetVars);
- }
- UA_TargetVariablesDataType_deleteMembers(&targetVars);
- return retval;
- }
- /**
- * Process a NetworkMessage with a DataSetReader.
- *
- * @param server
- * @param dataSetReader
- * @param dataSetMsg
- */
- void UA_Server_DataSetReader_process(UA_Server *server, UA_DataSetReader *dataSetReader, UA_DataSetMessage* dataSetMsg) {
- if((dataSetReader == NULL) || (dataSetMsg == NULL) || (server == NULL)) {
- return;
- }
- if(!dataSetMsg->header.dataSetMessageValid) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "DataSetMessage is discarded: message is not valid");
- /* To Do check ConfigurationVersion*/
- /*if(dataSetMsg->header.configVersionMajorVersionEnabled)
- * {
- * if(dataSetMsg->header.configVersionMajorVersion != dataSetReader->config.dataSetMetaData.configurationVersion.majorVersion)
- * {
- * UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_SERVER, "DataSetMessage is discarded: ConfigurationVersion MajorVersion does not match");
- * return;
- * }
- } */
- }
- else {
- if(dataSetMsg->header.dataSetMessageType == UA_DATASETMESSAGE_DATAKEYFRAME) {
- if(dataSetMsg->header.fieldEncoding != UA_FIELDENCODING_RAWDATA) {
- size_t anzFields = dataSetMsg->data.keyFrameData.fieldCount;
- if(dataSetReader->config.dataSetMetaData.fieldsSize < anzFields) {
- anzFields = dataSetReader->config.dataSetMetaData.fieldsSize;
- }
- if(dataSetReader->subscribedDataSetTarget.targetVariablesSize < anzFields) {
- anzFields = dataSetReader->subscribedDataSetTarget.targetVariablesSize;
- }
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- for (UA_UInt16 iteratorField = 0; iteratorField < anzFields; iteratorField++) {
- if(dataSetMsg->data.keyFrameData.dataSetFields[iteratorField].hasValue) {
- if(dataSetReader->subscribedDataSetTarget.targetVariables[iteratorField].attributeId == UA_ATTRIBUTEID_VALUE) {
- retVal = UA_Server_writeValue(server, dataSetReader->subscribedDataSetTarget.targetVariables[iteratorField].targetNodeId, dataSetMsg->data.keyFrameData.dataSetFields[iteratorField].value);
- if(retVal != UA_STATUSCODE_GOOD) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "Error Write Value KF %u: 0x%x", iteratorField, retVal);
- }
- }
- else {
- UA_WriteValue writeVal;
- UA_WriteValue_init(&writeVal);
- writeVal.attributeId = dataSetReader->subscribedDataSetTarget.targetVariables[iteratorField].attributeId;
- writeVal.indexRange = dataSetReader->subscribedDataSetTarget.targetVariables[iteratorField].receiverIndexRange;
- writeVal.nodeId = dataSetReader->subscribedDataSetTarget.targetVariables[iteratorField].targetNodeId;
- UA_DataValue_copy(&dataSetMsg->data.keyFrameData.dataSetFields[iteratorField], &writeVal.value);
- retVal = UA_Server_write(server, &writeVal);
- if(retVal != UA_STATUSCODE_GOOD) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "Error Write KF %u: 0x%x", iteratorField, retVal);
- }
- }
- }
- }
- }
- }
- }
- }
- /**
- * Copy the config of the DataSetReader.
- *
- * @param src
- * @param dst
- * @return UA_STATUSCODE_GOOD on success
- */
- UA_StatusCode
- UA_DataSetReaderConfig_copy(const UA_DataSetReaderConfig *src,
- UA_DataSetReaderConfig *dst) {
- memset(dst, 0, sizeof(UA_DataSetReaderConfig));
- UA_StatusCode retVal = UA_String_copy(&src->name, &dst->name);
- if(retVal != UA_STATUSCODE_GOOD) {
- return retVal;
- }
- retVal = UA_Variant_copy(&src->publisherId, &dst->publisherId);
- if(retVal != UA_STATUSCODE_GOOD) {
- return retVal;
- }
- dst->writerGroupId = src->writerGroupId;
- dst->dataSetWriterId = src->dataSetWriterId;
- retVal = UA_DataSetMetaDataType_copy(&src->dataSetMetaData, &dst->dataSetMetaData);
- if(retVal != UA_STATUSCODE_GOOD) {
- return retVal;
- }
- dst->dataSetFieldContentMask = src->dataSetFieldContentMask;
- dst->messageReceiveTimeout = src->messageReceiveTimeout;
- /* Currently memcpy is used to copy the securityParameters */
- memcpy(&dst->securityParameters, &src->securityParameters, sizeof(UA_PubSubSecurityParameters));
- retVal = UA_UadpDataSetReaderMessageDataType_copy(&src->messageSettings, &dst->messageSettings);
- if(retVal != UA_STATUSCODE_GOOD) {
- return retVal;
- }
- return UA_STATUSCODE_GOOD;
- }
- /**
- * Delete the DataSetReader.
- *
- * @param server
- * @param dataSetReader
- */
- void UA_DataSetReader_delete(UA_Server *server, UA_DataSetReader *dataSetReader) {
- /* Delete DataSetReader config */
- UA_String_deleteMembers(&dataSetReader->config.name);
- UA_Variant_deleteMembers(&dataSetReader->config.publisherId);
- UA_DataSetMetaDataType_deleteMembers(&dataSetReader->config.dataSetMetaData);
- UA_UadpDataSetReaderMessageDataType_deleteMembers(&dataSetReader->config.messageSettings);
- UA_TargetVariablesDataType_deleteMembers(&dataSetReader->subscribedDataSetTarget);
- /* Delete DataSetReader */
- UA_ReaderGroup* pGroup = UA_ReaderGroup_findRGbyId(server, dataSetReader->linkedReaderGroup);
- if(pGroup != NULL) {
- pGroup->readersCount--;
- }
- UA_NodeId_deleteMembers(&dataSetReader->identifier);
- UA_NodeId_deleteMembers(&dataSetReader->linkedReaderGroup);
- /* Remove DataSetReader from group */
- LIST_REMOVE(dataSetReader, listEntry);
- /* Free memory allocated for DataSetReader */
- UA_free(dataSetReader);
- }
- /**********************************************/
- /* PublishedDataSet */
- /**********************************************/
- UA_StatusCode
- UA_PublishedDataSetConfig_copy(const UA_PublishedDataSetConfig *src,
- UA_PublishedDataSetConfig *dst) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- memcpy(dst, src, sizeof(UA_PublishedDataSetConfig));
- retVal |= UA_String_copy(&src->name, &dst->name);
- switch(src->publishedDataSetType){
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS:
- //no additional items
- break;
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE:
- if(src->config.itemsTemplate.variablesToAddSize > 0){
- dst->config.itemsTemplate.variablesToAdd = (UA_PublishedVariableDataType *) UA_calloc(
- src->config.itemsTemplate.variablesToAddSize, sizeof(UA_PublishedVariableDataType));
- }
- for(size_t i = 0; i < src->config.itemsTemplate.variablesToAddSize; i++){
- retVal |= UA_PublishedVariableDataType_copy(&src->config.itemsTemplate.variablesToAdd[i],
- &dst->config.itemsTemplate.variablesToAdd[i]);
- }
- retVal |= UA_DataSetMetaDataType_copy(&src->config.itemsTemplate.metaData,
- &dst->config.itemsTemplate.metaData);
- break;
- default:
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- }
- return retVal;
- }
- UA_StatusCode
- UA_Server_getPublishedDataSetConfig(UA_Server *server, const UA_NodeId pds,
- UA_PublishedDataSetConfig *config){
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_PublishedDataSet *currentPublishedDataSet = UA_PublishedDataSet_findPDSbyId(server, pds);
- if(!currentPublishedDataSet)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_PublishedDataSetConfig tmpPublishedDataSetConfig;
- //deep copy of the actual config
- UA_PublishedDataSetConfig_copy(¤tPublishedDataSet->config, &tmpPublishedDataSetConfig);
- *config = tmpPublishedDataSetConfig;
- return UA_STATUSCODE_GOOD;
- }
- UA_PublishedDataSet *
- UA_PublishedDataSet_findPDSbyId(UA_Server *server, UA_NodeId identifier){
- for(size_t i = 0; i < server->pubSubManager.publishedDataSetsSize; i++){
- if(UA_NodeId_equal(&server->pubSubManager.publishedDataSets[i].identifier, &identifier)){
- return &server->pubSubManager.publishedDataSets[i];
- }
- }
- return NULL;
- }
- void
- UA_PublishedDataSetConfig_deleteMembers(UA_PublishedDataSetConfig *pdsConfig){
- //delete pds config
- UA_String_deleteMembers(&pdsConfig->name);
- switch (pdsConfig->publishedDataSetType){
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS:
- //no additional items
- break;
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE:
- if(pdsConfig->config.itemsTemplate.variablesToAddSize > 0){
- for(size_t i = 0; i < pdsConfig->config.itemsTemplate.variablesToAddSize; i++){
- UA_PublishedVariableDataType_deleteMembers(&pdsConfig->config.itemsTemplate.variablesToAdd[i]);
- }
- UA_free(pdsConfig->config.itemsTemplate.variablesToAdd);
- }
- UA_DataSetMetaDataType_deleteMembers(&pdsConfig->config.itemsTemplate.metaData);
- break;
- default:
- break;
- }
- }
- void
- UA_PublishedDataSet_deleteMembers(UA_Server *server, UA_PublishedDataSet *publishedDataSet){
- UA_PublishedDataSetConfig_deleteMembers(&publishedDataSet->config);
- //delete PDS
- UA_DataSetMetaDataType_deleteMembers(&publishedDataSet->dataSetMetaData);
- UA_DataSetField *field, *tmpField;
- LIST_FOREACH_SAFE(field, &publishedDataSet->fields, listEntry, tmpField) {
- UA_Server_removeDataSetField(server, field->identifier);
- }
- UA_NodeId_deleteMembers(&publishedDataSet->identifier);
- }
- UA_DataSetFieldResult
- UA_Server_addDataSetField(UA_Server *server, const UA_NodeId publishedDataSet,
- const UA_DataSetFieldConfig *fieldConfig,
- UA_NodeId *fieldIdentifier) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- UA_DataSetFieldResult result = {UA_STATUSCODE_BADINVALIDARGUMENT, {0, 0}};
- if(!fieldConfig)
- return result;
- UA_PublishedDataSet *currentDataSet = UA_PublishedDataSet_findPDSbyId(server, publishedDataSet);
- if(currentDataSet == NULL){
- result.result = UA_STATUSCODE_BADNOTFOUND;
- return result;
- }
- if(currentDataSet->config.publishedDataSetType != UA_PUBSUB_DATASET_PUBLISHEDITEMS){
- result.result = UA_STATUSCODE_BADNOTIMPLEMENTED;
- return result;
- }
- UA_DataSetField *newField = (UA_DataSetField *) UA_calloc(1, sizeof(UA_DataSetField));
- if(!newField){
- result.result = UA_STATUSCODE_BADINTERNALERROR;
- return result;
- }
- UA_DataSetFieldConfig tmpFieldConfig;
- retVal |= UA_DataSetFieldConfig_copy(fieldConfig, &tmpFieldConfig);
- newField->config = tmpFieldConfig;
- UA_PubSubManager_generateUniqueNodeId(server, &newField->identifier);
- if(fieldIdentifier != NULL){
- UA_NodeId_copy(&newField->identifier, fieldIdentifier);
- }
- newField->publishedDataSet = currentDataSet->identifier;
- //update major version of parent published data set
- currentDataSet->dataSetMetaData.configurationVersion.majorVersion = UA_PubSubConfigurationVersionTimeDifference();
- LIST_INSERT_HEAD(¤tDataSet->fields, newField, listEntry);
- if(newField->config.field.variable.promotedField)
- currentDataSet->promotedFieldsCount++;
- currentDataSet->fieldSize++;
- result.result = retVal;
- result.configurationVersion.majorVersion = currentDataSet->dataSetMetaData.configurationVersion.majorVersion;
- result.configurationVersion.minorVersion = currentDataSet->dataSetMetaData.configurationVersion.minorVersion;
- return result;
- }
- UA_DataSetFieldResult
- UA_Server_removeDataSetField(UA_Server *server, const UA_NodeId dsf) {
- UA_DataSetField *currentField = UA_DataSetField_findDSFbyId(server, dsf);
- UA_DataSetFieldResult result = {UA_STATUSCODE_BADNOTFOUND, {0, 0}};
- if(!currentField)
- return result;
- UA_PublishedDataSet *parentPublishedDataSet =
- UA_PublishedDataSet_findPDSbyId(server, currentField->publishedDataSet);
- if(!parentPublishedDataSet)
- return result;
- parentPublishedDataSet->fieldSize--;
- if(currentField->config.field.variable.promotedField)
- parentPublishedDataSet->promotedFieldsCount--;
- /* update major version of PublishedDataSet */
- parentPublishedDataSet->dataSetMetaData.configurationVersion.majorVersion =
- UA_PubSubConfigurationVersionTimeDifference();
- UA_DataSetField_deleteMembers(currentField);
- LIST_REMOVE(currentField, listEntry);
- UA_free(currentField);
- result.result = UA_STATUSCODE_GOOD;
- result.configurationVersion.majorVersion = parentPublishedDataSet->dataSetMetaData.configurationVersion.majorVersion;
- result.configurationVersion.minorVersion = parentPublishedDataSet->dataSetMetaData.configurationVersion.minorVersion;
- return result;
- }
- /**********************************************/
- /* DataSetWriter */
- /**********************************************/
- UA_StatusCode
- UA_DataSetWriterConfig_copy(const UA_DataSetWriterConfig *src,
- UA_DataSetWriterConfig *dst){
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- memcpy(dst, src, sizeof(UA_DataSetWriterConfig));
- retVal |= UA_String_copy(&src->name, &dst->name);
- retVal |= UA_String_copy(&src->dataSetName, &dst->dataSetName);
- retVal |= UA_ExtensionObject_copy(&src->messageSettings, &dst->messageSettings);
- dst->dataSetWriterProperties = (UA_KeyValuePair *)
- UA_calloc(src->dataSetWriterPropertiesSize, sizeof(UA_KeyValuePair));
- if(!dst->dataSetWriterProperties)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- for(size_t i = 0; i < src->dataSetWriterPropertiesSize; i++){
- retVal |= UA_KeyValuePair_copy(&src->dataSetWriterProperties[i], &dst->dataSetWriterProperties[i]);
- }
- return retVal;
- }
- UA_StatusCode
- UA_Server_getDataSetWriterConfig(UA_Server *server, const UA_NodeId dsw,
- UA_DataSetWriterConfig *config){
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_DataSetWriter *currentDataSetWriter = UA_DataSetWriter_findDSWbyId(server, dsw);
- if(!currentDataSetWriter)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_DataSetWriterConfig tmpWriterConfig;
- //deep copy of the actual config
- retVal |= UA_DataSetWriterConfig_copy(¤tDataSetWriter->config, &tmpWriterConfig);
- *config = tmpWriterConfig;
- return retVal;
- }
- UA_DataSetWriter *
- UA_DataSetWriter_findDSWbyId(UA_Server *server, UA_NodeId identifier) {
- for(size_t i = 0; i < server->pubSubManager.connectionsSize; i++){
- UA_WriterGroup *tmpWriterGroup;
- LIST_FOREACH(tmpWriterGroup, &server->pubSubManager.connections[i].writerGroups, listEntry){
- UA_DataSetWriter *tmpWriter;
- LIST_FOREACH(tmpWriter, &tmpWriterGroup->writers, listEntry){
- if(UA_NodeId_equal(&tmpWriter->identifier, &identifier)){
- return tmpWriter;
- }
- }
- }
- }
- return NULL;
- }
- void
- UA_DataSetWriterConfig_deleteMembers(UA_DataSetWriterConfig *pdsConfig) {
- UA_String_deleteMembers(&pdsConfig->name);
- UA_String_deleteMembers(&pdsConfig->dataSetName);
- for(size_t i = 0; i < pdsConfig->dataSetWriterPropertiesSize; i++){
- UA_KeyValuePair_deleteMembers(&pdsConfig->dataSetWriterProperties[i]);
- }
- UA_free(pdsConfig->dataSetWriterProperties);
- UA_ExtensionObject_deleteMembers(&pdsConfig->messageSettings);
- }
- static void
- UA_DataSetWriter_deleteMembers(UA_Server *server, UA_DataSetWriter *dataSetWriter) {
- UA_DataSetWriterConfig_deleteMembers(&dataSetWriter->config);
- //delete DataSetWriter
- UA_NodeId_deleteMembers(&dataSetWriter->identifier);
- UA_NodeId_deleteMembers(&dataSetWriter->linkedWriterGroup);
- UA_NodeId_deleteMembers(&dataSetWriter->connectedDataSet);
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- //delete lastSamples store
- for(size_t i = 0; i < dataSetWriter->lastSamplesCount; i++) {
- UA_DataValue_deleteMembers(&dataSetWriter->lastSamples[i].value);
- }
- UA_free(dataSetWriter->lastSamples);
- dataSetWriter->lastSamples = NULL;
- dataSetWriter->lastSamplesCount = 0;
- #endif
- }
- /**********************************************/
- /* WriterGroup */
- /**********************************************/
- UA_StatusCode
- UA_WriterGroupConfig_copy(const UA_WriterGroupConfig *src,
- UA_WriterGroupConfig *dst){
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- memcpy(dst, src, sizeof(UA_WriterGroupConfig));
- retVal |= UA_String_copy(&src->name, &dst->name);
- retVal |= UA_ExtensionObject_copy(&src->transportSettings, &dst->transportSettings);
- retVal |= UA_ExtensionObject_copy(&src->messageSettings, &dst->messageSettings);
- dst->groupProperties = (UA_KeyValuePair *) UA_calloc(src->groupPropertiesSize, sizeof(UA_KeyValuePair));
- if(!dst->groupProperties)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- for(size_t i = 0; i < src->groupPropertiesSize; i++){
- retVal |= UA_KeyValuePair_copy(&src->groupProperties[i], &dst->groupProperties[i]);
- }
- return retVal;
- }
- UA_StatusCode
- UA_Server_getWriterGroupConfig(UA_Server *server, const UA_NodeId writerGroup,
- UA_WriterGroupConfig *config){
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_WriterGroup *currentWriterGroup = UA_WriterGroup_findWGbyId(server, writerGroup);
- if(!currentWriterGroup){
- return UA_STATUSCODE_BADNOTFOUND;
- }
- UA_WriterGroupConfig tmpWriterGroupConfig;
- //deep copy of the actual config
- retVal |= UA_WriterGroupConfig_copy(¤tWriterGroup->config, &tmpWriterGroupConfig);
- *config = tmpWriterGroupConfig;
- return retVal;
- }
- UA_StatusCode
- UA_Server_updateWriterGroupConfig(UA_Server *server, UA_NodeId writerGroupIdentifier,
- const UA_WriterGroupConfig *config){
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_WriterGroup *currentWriterGroup = UA_WriterGroup_findWGbyId(server, writerGroupIdentifier);
- if(!currentWriterGroup)
- return UA_STATUSCODE_BADNOTFOUND;
- //The update functionality will be extended during the next PubSub batches.
- //Currently is only a change of the publishing interval possible.
- if(currentWriterGroup->config.publishingInterval != config->publishingInterval) {
- UA_PubSubManager_removeRepeatedPubSubCallback(server, currentWriterGroup->publishCallbackId);
- currentWriterGroup->config.publishingInterval = config->publishingInterval;
- UA_WriterGroup_addPublishCallback(server, currentWriterGroup);
- } else if(currentWriterGroup->config.priority != config->priority) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "No or unsupported WriterGroup update.");
- }
- return UA_STATUSCODE_GOOD;
- }
- UA_WriterGroup *
- UA_WriterGroup_findWGbyId(UA_Server *server, UA_NodeId identifier){
- for(size_t i = 0; i < server->pubSubManager.connectionsSize; i++){
- UA_WriterGroup *tmpWriterGroup;
- LIST_FOREACH(tmpWriterGroup, &server->pubSubManager.connections[i].writerGroups, listEntry) {
- if(UA_NodeId_equal(&identifier, &tmpWriterGroup->identifier)){
- return tmpWriterGroup;
- }
- }
- }
- return NULL;
- }
- void
- UA_WriterGroupConfig_deleteMembers(UA_WriterGroupConfig *writerGroupConfig){
- //delete writerGroup config
- UA_String_deleteMembers(&writerGroupConfig->name);
- UA_ExtensionObject_deleteMembers(&writerGroupConfig->transportSettings);
- UA_ExtensionObject_deleteMembers(&writerGroupConfig->messageSettings);
- for(size_t i = 0; i < writerGroupConfig->groupPropertiesSize; i++){
- UA_KeyValuePair_deleteMembers(&writerGroupConfig->groupProperties[i]);
- }
- UA_free(writerGroupConfig->groupProperties);
- }
- static void
- UA_WriterGroup_deleteMembers(UA_Server *server, UA_WriterGroup *writerGroup) {
- UA_WriterGroupConfig_deleteMembers(&writerGroup->config);
- //delete WriterGroup
- //delete all writers. Therefore removeDataSetWriter is called from PublishedDataSet
- UA_DataSetWriter *dataSetWriter, *tmpDataSetWriter;
- LIST_FOREACH_SAFE(dataSetWriter, &writerGroup->writers, listEntry, tmpDataSetWriter){
- UA_Server_removeDataSetWriter(server, dataSetWriter->identifier);
- }
- UA_NodeId_deleteMembers(&writerGroup->linkedConnection);
- UA_NodeId_deleteMembers(&writerGroup->identifier);
- }
- UA_StatusCode
- UA_Server_addDataSetWriter(UA_Server *server,
- const UA_NodeId writerGroup, const UA_NodeId dataSet,
- const UA_DataSetWriterConfig *dataSetWriterConfig,
- UA_NodeId *writerIdentifier) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- if(!dataSetWriterConfig)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_PublishedDataSet *currentDataSetContext = UA_PublishedDataSet_findPDSbyId(server, dataSet);
- if(!currentDataSetContext)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_WriterGroup *wg = UA_WriterGroup_findWGbyId(server, writerGroup);
- if(!wg)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_DataSetWriter *newDataSetWriter = (UA_DataSetWriter *) UA_calloc(1, sizeof(UA_DataSetWriter));
- if(!newDataSetWriter)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- //copy the config into the new dataSetWriter
- UA_DataSetWriterConfig tmpDataSetWriterConfig;
- retVal |= UA_DataSetWriterConfig_copy(dataSetWriterConfig, &tmpDataSetWriterConfig);
- newDataSetWriter->config = tmpDataSetWriterConfig;
- //save the current version of the connected PublishedDataSet
- newDataSetWriter->connectedDataSetVersion = currentDataSetContext->dataSetMetaData.configurationVersion;
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- //initialize the queue for the last values
- newDataSetWriter->lastSamples = (UA_DataSetWriterSample * )
- UA_calloc(currentDataSetContext->fieldSize, sizeof(UA_DataSetWriterSample));
- if(!newDataSetWriter->lastSamples) {
- UA_DataSetWriterConfig_deleteMembers(&newDataSetWriter->config);
- UA_free(newDataSetWriter);
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- newDataSetWriter->lastSamplesCount = currentDataSetContext->fieldSize;
- #endif
- //connect PublishedDataSet with DataSetWriter
- newDataSetWriter->connectedDataSet = currentDataSetContext->identifier;
- newDataSetWriter->linkedWriterGroup = wg->identifier;
- UA_PubSubManager_generateUniqueNodeId(server, &newDataSetWriter->identifier);
- if(writerIdentifier != NULL)
- UA_NodeId_copy(&newDataSetWriter->identifier, writerIdentifier);
- //add the new writer to the group
- LIST_INSERT_HEAD(&wg->writers, newDataSetWriter, listEntry);
- wg->writersCount++;
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addDataSetWriterRepresentation(server, newDataSetWriter);
- #endif
- return retVal;
- }
- UA_StatusCode
- UA_Server_removeDataSetWriter(UA_Server *server, const UA_NodeId dsw){
- UA_DataSetWriter *dataSetWriter = UA_DataSetWriter_findDSWbyId(server, dsw);
- if(!dataSetWriter)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_WriterGroup *linkedWriterGroup = UA_WriterGroup_findWGbyId(server, dataSetWriter->linkedWriterGroup);
- if(!linkedWriterGroup)
- return UA_STATUSCODE_BADNOTFOUND;
- linkedWriterGroup->writersCount--;
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- removeDataSetWriterRepresentation(server, dataSetWriter);
- #endif
- //remove DataSetWriter from group
- UA_DataSetWriter_deleteMembers(server, dataSetWriter);
- LIST_REMOVE(dataSetWriter, listEntry);
- UA_free(dataSetWriter);
- return UA_STATUSCODE_GOOD;
- }
- /**********************************************/
- /* DataSetField */
- /**********************************************/
- UA_StatusCode
- UA_DataSetFieldConfig_copy(const UA_DataSetFieldConfig *src, UA_DataSetFieldConfig *dst){
- memcpy(dst, src, sizeof(UA_DataSetFieldConfig));
- if(src->dataSetFieldType == UA_PUBSUB_DATASETFIELD_VARIABLE) {
- UA_String_copy(&src->field.variable.fieldNameAlias, &dst->field.variable.fieldNameAlias);
- UA_PublishedVariableDataType_copy(&src->field.variable.publishParameters,
- &dst->field.variable.publishParameters);
- } else {
- return UA_STATUSCODE_BADNOTSUPPORTED;
- }
- return UA_STATUSCODE_GOOD;
- }
- UA_StatusCode
- UA_Server_getDataSetFieldConfig(UA_Server *server, const UA_NodeId dsf,
- UA_DataSetFieldConfig *config) {
- UA_StatusCode retVal = UA_STATUSCODE_GOOD;
- if(!config)
- return UA_STATUSCODE_BADINVALIDARGUMENT;
- UA_DataSetField *currentDataSetField = UA_DataSetField_findDSFbyId(server, dsf);
- if(!currentDataSetField)
- return UA_STATUSCODE_BADNOTFOUND;
- UA_DataSetFieldConfig tmpFieldConfig;
- //deep copy of the actual config
- retVal |= UA_DataSetFieldConfig_copy(¤tDataSetField->config, &tmpFieldConfig);
- *config = tmpFieldConfig;
- return retVal;
- }
- UA_DataSetField *
- UA_DataSetField_findDSFbyId(UA_Server *server, UA_NodeId identifier) {
- for(size_t i = 0; i < server->pubSubManager.publishedDataSetsSize; i++){
- UA_DataSetField *tmpField;
- LIST_FOREACH(tmpField, &server->pubSubManager.publishedDataSets[i].fields, listEntry){
- if(UA_NodeId_equal(&tmpField->identifier, &identifier)){
- return tmpField;
- }
- }
- }
- return NULL;
- }
- void
- UA_DataSetFieldConfig_deleteMembers(UA_DataSetFieldConfig *dataSetFieldConfig){
- if(dataSetFieldConfig->dataSetFieldType == UA_PUBSUB_DATASETFIELD_VARIABLE){
- UA_String_deleteMembers(&dataSetFieldConfig->field.variable.fieldNameAlias);
- UA_PublishedVariableDataType_deleteMembers(&dataSetFieldConfig->field.variable.publishParameters);
- }
- }
- static void
- UA_DataSetField_deleteMembers(UA_DataSetField *field) {
- UA_DataSetFieldConfig_deleteMembers(&field->config);
- //delete DataSetField
- UA_NodeId_deleteMembers(&field->identifier);
- UA_NodeId_deleteMembers(&field->publishedDataSet);
- UA_FieldMetaData_deleteMembers(&field->fieldMetaData);
- }
- /*********************************************************/
- /* PublishValues handling */
- /*********************************************************/
- /**
- * Compare two variants. Internally used for value change detection.
- *
- * @return true if the value has changed
- */
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- static UA_Boolean
- valueChangedVariant(UA_Variant *oldValue, UA_Variant *newValue){
- if(! (oldValue && newValue))
- return false;
- UA_ByteString *oldValueEncoding = UA_ByteString_new(), *newValueEncoding = UA_ByteString_new();
- size_t oldValueEncodingSize, newValueEncodingSize;
- oldValueEncodingSize = UA_calcSizeBinary(oldValue, &UA_TYPES[UA_TYPES_VARIANT]);
- newValueEncodingSize = UA_calcSizeBinary(newValue, &UA_TYPES[UA_TYPES_VARIANT]);
- if((oldValueEncodingSize == 0) || (newValueEncodingSize == 0))
- return false;
- if(oldValueEncodingSize != newValueEncodingSize)
- return true;
- if(UA_ByteString_allocBuffer(oldValueEncoding, oldValueEncodingSize) != UA_STATUSCODE_GOOD)
- return false;
- if(UA_ByteString_allocBuffer(newValueEncoding, newValueEncodingSize) != UA_STATUSCODE_GOOD)
- return false;
- UA_Byte *bufPosOldValue = oldValueEncoding->data;
- const UA_Byte *bufEndOldValue = &oldValueEncoding->data[oldValueEncoding->length];
- UA_Byte *bufPosNewValue = newValueEncoding->data;
- const UA_Byte *bufEndNewValue = &newValueEncoding->data[newValueEncoding->length];
- if(UA_encodeBinary(oldValue, &UA_TYPES[UA_TYPES_VARIANT],
- &bufPosOldValue, &bufEndOldValue, NULL, NULL) != UA_STATUSCODE_GOOD){
- return false;
- }
- if(UA_encodeBinary(newValue, &UA_TYPES[UA_TYPES_VARIANT],
- &bufPosNewValue, &bufEndNewValue, NULL, NULL) != UA_STATUSCODE_GOOD){
- return false;
- }
- oldValueEncoding->length = (uintptr_t)bufPosOldValue - (uintptr_t)oldValueEncoding->data;
- newValueEncoding->length = (uintptr_t)bufPosNewValue - (uintptr_t)newValueEncoding->data;
- UA_Boolean compareResult = !UA_ByteString_equal(oldValueEncoding, newValueEncoding);
- UA_ByteString_delete(oldValueEncoding);
- UA_ByteString_delete(newValueEncoding);
- return compareResult;
- }
- #endif
- /**
- * Obtain the latest value for a specific DataSetField. This method is currently
- * called inside the DataSetMessage generation process.
- */
- static void
- UA_PubSubDataSetField_sampleValue(UA_Server *server, UA_DataSetField *field,
- UA_DataValue *value) {
- /* Read the value */
- UA_ReadValueId rvid;
- UA_ReadValueId_init(&rvid);
- rvid.nodeId = field->config.field.variable.publishParameters.publishedVariable;
- rvid.attributeId = field->config.field.variable.publishParameters.attributeId;
- rvid.indexRange = field->config.field.variable.publishParameters.indexRange;
- *value = UA_Server_read(server, &rvid, UA_TIMESTAMPSTORETURN_BOTH);
- }
- static UA_StatusCode
- UA_PubSubDataSetWriter_generateKeyFrameMessage(UA_Server *server, UA_DataSetMessage *dataSetMessage,
- UA_DataSetWriter *dataSetWriter) {
- UA_PublishedDataSet *currentDataSet =
- UA_PublishedDataSet_findPDSbyId(server, dataSetWriter->connectedDataSet);
- if(!currentDataSet)
- return UA_STATUSCODE_BADNOTFOUND;
- /* Prepare DataSetMessageContent */
- dataSetMessage->header.dataSetMessageValid = true;
- dataSetMessage->header.dataSetMessageType = UA_DATASETMESSAGE_DATAKEYFRAME;
- dataSetMessage->data.keyFrameData.fieldCount = currentDataSet->fieldSize;
- dataSetMessage->data.keyFrameData.dataSetFields = (UA_DataValue *)
- UA_Array_new(currentDataSet->fieldSize, &UA_TYPES[UA_TYPES_DATAVALUE]);
- if(!dataSetMessage->data.keyFrameData.dataSetFields)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- #ifdef UA_ENABLE_JSON_ENCODING
- /* json: insert fieldnames used as json keys */
- dataSetMessage->data.keyFrameData.fieldNames =
- (UA_String *)UA_Array_new(currentDataSet->fieldSize, &UA_TYPES[UA_TYPES_STRING]);
- if(!dataSetMessage->data.keyFrameData.fieldNames)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- #endif
- /* Loop over the fields */
- size_t counter = 0;
- UA_DataSetField *dsf;
- LIST_FOREACH(dsf, ¤tDataSet->fields, listEntry) {
- #ifdef UA_ENABLE_JSON_ENCODING
- /* json: store the fieldNameAlias*/
- UA_String_copy(&dsf->config.field.variable.fieldNameAlias,
- &dataSetMessage->data.keyFrameData.fieldNames[counter]);
- #endif
- /* Sample the value */
- UA_DataValue *dfv = &dataSetMessage->data.keyFrameData.dataSetFields[counter];
- UA_PubSubDataSetField_sampleValue(server, dsf, dfv);
- /* Deactivate statuscode? */
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_STATUSCODE) == 0)
- dfv->hasStatus = false;
- /* Deactivate timestamps */
- if(((u64)dataSetWriter->config.dataSetFieldContentMask &
- (u64)UA_DATASETFIELDCONTENTMASK_SOURCETIMESTAMP) == 0)
- dfv->hasSourceTimestamp = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask &
- (u64)UA_DATASETFIELDCONTENTMASK_SOURCEPICOSECONDS) == 0)
- dfv->hasSourcePicoseconds = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask &
- (u64)UA_DATASETFIELDCONTENTMASK_SERVERTIMESTAMP) == 0)
- dfv->hasServerTimestamp = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask &
- (u64)UA_DATASETFIELDCONTENTMASK_SERVERPICOSECONDS) == 0)
- dfv->hasServerPicoseconds = false;
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- /* Update lastValue store */
- UA_DataValue_deleteMembers(&dataSetWriter->lastSamples[counter].value);
- UA_DataValue_copy(dfv, &dataSetWriter->lastSamples[counter].value);
- #endif
- counter++;
- }
- return UA_STATUSCODE_GOOD;
- }
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- static UA_StatusCode
- UA_PubSubDataSetWriter_generateDeltaFrameMessage(UA_Server *server,
- UA_DataSetMessage *dataSetMessage,
- UA_DataSetWriter *dataSetWriter) {
- UA_PublishedDataSet *currentDataSet =
- UA_PublishedDataSet_findPDSbyId(server, dataSetWriter->connectedDataSet);
- if(!currentDataSet)
- return UA_STATUSCODE_BADNOTFOUND;
- /* Prepare DataSetMessageContent */
- memset(dataSetMessage, 0, sizeof(UA_DataSetMessage));
- dataSetMessage->header.dataSetMessageValid = true;
- dataSetMessage->header.dataSetMessageType = UA_DATASETMESSAGE_DATADELTAFRAME;
- UA_DataSetField *dsf;
- size_t counter = 0;
- LIST_FOREACH(dsf, ¤tDataSet->fields, listEntry) {
- /* Sample the value */
- UA_DataValue value;
- UA_DataValue_init(&value);
- UA_PubSubDataSetField_sampleValue(server, dsf, &value);
- /* Check if the value has changed */
- if(valueChangedVariant(&dataSetWriter->lastSamples[counter].value.value, &value.value)) {
- /* increase fieldCount for current delta message */
- dataSetMessage->data.deltaFrameData.fieldCount++;
- dataSetWriter->lastSamples[counter].valueChanged = true;
- /* Update last stored sample */
- UA_DataValue_deleteMembers(&dataSetWriter->lastSamples[counter].value);
- dataSetWriter->lastSamples[counter].value = value;
- } else {
- UA_DataValue_deleteMembers(&value);
- dataSetWriter->lastSamples[counter].valueChanged = false;
- }
- counter++;
- }
- /* Allocate DeltaFrameFields */
- UA_DataSetMessage_DeltaFrameField *deltaFields = (UA_DataSetMessage_DeltaFrameField *)
- UA_calloc(dataSetMessage->data.deltaFrameData.fieldCount, sizeof(UA_DataSetMessage_DeltaFrameField));
- if(!deltaFields)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- dataSetMessage->data.deltaFrameData.deltaFrameFields = deltaFields;
- size_t currentDeltaField = 0;
- for(size_t i = 0; i < currentDataSet->fieldSize; i++) {
- if(!dataSetWriter->lastSamples[i].valueChanged)
- continue;
- UA_DataSetMessage_DeltaFrameField *dff = &deltaFields[currentDeltaField];
- dff->fieldIndex = (UA_UInt16) i;
- UA_DataValue_copy(&dataSetWriter->lastSamples[i].value, &dff->fieldValue);
- dataSetWriter->lastSamples[i].valueChanged = false;
- /* Deactivate statuscode? */
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_STATUSCODE) == 0)
- dff->fieldValue.hasStatus = false;
- /* Deactivate timestamps? */
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_SOURCETIMESTAMP) == 0)
- dff->fieldValue.hasSourceTimestamp = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_SOURCEPICOSECONDS) == 0)
- dff->fieldValue.hasServerPicoseconds = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_SERVERTIMESTAMP) == 0)
- dff->fieldValue.hasServerTimestamp = false;
- if(((u64)dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_SERVERPICOSECONDS) == 0)
- dff->fieldValue.hasServerPicoseconds = false;
- currentDeltaField++;
- }
- return UA_STATUSCODE_GOOD;
- }
- #endif
- /**
- * Generate a DataSetMessage for the given writer.
- *
- * @param dataSetWriter ptr to corresponding writer
- * @return ptr to generated DataSetMessage
- */
- static UA_StatusCode
- UA_DataSetWriter_generateDataSetMessage(UA_Server *server, UA_DataSetMessage *dataSetMessage,
- UA_DataSetWriter *dataSetWriter) {
- UA_PublishedDataSet *currentDataSet =
- UA_PublishedDataSet_findPDSbyId(server, dataSetWriter->connectedDataSet);
- if(!currentDataSet)
- return UA_STATUSCODE_BADNOTFOUND;
- /* Reset the message */
- memset(dataSetMessage, 0, sizeof(UA_DataSetMessage));
- /* store messageType to switch between json or uadp (default) */
- UA_UInt16 messageType = UA_TYPES_UADPDATASETWRITERMESSAGEDATATYPE;
- UA_JsonDataSetWriterMessageDataType *jsonDataSetWriterMessageDataType = NULL;
- /* The configuration Flags are included
- * inside the std. defined UA_UadpDataSetWriterMessageDataType */
- UA_UadpDataSetWriterMessageDataType defaultUadpConfiguration;
- UA_UadpDataSetWriterMessageDataType *dataSetWriterMessageDataType = NULL;
- if((dataSetWriter->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED ||
- dataSetWriter->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED_NODELETE) &&
- (dataSetWriter->config.messageSettings.content.decoded.type ==
- &UA_TYPES[UA_TYPES_UADPDATASETWRITERMESSAGEDATATYPE])) {
- dataSetWriterMessageDataType = (UA_UadpDataSetWriterMessageDataType *)
- dataSetWriter->config.messageSettings.content.decoded.data;
- /* type is UADP */
- messageType = UA_TYPES_UADPDATASETWRITERMESSAGEDATATYPE;
- } else if((dataSetWriter->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED ||
- dataSetWriter->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED_NODELETE) &&
- (dataSetWriter->config.messageSettings.content.decoded.type ==
- &UA_TYPES[UA_TYPES_JSONDATASETWRITERMESSAGEDATATYPE])) {
- jsonDataSetWriterMessageDataType = (UA_JsonDataSetWriterMessageDataType *)
- dataSetWriter->config.messageSettings.content.decoded.data;
- /* type is JSON */
- messageType = UA_TYPES_JSONDATASETWRITERMESSAGEDATATYPE;
- } else {
- /* create default flag configuration if no
- * UadpDataSetWriterMessageDataType was passed in */
- memset(&defaultUadpConfiguration, 0, sizeof(UA_UadpDataSetWriterMessageDataType));
- defaultUadpConfiguration.dataSetMessageContentMask = (UA_UadpDataSetMessageContentMask)
- ((u64)UA_UADPDATASETMESSAGECONTENTMASK_TIMESTAMP | (u64)UA_UADPDATASETMESSAGECONTENTMASK_MAJORVERSION |
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_MINORVERSION);
- dataSetWriterMessageDataType = &defaultUadpConfiguration;
- }
- /* Sanity-test the configuration */
- if(dataSetWriterMessageDataType &&
- (dataSetWriterMessageDataType->networkMessageNumber != 0 ||
- dataSetWriterMessageDataType->dataSetOffset != 0 ||
- dataSetWriterMessageDataType->configuredSize != 0)) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "Static DSM configuration not supported. Using defaults");
- dataSetWriterMessageDataType->networkMessageNumber = 0;
- dataSetWriterMessageDataType->dataSetOffset = 0;
- dataSetWriterMessageDataType->configuredSize = 0;
- }
- /* The field encoding depends on the flags inside the writer config.
- * TODO: This can be moved to the encoding layer. */
- if(dataSetWriter->config.dataSetFieldContentMask & (u64)UA_DATASETFIELDCONTENTMASK_RAWDATA
- ) {
- dataSetMessage->header.fieldEncoding = UA_FIELDENCODING_RAWDATA;
- } else if((u64)dataSetWriter->config.dataSetFieldContentMask &
- ((u64)UA_DATASETFIELDCONTENTMASK_SOURCETIMESTAMP | (u64)UA_DATASETFIELDCONTENTMASK_SERVERPICOSECONDS |
- (u64)UA_DATASETFIELDCONTENTMASK_SOURCEPICOSECONDS | (u64)UA_DATASETFIELDCONTENTMASK_STATUSCODE)) {
- dataSetMessage->header.fieldEncoding = UA_FIELDENCODING_DATAVALUE;
- } else {
- dataSetMessage->header.fieldEncoding = UA_FIELDENCODING_VARIANT;
- }
- if(messageType == UA_TYPES_UADPDATASETWRITERMESSAGEDATATYPE) {
- /* Std: 'The DataSetMessageContentMask defines the flags for the content of the DataSetMessage header.' */
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_MAJORVERSION) {
- dataSetMessage->header.configVersionMajorVersionEnabled = true;
- dataSetMessage->header.configVersionMajorVersion =
- currentDataSet->dataSetMetaData.configurationVersion.majorVersion;
- }
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_MINORVERSION) {
- dataSetMessage->header.configVersionMinorVersionEnabled = true;
- dataSetMessage->header.configVersionMinorVersion =
- currentDataSet->dataSetMetaData.configurationVersion.minorVersion;
- }
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_SEQUENCENUMBER) {
- dataSetMessage->header.dataSetMessageSequenceNrEnabled = true;
- dataSetMessage->header.dataSetMessageSequenceNr =
- dataSetWriter->actualDataSetMessageSequenceCount;
- }
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_TIMESTAMP) {
- dataSetMessage->header.timestampEnabled = true;
- dataSetMessage->header.timestamp = UA_DateTime_now();
- }
- /* TODO: Picoseconds resolution not supported atm */
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_PICOSECONDS) {
- dataSetMessage->header.picoSecondsIncluded = false;
- }
- /* TODO: Statuscode not supported yet */
- if((u64)dataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_UADPDATASETMESSAGECONTENTMASK_STATUS) {
- dataSetMessage->header.statusEnabled = false;
- }
- } else if(messageType == UA_TYPES_JSONDATASETWRITERMESSAGEDATATYPE) {
- if((u64)jsonDataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_JSONDATASETMESSAGECONTENTMASK_METADATAVERSION) {
- dataSetMessage->header.configVersionMajorVersionEnabled = true;
- dataSetMessage->header.configVersionMajorVersion =
- currentDataSet->dataSetMetaData.configurationVersion.majorVersion;
- }
- if((u64)jsonDataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_JSONDATASETMESSAGECONTENTMASK_METADATAVERSION) {
- dataSetMessage->header.configVersionMinorVersionEnabled = true;
- dataSetMessage->header.configVersionMinorVersion =
- currentDataSet->dataSetMetaData.configurationVersion.minorVersion;
- }
- if((u64)jsonDataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_JSONDATASETMESSAGECONTENTMASK_SEQUENCENUMBER) {
- dataSetMessage->header.dataSetMessageSequenceNrEnabled = true;
- dataSetMessage->header.dataSetMessageSequenceNr =
- dataSetWriter->actualDataSetMessageSequenceCount;
- }
- if((u64)jsonDataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_JSONDATASETMESSAGECONTENTMASK_TIMESTAMP) {
- dataSetMessage->header.timestampEnabled = true;
- dataSetMessage->header.timestamp = UA_DateTime_now();
- }
- /* TODO: Statuscode not supported yet */
- if((u64)jsonDataSetWriterMessageDataType->dataSetMessageContentMask &
- (u64)UA_JSONDATASETMESSAGECONTENTMASK_STATUS) {
- dataSetMessage->header.statusEnabled = false;
- }
- }
- /* Set the sequence count. Automatically rolls over to zero */
- dataSetWriter->actualDataSetMessageSequenceCount++;
- /* JSON does not differ between deltaframes and keyframes, only keyframes are currently used. */
- if(messageType != UA_TYPES_JSONDATASETWRITERMESSAGEDATATYPE){
- #ifdef UA_ENABLE_PUBSUB_DELTAFRAMES
- /* Check if the PublishedDataSet version has changed -> if yes flush the lastValue store and send a KeyFrame */
- if(dataSetWriter->connectedDataSetVersion.majorVersion != currentDataSet->dataSetMetaData.configurationVersion.majorVersion ||
- dataSetWriter->connectedDataSetVersion.minorVersion != currentDataSet->dataSetMetaData.configurationVersion.minorVersion) {
- /* Remove old samples */
- for(size_t i = 0; i < dataSetWriter->lastSamplesCount; i++)
- UA_DataValue_deleteMembers(&dataSetWriter->lastSamples[i].value);
- /* Realloc pds dependent memory */
- dataSetWriter->lastSamplesCount = currentDataSet->fieldSize;
- UA_DataSetWriterSample *newSamplesArray = (UA_DataSetWriterSample * )
- UA_realloc(dataSetWriter->lastSamples, sizeof(UA_DataSetWriterSample) * dataSetWriter->lastSamplesCount);
- if(!newSamplesArray)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- dataSetWriter->lastSamples = newSamplesArray;
- memset(dataSetWriter->lastSamples, 0, sizeof(UA_DataSetWriterSample) * dataSetWriter->lastSamplesCount);
- dataSetWriter->connectedDataSetVersion = currentDataSet->dataSetMetaData.configurationVersion;
- UA_PubSubDataSetWriter_generateKeyFrameMessage(server, dataSetMessage, dataSetWriter);
- dataSetWriter->deltaFrameCounter = 0;
- return UA_STATUSCODE_GOOD;
- }
- /* The standard defines: if a PDS contains only one fields no delta messages
- * should be generated because they need more memory than a keyframe with 1
- * field. */
- if(currentDataSet->fieldSize > 1 && dataSetWriter->deltaFrameCounter > 0 &&
- dataSetWriter->deltaFrameCounter <= dataSetWriter->config.keyFrameCount) {
- UA_PubSubDataSetWriter_generateDeltaFrameMessage(server, dataSetMessage, dataSetWriter);
- dataSetWriter->deltaFrameCounter++;
- return UA_STATUSCODE_GOOD;
- }
- dataSetWriter->deltaFrameCounter = 1;
- #endif
- }
- UA_PubSubDataSetWriter_generateKeyFrameMessage(server, dataSetMessage, dataSetWriter);
- return UA_STATUSCODE_GOOD;
- }
- static UA_StatusCode
- sendNetworkMessageJson(UA_PubSubConnection *connection, UA_DataSetMessage *dsm,
- UA_UInt16 *writerIds, UA_Byte dsmCount, UA_ExtensionObject *transportSettings) {
- UA_StatusCode retval = UA_STATUSCODE_BADNOTSUPPORTED;
- #ifdef UA_ENABLE_JSON_ENCODING
- UA_NetworkMessage nm;
- memset(&nm, 0, sizeof(UA_NetworkMessage));
- nm.version = 1;
- nm.networkMessageType = UA_NETWORKMESSAGE_DATASET;
- nm.payloadHeaderEnabled = true;
- nm.payloadHeader.dataSetPayloadHeader.count = dsmCount;
- nm.payloadHeader.dataSetPayloadHeader.dataSetWriterIds = writerIds;
- nm.payload.dataSetPayload.dataSetMessages = dsm;
- /* Allocate the buffer. Allocate on the stack if the buffer is small. */
- UA_ByteString buf;
- size_t msgSize = UA_NetworkMessage_calcSizeJson(&nm, NULL, 0, NULL, 0, true);
- size_t stackSize = 1;
- if(msgSize <= UA_MAX_STACKBUF)
- stackSize = msgSize;
- UA_STACKARRAY(UA_Byte, stackBuf, stackSize);
- buf.data = stackBuf;
- buf.length = msgSize;
- if(msgSize > UA_MAX_STACKBUF) {
- retval = UA_ByteString_allocBuffer(&buf, msgSize);
- if(retval != UA_STATUSCODE_GOOD)
- return retval;
- }
- /* Encode the message */
- UA_Byte *bufPos = buf.data;
- memset(bufPos, 0, msgSize);
- const UA_Byte *bufEnd = &buf.data[buf.length];
- retval = UA_NetworkMessage_encodeJson(&nm, &bufPos, &bufEnd, NULL, 0, NULL, 0, true);
- if(retval != UA_STATUSCODE_GOOD) {
- if(msgSize > UA_MAX_STACKBUF)
- UA_ByteString_deleteMembers(&buf);
- return retval;
- }
- /* Send the prepared messages */
- retval = connection->channel->send(connection->channel, transportSettings, &buf);
- if(msgSize > UA_MAX_STACKBUF)
- UA_ByteString_deleteMembers(&buf);
- #endif
- return retval;
- }
- static UA_StatusCode
- sendNetworkMessage(UA_PubSubConnection *connection, UA_WriterGroup *wg,
- UA_DataSetMessage *dsm, UA_UInt16 *writerIds, UA_Byte dsmCount,
- UA_ExtensionObject *messageSettings,
- UA_ExtensionObject *transportSettings) {
- if(messageSettings->content.decoded.type !=
- &UA_TYPES[UA_TYPES_UADPWRITERGROUPMESSAGEDATATYPE])
- return UA_STATUSCODE_BADINTERNALERROR;
- UA_UadpWriterGroupMessageDataType *wgm = (UA_UadpWriterGroupMessageDataType*)
- messageSettings->content.decoded.data;
- UA_NetworkMessage nm;
- memset(&nm, 0, sizeof(UA_NetworkMessage));
- nm.publisherIdEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_PUBLISHERID) != 0;
- nm.groupHeaderEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_GROUPHEADER) != 0;
- nm.groupHeader.writerGroupIdEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_WRITERGROUPID) != 0;
- nm.groupHeader.groupVersionEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_GROUPVERSION) != 0;
- nm.groupHeader.networkMessageNumberEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_NETWORKMESSAGENUMBER) != 0;
- nm.groupHeader.sequenceNumberEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_SEQUENCENUMBER) != 0;
- nm.payloadHeaderEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_PAYLOADHEADER) != 0;
- nm.timestampEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_TIMESTAMP) != 0;
- nm.picosecondsEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_PICOSECONDS) != 0;
- nm.dataSetClassIdEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_DATASETCLASSID) != 0;
- nm.promotedFieldsEnabled =
- ((u64)wgm->networkMessageContentMask & (u64)UA_UADPNETWORKMESSAGECONTENTMASK_PROMOTEDFIELDS) != 0;
- nm.version = 1;
- nm.networkMessageType = UA_NETWORKMESSAGE_DATASET;
- if(connection->config->publisherIdType == UA_PUBSUB_PUBLISHERID_NUMERIC) {
- nm.publisherIdType = UA_PUBLISHERDATATYPE_UINT16;
- nm.publisherId.publisherIdUInt32 = connection->config->publisherId.numeric;
- } else if(connection->config->publisherIdType == UA_PUBSUB_PUBLISHERID_STRING){
- nm.publisherIdType = UA_PUBLISHERDATATYPE_STRING;
- nm.publisherId.publisherIdString = connection->config->publisherId.string;
- }
- /* Compute the length of the dsm separately for the header */
- UA_STACKARRAY(UA_UInt16, dsmLengths, dsmCount);
- for(UA_Byte i = 0; i < dsmCount; i++)
- dsmLengths[i] = (UA_UInt16)UA_DataSetMessage_calcSizeBinary(&dsm[i]);
- nm.payloadHeader.dataSetPayloadHeader.count = dsmCount;
- nm.payloadHeader.dataSetPayloadHeader.dataSetWriterIds = writerIds;
- nm.groupHeader.writerGroupId = wg->config.writerGroupId;
- nm.groupHeader.networkMessageNumber = 1;
- nm.payload.dataSetPayload.sizes = dsmLengths;
- nm.payload.dataSetPayload.dataSetMessages = dsm;
- /* Allocate the buffer. Allocate on the stack if the buffer is small. */
- UA_ByteString buf;
- size_t msgSize = UA_NetworkMessage_calcSizeBinary(&nm);
- size_t stackSize = 1;
- if(msgSize <= UA_MAX_STACKBUF)
- stackSize = msgSize;
- UA_STACKARRAY(UA_Byte, stackBuf, stackSize);
- buf.data = stackBuf;
- buf.length = msgSize;
- UA_StatusCode retval;
- if(msgSize > UA_MAX_STACKBUF) {
- retval = UA_ByteString_allocBuffer(&buf, msgSize);
- if(retval != UA_STATUSCODE_GOOD)
- return retval;
- }
- /* Encode the message */
- UA_Byte *bufPos = buf.data;
- memset(bufPos, 0, msgSize);
- const UA_Byte *bufEnd = &buf.data[buf.length];
- retval = UA_NetworkMessage_encodeBinary(&nm, &bufPos, bufEnd);
- if(retval != UA_STATUSCODE_GOOD) {
- if(msgSize > UA_MAX_STACKBUF)
- UA_ByteString_deleteMembers(&buf);
- return retval;
- }
- /* Send the prepared messages */
- retval = connection->channel->send(connection->channel, transportSettings, &buf);
- if(msgSize > UA_MAX_STACKBUF)
- UA_ByteString_deleteMembers(&buf);
- return retval;
- }
- /* This callback triggers the collection and publish of NetworkMessages and the
- * contained DataSetMessages. */
- void
- UA_WriterGroup_publishCallback(UA_Server *server, UA_WriterGroup *writerGroup) {
- UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER, "Publish Callback");
- if(!writerGroup) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "Publish failed. WriterGroup not found");
- return;
- }
- /* Nothing to do? */
- if(writerGroup->writersCount <= 0)
- return;
- /* Binary or Json encoding? */
- if(writerGroup->config.encodingMimeType != UA_PUBSUB_ENCODING_UADP &&
- writerGroup->config.encodingMimeType != UA_PUBSUB_ENCODING_JSON) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "Publish failed: Unknown encoding type.");
- return;
- }
- /* Find the connection associated with the writer */
- UA_PubSubConnection *connection =
- UA_PubSubConnection_findConnectionbyId(server, writerGroup->linkedConnection);
- if(!connection) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "Publish failed. PubSubConnection invalid.");
- return;
- }
- /* How many DSM can be sent in one NM? */
- UA_Byte maxDSM = (UA_Byte)writerGroup->config.maxEncapsulatedDataSetMessageCount;
- if(writerGroup->config.maxEncapsulatedDataSetMessageCount > UA_BYTE_MAX)
- maxDSM = UA_BYTE_MAX;
- /* If the maxEncapsulatedDataSetMessageCount is set to 0->1 */
- if(maxDSM == 0)
- maxDSM = 1;
- /* It is possible to put several DataSetMessages into one NetworkMessage.
- * But only if they do not contain promoted fields. NM with only DSM are
- * sent out right away. The others are kept in a buffer for "batching". */
- size_t dsmCount = 0;
- UA_DataSetWriter *dsw;
- UA_STACKARRAY(UA_UInt16, dsWriterIds, writerGroup->writersCount);
- UA_STACKARRAY(UA_DataSetMessage, dsmStore, writerGroup->writersCount);
- LIST_FOREACH(dsw, &writerGroup->writers, listEntry) {
- /* Find the dataset */
- UA_PublishedDataSet *pds =
- UA_PublishedDataSet_findPDSbyId(server, dsw->connectedDataSet);
- if(!pds) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Publish: PublishedDataSet not found");
- continue;
- }
- /* Generate the DSM */
- UA_StatusCode res =
- UA_DataSetWriter_generateDataSetMessage(server, &dsmStore[dsmCount], dsw);
- if(res != UA_STATUSCODE_GOOD) {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Publish: DataSetMessage creation failed");
- continue;
- }
- /* Send right away if there is only this DSM in a NM. If promoted fields
- * are contained in the PublishedDataSet, then this DSM must go into a
- * dedicated NM as well. */
- if(pds->promotedFieldsCount > 0 || maxDSM == 1) {
- if(writerGroup->config.encodingMimeType == UA_PUBSUB_ENCODING_UADP){
- res = sendNetworkMessage(connection, writerGroup, &dsmStore[dsmCount],
- &dsw->config.dataSetWriterId, 1,
- &writerGroup->config.messageSettings,
- &writerGroup->config.transportSettings);
- }else if(writerGroup->config.encodingMimeType == UA_PUBSUB_ENCODING_JSON){
- res = sendNetworkMessageJson(connection, &dsmStore[dsmCount],
- &dsw->config.dataSetWriterId, 1, &writerGroup->config.transportSettings);
- }
- if(res != UA_STATUSCODE_GOOD)
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Publish: Could not send a NetworkMessage");
- UA_DataSetMessage_free(&dsmStore[dsmCount]);
- continue;
- }
- dsWriterIds[dsmCount] = dsw->config.dataSetWriterId;
- dsmCount++;
- }
- /* Send the NetworkMessages with batched DataSetMessages */
- size_t nmCount = (dsmCount / maxDSM) + ((dsmCount % maxDSM) == 0 ? 0 : 1);
- for(UA_UInt32 i = 0; i < nmCount; i++) {
- UA_Byte nmDsmCount = maxDSM;
- if(i == nmCount - 1)
- nmDsmCount = (UA_Byte)dsmCount % maxDSM;
- UA_StatusCode res3 = UA_STATUSCODE_GOOD;
- if(writerGroup->config.encodingMimeType == UA_PUBSUB_ENCODING_UADP){
- res3 = sendNetworkMessage(connection, writerGroup, &dsmStore[i * maxDSM],
- &dsWriterIds[i * maxDSM], nmDsmCount,
- &writerGroup->config.messageSettings,
- &writerGroup->config.transportSettings);
- }else if(writerGroup->config.encodingMimeType == UA_PUBSUB_ENCODING_JSON){
- res3 = sendNetworkMessageJson(connection, &dsmStore[i * maxDSM],
- &dsWriterIds[i * maxDSM], nmDsmCount, &writerGroup->config.transportSettings);
- }
- if(res3 != UA_STATUSCODE_GOOD)
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Publish: Sending a NetworkMessage failed");
- }
- /* Clean up DSM */
- for(size_t i = 0; i < dsmCount; i++)
- UA_DataSetMessage_free(&dsmStore[i]);
- }
- /* Add new publishCallback. The first execution is triggered directly after
- * creation. */
- UA_StatusCode
- UA_WriterGroup_addPublishCallback(UA_Server *server, UA_WriterGroup *writerGroup) {
- UA_StatusCode retval =
- UA_PubSubManager_addRepeatedCallback(server,
- (UA_ServerCallback) UA_WriterGroup_publishCallback,
- writerGroup, writerGroup->config.publishingInterval,
- &writerGroup->publishCallbackId);
- if(retval == UA_STATUSCODE_GOOD)
- writerGroup->publishCallbackIsRegistered = true;
- /* Run once after creation */
- UA_WriterGroup_publishCallback(server, writerGroup);
- return retval;
- }
- /* This callback triggers the collection and reception of NetworkMessages and the
- * contained DataSetMessages. */
- void UA_ReaderGroup_subscribeCallback(UA_Server *server, UA_ReaderGroup *readerGroup) {
- UA_PubSubConnection *connection = UA_PubSubConnection_findConnectionbyId(server, readerGroup->linkedConnection);
- UA_ByteString buffer;
- if(UA_ByteString_allocBuffer(&buffer, 512) != UA_STATUSCODE_GOOD) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "Message buffer alloc failed!");
- return;
- }
- connection->channel->receive(connection->channel, &buffer, NULL, 300000);
- if(buffer.length > 0) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "Message received:");
- UA_NetworkMessage currentNetworkMessage;
- memset(¤tNetworkMessage, 0, sizeof(UA_NetworkMessage));
- size_t currentPosition = 0;
- UA_NetworkMessage_decodeBinary(&buffer, ¤tPosition, ¤tNetworkMessage);
- UA_Server_processNetworkMessage(server, ¤tNetworkMessage, connection);
- UA_NetworkMessage_deleteMembers(¤tNetworkMessage);
- }
- UA_ByteString_deleteMembers(&buffer);
- }
- /* Add new subscribeCallback. The first execution is triggered directly after
- * creation. */
- UA_StatusCode
- UA_ReaderGroup_addSubscribeCallback(UA_Server *server, UA_ReaderGroup *readerGroup) {
- UA_StatusCode retval = UA_STATUSCODE_GOOD;
- UA_PubSubConnection *connection = UA_PubSubConnection_findConnectionbyId(server, readerGroup->linkedConnection);
- if(connection != NULL) {
- retval = connection->channel->regist(connection->channel, NULL, NULL);
- if(retval == UA_STATUSCODE_GOOD) {
- retval = UA_PubSubManager_addRepeatedCallback(server,
- (UA_ServerCallback) UA_ReaderGroup_subscribeCallback,
- readerGroup, 5,
- &readerGroup->subscribeCallbackId);
- }
- else {
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER, "register channel failed: 0x%x!", retval);
- }
- }
- if(retval == UA_STATUSCODE_GOOD) {
- readerGroup->subscribeCallbackIsRegistered = true;
- }
- /* Run once after creation */
- UA_ReaderGroup_subscribeCallback(server, readerGroup);
- return retval;
- }
- #endif /* UA_ENABLE_PUBSUB */
|