GithubStar/open62541
1
0
Fork 0
mirror of https://github.com/open62541/open62541.git synced 2026-10-29 13:20:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(pubsub): relocate udp specifics to layer

Browse Source
This commit is contained in:
Jan Hermes
2023-01-04 00:45:39 +01:00
committed by Julius Pfrommer
parent a0bfc92a07
commit 67898e9847
17 changed files with 96 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/open62541/plugin/pubsub.h
+3
View File
@@ -112,14 +112,17 @@ typedef struct UA_PubSubTransportLayer {
UA_ConnectionManager *connectionManager;
// UA_Server *server;
UA_PubSubChannel *(*createPubSubChannel)(struct UA_PubSubTransportLayer *tl, void *ctx);
UA_StatusCode (*createWriterGroupPubSubChannel)(UA_PubSubChannel** outChannel, struct UA_PubSubTransportLayer *tl, const UA_ExtensionObject *writerGroupTransportSettings, void *ctx);
} UA_PubSubTransportLayer;
typedef struct {
void *connection;
UA_NetworkAddressUrlDataType *connectionAddress;
UA_PubSubConnectionConfig *connectionConfig;
UA_NetworkAddressUrlDataType *writerGroupAddress;
UA_Server *server;
UA_Logger *logger;
UA_StatusCode (*decodeAndProcessNetworkMessage)(UA_Server *server,
void *connection,
UA_ByteString *buffer);
plugins/ua_pubsub_ethernet.c
+7
View File
@@ -1344,6 +1344,12 @@ TransportLayerEthernet_addChannel(UA_PubSubTransportLayer *tl, void *ctx) {
return pubSubChannel;
}
static UA_StatusCode
TransportLayerEthernet_addWritergroupChannel(UA_PubSubChannel **out, UA_PubSubTransportLayer *tl, const UA_ExtensionObject *writerGroupTransportSettings, void* ctx) {
*out = NULL;
return UA_STATUSCODE_GOOD;
}
UA_PubSubTransportLayer
UA_PubSubTransportLayerEthernet(void) {
UA_PubSubTransportLayer pubSubTransportLayer;
@@ -1351,6 +1357,7 @@ UA_PubSubTransportLayerEthernet(void) {
pubSubTransportLayer.transportProfileUri =
UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-eth-uadp");
pubSubTransportLayer.createPubSubChannel = &TransportLayerEthernet_addChannel;
pubSubTransportLayer.createWriterGroupPubSubChannel= &TransportLayerEthernet_addWritergroupChannel;
pubSubTransportLayer.connectionManager = NULL;
return pubSubTransportLayer;
plugins/ua_pubsub_mqtt.c
+7
View File
@@ -411,6 +411,12 @@ TransportLayerMQTT_addChannel(UA_PubSubTransportLayer *tl, void *ctx) {
return pubSubChannel;
}
static UA_StatusCode
TransportLayerMQTT_addWritergroupChannel(UA_PubSubChannel **out, UA_PubSubTransportLayer *tl, const UA_ExtensionObject *writerGroupTransportSettings, void* ctx) {
*out = NULL;
return UA_STATUSCODE_GOOD;
}
//MQTT channel factory
UA_PubSubTransportLayer
UA_PubSubTransportLayerMQTT(void){
@@ -418,6 +424,7 @@ UA_PubSubTransportLayerMQTT(void){
pubSubTransportLayer.transportProfileUri = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt");
pubSubTransportLayer.createPubSubChannel = &TransportLayerMQTT_addChannel;
pubSubTransportLayer.createWriterGroupPubSubChannel= &TransportLayerMQTT_addWritergroupChannel;
pubSubTransportLayer.connectionManager = NULL;
return pubSubTransportLayer;
}
plugins/ua_pubsub_udp.c
+35
View File
@@ -523,6 +523,40 @@ TransportLayerUDP_addChannel(UA_PubSubTransportLayer *tl, void *ctx) {
return pubSubChannel;
}
static UA_StatusCode
TransportLayerUDP_addWritergroupChannel(UA_PubSubChannel **out, UA_PubSubTransportLayer *tl, const UA_ExtensionObject *writerGroupTransportSettings, void* ctx) {
UA_TransportLayerContext *tctx = (UA_TransportLayerContext *) ctx;
if(writerGroupTransportSettings && writerGroupTransportSettings->content.decoded.type == &UA_TYPES[UA_TYPES_DATAGRAMWRITERGROUPTRANSPORT2DATATYPE]) {
UA_DatagramWriterGroupTransport2DataType *ts = (UA_DatagramWriterGroupTransport2DataType *) writerGroupTransportSettings->content.decoded.data;
const char *assertedPrefix = "opc.udp://localhost:";
if(strncmp(assertedPrefix, (char *) tctx->connectionAddress->url.data, strlen(assertedPrefix)) != 0) {
UA_LOG_ERROR(tctx->logger, UA_LOGCATEGORY_PUBSUB,
"PubSub WriterGroup creating failed. Invalid Address of PubSub Connection.");
return UA_STATUSCODE_BADCONNECTIONREJECTED;
}
if(ts->address.content.decoded.type == &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]) {
UA_NetworkAddressUrlDataType *address = (UA_NetworkAddressUrlDataType *) ts->address.content.decoded.data;
const char *prefix = "opc.udp://";
if(strncmp(prefix, (char *) address->url.data, strlen(prefix)) == 0) {
/* Retrieve the transport layer for the given profile uri */
tctx->writerGroupAddress = address;
*out = tl->createPubSubChannel(tl, tctx);
return UA_STATUSCODE_GOOD;
} else {
return UA_STATUSCODE_BADCONNECTIONREJECTED;
}
} else {
return UA_STATUSCODE_BADCONNECTIONREJECTED;
}
} else {
return UA_STATUSCODE_GOOD;
}
}
//UDP channel factory
UA_PubSubTransportLayer
UA_PubSubTransportLayerUDP(void) {
@@ -538,6 +572,7 @@ UA_PubSubTransportLayerUDP(void) {
pubSubTransportLayer.transportProfileUri =
UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
pubSubTransportLayer.createPubSubChannel = &TransportLayerUDP_addChannel;
pubSubTransportLayer.createWriterGroupPubSubChannel= &TransportLayerUDP_addWritergroupChannel;
pubSubTransportLayer.connectionManager = NULL;
return pubSubTransportLayer;
}
plugins/ua_pubsub_udp_multicast.c
+6
View File
@@ -869,6 +869,11 @@ TransportLayerUDPMC_addChannel(UA_PubSubTransportLayer *tl, void *ctx) {
}
return pubSubChannel;
}
static UA_StatusCode
TransportLayerUDPMC_addWritergroupChannel(UA_PubSubChannel **out, UA_PubSubTransportLayer *tl, const UA_ExtensionObject *writerGroupTransportSettings, void* ctx) {
*out = NULL;
return UA_STATUSCODE_GOOD;
}
//UDPMC channel factory
UA_PubSubTransportLayer
@@ -877,5 +882,6 @@ UA_PubSubTransportLayerUDPMP(void) {
pubSubTransportLayer.transportProfileUri =
UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
pubSubTransportLayer.createPubSubChannel = &TransportLayerUDPMC_addChannel;
pubSubTransportLayer.createWriterGroupPubSubChannel= &TransportLayerUDPMC_addWritergroupChannel;
return pubSubTransportLayer;
}
src/pubsub/ua_pubsub_writergroup.c
+25 -45
View File
@@ -106,52 +106,32 @@ UA_Server_addWriterGroup(UA_Server *server, const UA_NodeId connection,
newConfig->messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED;
}
/* writerGroupTransportSettings */
newWriterGroup->channel = NULL;
const UA_ExtensionObject *writerGroupTransportSettings = &writerGroupConfig->transportSettings;
if(writerGroupTransportSettings && writerGroupTransportSettings->content.decoded.type == &UA_TYPES[UA_TYPES_DATAGRAMWRITERGROUPTRANSPORT2DATATYPE]) {
UA_DatagramWriterGroupTransport2DataType *ts = (UA_DatagramWriterGroupTransport2DataType *) writerGroupTransportSettings->content.decoded.data;
if(UA_Variant_hasScalarType(&currentConnectionContext->config->address,
&UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE])) {
UA_NetworkAddressUrlDataType *address =
(UA_NetworkAddressUrlDataType *)currentConnectionContext->config->address.data;
const char *prefix = "opc.udp://localhost:";
if(strncmp(prefix, (char *) address->url.data, strlen(prefix)) != 0) {
UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
"PubSub WriterGroup creating failed. Invalid Address of PubSub Connection.");
UA_free(newWriterGroup);
return UA_STATUSCODE_BADINTERNALERROR;
}
}
if(ts->address.content.decoded.type == &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]) {
UA_NetworkAddressUrlDataType *address = (UA_NetworkAddressUrlDataType *) ts->address.content.decoded.data;
const char *prefix = "opc.udp://";
if(strncmp(prefix, (char *) address->url.data, strlen(prefix)) == 0) {
UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER,
"adding udp unicast WriterGroup connecting to "
);
/* Retrieve the transport layer for the given profile uri */
UA_PubSubTransportLayer *tl = UA_getTransportProtocolLayer(server, &currentConnectionContext->config->transportProfileUri);
UA_CHECK_MEM_ERROR(tl, UA_free(newWriterGroup); return UA_STATUSCODE_BADNOTFOUND, &server->config.logger,
UA_LOGCATEGORY_SERVER, "PubSub Connection creation failed. Requested transport layer not found.");
UA_TransportLayerContext ctx;
ctx.writerGroupAddress = address;
ctx.connection = currentConnectionContext;
ctx.connectionConfig = currentConnectionContext->config;
/* TODO: The callback is for readers, not writers. Currently unused. */
ctx.decodeAndProcessNetworkMessage =
(UA_StatusCode (*)(UA_Server *, void *, UA_ByteString *))
UA_decodeAndProcessNetworkMessage;
ctx.server = server;
newWriterGroup->channel = tl->createPubSubChannel(tl, &ctx);
}
}
/* Retrieve the transport layer for the given profile uri */
UA_PubSubTransportLayer *tl = UA_getTransportProtocolLayer(server, &currentConnectionContext->config->transportProfileUri);
UA_CHECK_MEM_ERROR(tl, UA_free(newWriterGroup); return UA_STATUSCODE_BADNOTFOUND, &server->config.logger,
UA_LOGCATEGORY_SERVER, "PubSub Connection creation failed. Requested transport layer not found.");
UA_TransportLayerContext ctx;
ctx.writerGroupAddress = NULL;
ctx.connection = currentConnectionContext;
if(UA_Variant_hasScalarType(&currentConnectionContext->config->address,
&UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE])) {
UA_NetworkAddressUrlDataType *address =
(UA_NetworkAddressUrlDataType *)currentConnectionContext->config->address.data;
ctx.connectionAddress = address;
} else {
UA_free(newWriterGroup);
return UA_STATUSCODE_BADCONNECTIONREJECTED;
}
ctx.connectionConfig = currentConnectionContext->config;
/* TODO: The callback is for readers, not writers. Currently unused. */
ctx.decodeAndProcessNetworkMessage =
(UA_StatusCode (*)(UA_Server *, void *, UA_ByteString *))
UA_decodeAndProcessNetworkMessage;
ctx.server = server;
ctx.logger = &server->config.logger;
res = tl->createWriterGroupPubSubChannel(&newWriterGroup->channel, tl, &writerGroupConfig->transportSettings, &ctx);
UA_CHECK_STATUS_ERROR(res, UA_free(newWriterGroup); return res, &server->config.logger, UA_LOGCATEGORY_PUBSUB,
"PubSub Connection creation failed. WriterGroup specific PubSub channel failed");
/* Attach to the connection */
LIST_INSERT_HEAD(&currentConnectionContext->writerGroups, newWriterGroup, listEntry);
tests/pubsub/check_pubsub_config_freeze.c
+1 -1
View File
@@ -21,7 +21,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
}
tests/pubsub/check_pubsub_get_state.c
+1 -1
View File
@@ -18,7 +18,7 @@ static void setup(void) {
assert(server != 0);
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
}
tests/pubsub/check_pubsub_informationmodel.c
+1 -1
View File
@@ -30,7 +30,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
}
tests/pubsub/check_pubsub_informationmodel_methods.c
+1 -1
View File
@@ -33,7 +33,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
THREAD_CREATE(server_thread, serverloop);
}
tests/pubsub/check_pubsub_keystorage.c
+1 -1
View File
@@ -183,7 +183,7 @@ setup(void) {
SecurityGroupId = UA_STRING("TestSecurityGroup");
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
config->pubSubConfig.securityPolicies = (UA_PubSubSecurityPolicy*)
UA_malloc(sizeof(UA_PubSubSecurityPolicy));
tests/pubsub/check_pubsub_multiple_layer.c
+3 -3
View File
@@ -27,13 +27,13 @@ START_TEST(AddMultipleTransportLayers){
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
ck_assert_uint_eq(config->pubSubConfig.transportLayersSize, 1);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
ck_assert_uint_eq(config->pubSubConfig.transportLayersSize, 2);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
ck_assert_uint_eq(config->pubSubConfig.transportLayersSize, 3);
UA_Server_delete(server);
tests/pubsub/check_pubsub_pds.c
+1 -1
View File
@@ -19,7 +19,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
}
tests/pubsub/check_pubsub_publish.c
+1 -1
View File
@@ -24,7 +24,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
//add 2 connections
tests/pubsub/check_pubsub_publish_json.c
+1 -1
View File
@@ -23,7 +23,7 @@ static void setup(void) {
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
UA_PubSubConnectionConfig connectionConfig;
tests/pubsub/check_pubsub_publishspeed.c
+1 -1
View File
@@ -22,7 +22,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_PubSubConnectionConfig connectionConfig;
memset(&connectionConfig, 0, sizeof(UA_PubSubConnectionConfig));
tests/pubsub/check_pubsub_subscribe_config_freeze.c
+1 -1
View File
@@ -21,7 +21,7 @@ static void setup(void) {
server = UA_Server_new();
UA_ServerConfig *config = UA_Server_getConfig(server);
UA_ServerConfig_setDefault(config);
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDPMP());
UA_ServerConfig_addPubSubTransportLayer(config, UA_PubSubTransportLayerUDP());
UA_Server_run_startup(server);
}