Expose RxBufferSize/TxBufferSize config properties, wire to ws-streaming

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
 )

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<IStreaming, WsStreaming>(str, context);
+    return createWithImplementation<IStreaming, WsStreaming>(str, context, config);
 }
 
 Bool WebsocketStreamingClientModule::onCompleteServerCapability(const ServerCapabilityPtr& source, const ServerCapabilityConfigPtr& target)

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.

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<std::size_t>(static_cast<Int>(populateDefaultConfig(config).getPropertyValue("RxBufferSize"))),
+        static_cast<std::size_t>(static_cast<Int>(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)
 {
 }

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<std::size_t>(static_cast<Int>(config.getPropertyValue("RxBufferSize"))),
+        static_cast<std::size_t>(static_cast<Int>(config.getPropertyValue("TxBufferSize")))}
 {
     _port = config.getPropertyValue("WebsocketStreamingPort");