config max subscriptions per connection (#198)

* config max subscriptions per connection

* fix

* fix

Author: xlc

benches/bench/main.rs

@@ -222,6 +222,7 @@ fn config() -> Config {
                 listen_address: SUBWAY_SERVER_ADDR.to_string(),
                 port: SUBWAY_SERVER_PORT,
                 max_connections: 1024 * 1024,
+                max_subscriptions_per_connection: 1024,
                 max_batch_size: None,
                 request_timeout_seconds: 120,
                 http_methods: Vec::new(),

src/extensions/server/mod.rs

@@ -59,6 +59,8 @@ pub struct ServerConfig {
     pub port: u16,
     pub listen_address: String,
     pub max_connections: u32,
+    #[serde(default = "default_max_subscriptions_per_connection")]
+    pub max_subscriptions_per_connection: u32,
     pub max_batch_size: Option<u32>,
     #[serde(default)]
     pub http_methods: Vec<HttpMethodsConfig>,
@@ -72,6 +74,10 @@ fn default_request_timeout_seconds() -> u64 {
     120
 }
 
+fn default_max_subscriptions_per_connection() -> u32 {
+    1024
+}
+
 #[async_trait]
 impl Extension for SubwayServerBuilder {
     type Config = ServerConfig;
@@ -172,6 +178,7 @@ impl SubwayServerBuilder {
                 .set_http_middleware(http_middleware)
                 .set_batch_request_config(batch_request_config)
                 .max_connections(config.max_connections)
+                .max_subscriptions_per_connection(config.max_subscriptions_per_connection)
                 .set_id_provider(RandomStringIdProvider::new(16))
                 .to_service_builder(),
             rate_limit_builder,

src/server.rs

@@ -259,6 +259,7 @@ mod tests {
                     listen_address: "127.0.0.1".to_string(),
                     port,
                     max_connections: 1024,
+                    max_subscriptions_per_connection: 1024,
                     max_batch_size,
                     request_timeout_seconds: request_timeout_seconds.unwrap_or(10),
                     http_methods: Vec::new(),

src/tests/merge_subscription.rs

@@ -54,6 +54,7 @@ async fn merge_subscription_works() {
                 listen_address: "0.0.0.0".to_string(),
                 port: 0,
                 max_connections: 10,
+                max_subscriptions_per_connection: 1024,
                 max_batch_size: None,
                 request_timeout_seconds: 120,
                 http_methods: Vec::new(),

src/tests/upstream.rs

@@ -36,6 +36,7 @@ async fn upstream_error_propagate() {
                 listen_address: "0.0.0.0".to_string(),
                 port: 0,
                 max_connections: 10,
+                max_subscriptions_per_connection: 1024,
                 max_batch_size: None,
                 request_timeout_seconds: 120,
                 http_methods: Vec::new(),