feat: streaming now enabled/disabled with runtime flag (#605)

This PR takes streaming from being a compile-time feature, to a feature controlled via the CLI. The new arg `streaming` (requires `strict`) will cause Edge to start up in streaming mode instead of polling.

Note that streaming updates from Unleash is still experimental and this is not stable. But this'll allow us to start testing it in our own infrastructure.

server/Cargo.toml

@@ -15,9 +15,6 @@ repository = "https://github.com/Unleash/unleash-edge"
 rust-version = "1.81.0"
 version = "19.6.3"
 
-[features]
-streaming = ["actix-web-lab", "eventsource-client", "tokio-stream"]
-
 [package.metadata.wix]
 upgrade-guid = "11E5D83A-3034-48BB-9A84-9F589EBD648C"
 path-guid = "6F606A3B-C7E9-43EC-8B6E-91D7B74F80FC"
@@ -33,19 +30,19 @@ actix-http = "3.9.0"
 actix-middleware-etag = "0.4.2"
 actix-service = "2.0.2"
 actix-web = { version = "4.9.0", features = ["rustls-0_23", "compress-zstd"] }
-actix-web-lab = { version = "0.23.0", optional = true }
+actix-web-lab = { version = "0.23.0" }
 ahash = "0.8.11"
-anyhow = "1.0.91"
+anyhow = "1.0.94"
 async-trait = "0.1.83"
-aws-config = { version = "1.5.7", features = ["behavior-version-latest"] }
-aws-sdk-s3 = { version = "1.57.0", features = ["behavior-version-latest"] }
+aws-config = { version = "1.5.10", features = ["behavior-version-latest"] }
+aws-sdk-s3 = { version = "1.65.0", features = ["behavior-version-latest"] }
 base64 = "0.22.1"
-chrono = { version = "0.4.38", features = ["serde"] }
+chrono = { version = "0.4.39", features = ["serde"] }
 cidr = "0.3.0"
-clap = { version = "4.5.19", features = ["derive", "env"] }
+clap = { version = "4.5.23", features = ["derive", "env"] }
 clap-markdown = "0.1.4"
 dashmap = "6.0.1"
-eventsource-client = { version = "0.13.0", optional = true }
+eventsource-client = { version = "0.13.0" }
 futures = "0.3.30"
 futures-core = "0.3.30"
 iter_tools = "0.24.0"
@@ -93,12 +90,12 @@ tokio = { version = "1.42.0", features = [
     "tracing",
     "fs",
 ] }
-tokio-stream = { version = "0.1.16", optional = true }
+tokio-stream = { version = "0.1.17" }
 tracing = { version = "0.1.40", features = ["log"] }
 tracing-subscriber = { version = "0.3.18", features = ["json", "env-filter"] }
 ulid = "1.1.2"
 unleash-types = { version = "0.14", features = ["openapi", "hashes"] }
-unleash-yggdrasil = { version = "0.14.0" }
+unleash-yggdrasil = { version = "0.14.1" }
 utoipa = { version = "5", features = ["actix_extras", "chrono"] }
 utoipa-swagger-ui = { version = "8", features = ["actix-web"] }
 [dev-dependencies]

server/src/builder.rs

@@ -12,6 +12,7 @@ use unleash_yggdrasil::EngineState;
 
 use crate::cli::RedisMode;
 use crate::feature_cache::FeatureCache;
+use crate::http::feature_refresher::{FeatureRefreshConfig, FeatureRefresherMode};
 use crate::http::unleash_client::new_reqwest_client;
 use crate::offline::offline_hotload::{load_bootstrap, load_offline_engine_cache};
 use crate::persistence::file::FilePersister;
@@ -258,15 +259,22 @@ async fn build_edge(args: &EdgeArgs, app_name: &str) -> EdgeResult<EdgeInfo> {
         unleash_client: unleash_client.clone(),
         persistence: persistence.clone(),
     });
-
+    let refresher_mode = match (args.strict, args.streaming) {
+        (_, true) => FeatureRefresherMode::Streaming,
+        (true, _) => FeatureRefresherMode::Strict,
+        _ => FeatureRefresherMode::Dynamic,
+    };
+    let feature_config = FeatureRefreshConfig::new(
+        Duration::seconds(args.features_refresh_interval_seconds as i64),
+        refresher_mode,
+        app_name.to_string(),
+    );
     let feature_refresher = Arc::new(FeatureRefresher::new(
         unleash_client,
         feature_cache.clone(),
         engine_cache.clone(),
-        Duration::seconds(args.features_refresh_interval_seconds.try_into().unwrap()),
         persistence.clone(),
-        args.strict,
-        app_name,
+        feature_config,
     ));
     let _ = token_validator.register_tokens(args.tokens.clone()).await;
 
@@ -364,6 +372,7 @@ mod tests {
             prometheus_user_id: None,
             prometheus_password: None,
             prometheus_username: None,
+            streaming: false,
         };
 
         let result = build_edge(&args, "test-app").await;

server/src/cli.rs

@@ -213,6 +213,10 @@ pub struct EdgeArgs {
     #[clap(long, env, default_value_t = false, conflicts_with = "strict")]
     pub dynamic: bool,
 
+    /// If set to true. Edge connects to upstream using streaming instead of polling. Requires strict mode
+    #[clap(long, env, default_value_t = false, requires = "strict")]
+    pub streaming: bool,
+
     /// Sets a remote write url for prometheus metrics, if this is set, prometheus metrics will be written upstream
     #[clap(long, env)]
     pub prometheus_remote_write_url: Option<String>,