diff --git a/external/ws-streaming/CMakeLists.txt b/external/ws-streaming/CMakeLists.txt index d9a5b0b..08b9a51 100644 --- a/external/ws-streaming/CMakeLists.txt +++ b/external/ws-streaming/CMakeLists.txt @@ -4,8 +4,6 @@ opendaq_dependency( NAME ws-streaming REQUIRED_VERSION 3.0.7 GIT_REPOSITORY https://github.com/openDAQ/ws-streaming - GIT_REF 9d591b39ec4845a9f65c558411d809800d455a5d + GIT_REF configurable-buffer-sizes EXPECT_TARGET ws-streaming::ws-streaming - PATCH_FILES - ${CMAKE_CURRENT_SOURCE_DIR}/patches/0001-Fixes-plus.patch ) diff --git a/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp b/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp index db65658..1119ad7 100644 --- a/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp +++ b/modules/websocket_streaming_client_module/src/websocket_streaming_client_module_impl.cpp @@ -177,7 +177,7 @@ StreamingPtr WebsocketStreamingClientModule::onCreateStreaming(const StringPtr& DAQ_THROW_EXCEPTION(InvalidParameterException); const StringPtr str = formConnectionString(connectionString, config); - return createWithImplementation(str, context); + return createWithImplementation(str, context, config); } Bool WebsocketStreamingClientModule::onCompleteServerCapability(const ServerCapabilityPtr& source, const ServerCapabilityConfigPtr& target) diff --git a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h index 8150c0c..9aa3aba 100644 --- a/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h +++ b/shared/libraries/websocket_streaming/include/websocket_streaming/ws_streaming.h @@ -79,6 +79,17 @@ class WsStreaming : public Streaming */ static StreamingTypePtr createType(); + /*! + * @brief Merges the specified config with the default config, so that any properties + * missing from @p config are filled in with their default values. + * + * @param config The config to merge with the default config. May be null, in which case + * the default config is returned unmodified. + * @return The merged config. + */ + static PropertyObjectPtr populateDefaultConfig( + const PropertyObjectPtr& config); + public: /*! @@ -88,10 +99,13 @@ class WsStreaming : public Streaming * prefix. The remote peer address and TCP port number are parsed from the connection * string. * @param context The openDAQ context object. + * @param config The streaming configuration, as returned by createDefaultConfig(). May be + * null, in which case defaults are used for all properties. */ explicit WsStreaming( const StringPtr& connectionString, - const ContextPtr& context); + const ContextPtr& context, + const PropertyObjectPtr& config = nullptr); /*! * @brief Destroys a streaming object and stops the Boost.Asio I/O context's thread. diff --git a/shared/libraries/websocket_streaming/src/ws_streaming.cpp b/shared/libraries/websocket_streaming/src/ws_streaming.cpp index 5f2b5a2..da1cec9 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming.cpp @@ -51,10 +51,14 @@ StreamingTypePtr WsStreaming::createType() WsStreaming::WsStreaming( const StringPtr& connectionString, - const ContextPtr& context) + const ContextPtr& context, + const PropertyObjectPtr& config) : Streaming(connectionString, context, true) , ioContext{1} - , wsClient(ioContext.get_executor()) + , wsClient( + ioContext.get_executor(), + static_cast(static_cast(populateDefaultConfig(config).getPropertyValue("RxBufferSize"))), + static_cast(static_cast(populateDefaultConfig(config).getPropertyValue("TxBufferSize")))) { // The ws-streaming library wants a URL like ws://1.2.3.4:7418/foo. // So we simply need to replace the daq.lt:// prefix with ws://. @@ -97,9 +101,34 @@ PropertyObjectPtr WsStreaming::createDefaultConfig() { auto obj = PropertyObject(); obj.addProperty(IntProperty("Port", 7414)); + + // Upper bound, in bytes, on the size of a single frame this connection will receive/transmit; + // oversized frames cause the connection to be closed with an error rather than being + // buffered further. See wss::detail::peer::peer() for details. As a client, this connection + // mostly receives sample data and only sends small subscribe/RPC requests, so - unlike the + // server - it is the receive buffer that needs the larger default here. + obj.addProperty(IntPropertyBuilder("RxBufferSize", 4 * 1024 * 1024).setMinValue(1).build()); + obj.addProperty(IntPropertyBuilder("TxBufferSize", 1 * 1024 * 1024).setMinValue(1).build()); + return obj; } +PropertyObjectPtr WsStreaming::populateDefaultConfig(const PropertyObjectPtr& config) +{ + const auto defConfig = createDefaultConfig(); + if (!config.assigned()) + return defConfig; + + for (const auto& prop : defConfig.getAllProperties()) + { + const auto name = prop.getName(); + if (config.hasProperty(name)) + defConfig.setPropertyValue(name, config.getPropertyValue(name)); + } + + return defConfig; +} + void WsStreaming::onSetActive(bool active) { } diff --git a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp index faabec4..3396c2b 100644 --- a/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp +++ b/shared/libraries/websocket_streaming/src/ws_streaming_server.cpp @@ -56,6 +56,17 @@ PropertyObjectPtr WsStreamingServer::createDefaultConfig(const ContextPtr& conte defaultConfig.addProperty(StringProperty("Path", "/")); + // Upper bound, in bytes, on the size of a single frame a connection will receive/transmit; + // oversized frames cause the connection to be closed with an error rather than being + // buffered further. See wss::detail::peer::peer() for details. + const auto rxBufferSizeProp = + IntPropertyBuilder("RxBufferSize", 1 * 1024 * 1024).setMinValue(1).build(); + defaultConfig.addProperty(rxBufferSizeProp); + + const auto txBufferSizeProp = + IntPropertyBuilder("TxBufferSize", 4 * 1024 * 1024).setMinValue(1).build(); + defaultConfig.addProperty(txBufferSizeProp); + populateDefaultConfigFromProvider(context, defaultConfig); return defaultConfig; } @@ -102,7 +113,10 @@ WsStreamingServer::WsStreamingServer( : Server{ID, config, rootDevice, context} , _rootDevice{rootDevice} , _ioc{1} - , _server{_ioc.get_executor()} + , _server{ + _ioc.get_executor(), + static_cast(static_cast(config.getPropertyValue("RxBufferSize"))), + static_cast(static_cast(config.getPropertyValue("TxBufferSize")))} { _port = config.getPropertyValue("WebsocketStreamingPort");