feat!: Upgrade to hyper 1.0

.github/workflows/ci.yml

@@ -20,7 +20,7 @@ jobs:
 
       - name: Setup rust tooling
         run: |
-          rustup override set 1.83
+          rustup override set 1.85
           rustup component add rustfmt clippy
 
       - uses: ./.github/actions/ci
@@ -49,7 +49,7 @@ jobs:
 
       - name: Setup rust tooling
         run: |
-          rustup override set 1.83
+          rustup override set 1.85
           rustup component add rustfmt clippy
           rustup target add x86_64-unknown-linux-musl
 

.github/workflows/manual-publish.yml

@@ -19,7 +19,7 @@ jobs:
 
       - name: Setup rust tooling
         run: |
-          rustup override set 1.83
+          rustup override set 1.85
           rustup component add rustfmt clippy
 
       - uses: ./.github/actions/ci

.github/workflows/release-please.yml

@@ -24,7 +24,7 @@ jobs:
       - name: Setup rust tooling
         if: ${{ steps.release.outputs['launchdarkly-server-sdk--release_created'] == 'true' }}
         run: |
-          rustup override set 1.83
+          rustup override set 1.85
           rustup component add rustfmt clippy
 
       - uses: launchdarkly/gh-actions/actions/[email protected]

contract-tests/Cargo.toml

@@ -2,25 +2,27 @@
 name = "contract-tests"
 version = "0.1.0"
 edition = "2021"
-rust-version = "1.83.0"  # MSRV
+rust-version = "1.85.0"  # MSRV
 license = "Apache-2.0"
 
 [dependencies]
 actix = "0.13.0"
 actix-web = "4.2.1"
 env_logger = "0.10.0"
+# eventsource-client = { version = "0.16.0", default-features = false }
+eventsource-client = { git = "https://github.com/launchdarkly/rust-eventsource-client", branch = "feat/hyper-as-feature" }
 log = "0.4.14"
 launchdarkly-server-sdk = { path = "../launchdarkly-server-sdk/", default-features = false, features = ["event-compression"]}
 serde = { version = "1.0.132", features = ["derive"] }
 serde_json = "1.0.73"
 futures = "0.3.12"
-hyper = { version = "0.14.19", features = ["client"] }
-hyper-rustls = { version = "0.24.1" , optional = true, features = ["http2"]}
-hyper-tls = { version = "0.5.0", optional = true }
+hyper-util = { version = "0.1", features = ["client-legacy", "http1", "http2", "tokio"] }
+hyper-rustls = { version = "0.27", default-features = false, features = ["http1", "http2", "native-tokio", "ring", "webpki-roots"], optional = true }
+hyper-tls = { version = "0.6.0", optional = true }
 reqwest = { version = "0.12.4", features = ["default", "blocking", "json"] }
 async-mutex = "1.4.0"
 
 [features]
 default = ["rustls"]
-rustls = ["hyper-rustls/http1", "hyper-rustls/http2", "launchdarkly-server-sdk/rustls"]
+rustls = ["hyper-rustls", "launchdarkly-server-sdk/rustls"]
 tls = ["hyper-tls"]

contract-tests/src/client_entity.rs

@@ -1,3 +1,4 @@
+use eventsource_client as es;
 use futures::future::FutureExt;
 use launchdarkly_server_sdk::{
     Context, ContextBuilder, MigratorBuilder, MultiContextBuilder, Reference,
@@ -20,6 +21,7 @@ use crate::command_params::{
     MigrationOperationResponse, MigrationVariationResponse, SecureModeHashResponse,
 };
 use crate::HttpsConnector;
+use crate::StreamingHttpsConnector;
 use crate::{
     command_params::{
         CommandParams, CommandResponse, EvaluateAllFlagsParams, EvaluateAllFlagsResponse,
@@ -36,6 +38,7 @@ impl ClientEntity {
     pub async fn new(
         create_instance_params: CreateInstanceParams,
         connector: HttpsConnector,
+        streaming_https_connector: StreamingHttpsConnector,
     ) -> Result<Self, BuildError> {
         let mut config_builder =
             ConfigBuilder::new(&create_instance_params.configuration.credential);
@@ -71,6 +74,8 @@ impl ClientEntity {
         }
 
         if let Some(streaming) = create_instance_params.configuration.streaming {
+            let transport =
+                es::HyperTransport::builder().build_with_connector(streaming_https_connector);
             if let Some(base_uri) = streaming.base_uri {
                 service_endpoints_builder.streaming_base_url(&base_uri);
             }
@@ -79,7 +84,7 @@ impl ClientEntity {
             if let Some(delay) = streaming.initial_retry_delay_ms {
                 streaming_builder.initial_reconnect_delay(Duration::from_millis(delay));
             }
-            streaming_builder.https_connector(connector.clone());
+            streaming_builder.transport(transport);
 
             config_builder = config_builder.data_source(&streaming_builder);
         } else if let Some(polling) = create_instance_params.configuration.polling {
@@ -98,8 +103,10 @@ impl ClientEntity {
             // If we didn't specify streaming or polling, we fall back to basic streaming. The only
             // customization we provide is the https connector to support testing multiple
             // connectors.
+            let transport =
+                es::HyperTransport::builder().build_with_connector(streaming_https_connector);
             let mut streaming_builder = StreamingDataSourceBuilder::new();
-            streaming_builder.https_connector(connector.clone());
+            streaming_builder.transport(transport);
             config_builder = config_builder.data_source(&streaming_builder);
         }
 
@@ -350,7 +357,7 @@ impl ClientEntity {
                     )),
                 }
             }
-            command => Err(format!("Invalid command requested: {}", command)),
+            command => Err(format!("Invalid command requested: {command}")),
         }
     }
 

contract-tests/src/main.rs

@@ -7,7 +7,7 @@ use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer, Responder, Resu
 use async_mutex::Mutex;
 use client_entity::ClientEntity;
 use futures::executor;
-use hyper::client::HttpConnector;
+use hyper_util::client::legacy::connect::HttpConnector;
 use launchdarkly_server_sdk::Reference;
 use serde::{self, Deserialize, Serialize};
 use std::collections::{HashMap, HashSet};
@@ -132,11 +132,12 @@ async fn create_client(
     let client_entity = match ClientEntity::new(
         create_instance_params.into_inner(),
         app_state.https_connector.clone(),
+        app_state.streaming_https_connector.clone(),
     )
     .await
     {
         Ok(ce) => ce,
-        Err(e) => return HttpResponse::InternalServerError().body(format!("{}", e)),
+        Err(e) => return HttpResponse::InternalServerError().body(format!("{e}")),
     };
 
     let mut counter = app_state.counter.lock().await;
@@ -206,12 +207,18 @@ struct AppState {
     counter: Mutex<u32>,
     client_entities: Mutex<HashMap<u32, ClientEntity>>,
     https_connector: HttpsConnector,
+    streaming_https_connector: StreamingHttpsConnector,
 }
 
 #[cfg(feature = "rustls")]
 type HttpsConnector = hyper_rustls::HttpsConnector<HttpConnector>;
+#[cfg(feature = "rustls")]
+type StreamingHttpsConnector = hyper_util::client::legacy::connect::HttpConnector;
+
 #[cfg(feature = "tls")]
 type HttpsConnector = hyper_tls::HttpsConnector<HttpConnector>;
+#[cfg(feature = "tls")]
+type StreamingHttpsConnector = hyper_tls::HttpsConnector<HttpConnector>;
 
 #[actix_web::main]
 async fn main() -> std::io::Result<()> {
@@ -228,21 +235,27 @@ async fn main() -> std::io::Result<()> {
 
     let (tx, rx) = mpsc::channel::<()>();
 
+    #[cfg(feature = "rustls")]
+    let streaming_https_connector = hyper_util::client::legacy::connect::HttpConnector::new();
     #[cfg(feature = "rustls")]
     let connector = hyper_rustls::HttpsConnectorBuilder::new()
         .with_native_roots()
+        .expect("Failed to load native root certificates")
         .https_or_http()
         .enable_http1()
         .enable_http2()
         .build();
 
+    #[cfg(feature = "tls")]
+    let streaming_https_connector = hyper_tls::HttpsConnector::new();
     #[cfg(feature = "tls")]
     let connector = hyper_tls::HttpsConnector::new();
 
     let state = web::Data::new(AppState {
         counter: Mutex::new(0),
         client_entities: Mutex::new(HashMap::new()),
         https_connector: connector,
+        streaming_https_connector,
     });
 
     let server = HttpServer::new(move || {

launchdarkly-server-sdk/Cargo.toml

@@ -4,7 +4,7 @@ description = "LaunchDarkly Server-Side SDK"
 version = "2.6.2"
 authors = ["LaunchDarkly"]
 edition = "2021"
-rust-version = "1.83.0"  # MSRV
+rust-version = "1.85.0"  # MSRV
 license = "Apache-2.0"
 homepage = "https://docs.launchdarkly.com/sdk/server-side/rust"
 repository = "https://github.com/launchdarkly/rust-server-sdk"
@@ -20,7 +20,8 @@ features = ["event-compression"]
 chrono = "0.4.19"
 crossbeam-channel = "0.5.1"
 data-encoding = "2.3.2"
-eventsource-client = { version = "0.16.0", default-features = false }
+# eventsource-client = { version = "0.16.0", default-features = false }
+eventsource-client = { git = "https://github.com/launchdarkly/rust-eventsource-client", branch = "feat/hyper-as-feature" }
 futures = "0.3.12"
 lazy_static = "1.4.0"
 log = "0.4.14"
@@ -34,8 +35,13 @@ parking_lot = "0.12.0"
 tokio-stream = { version = "0.1.8", features = ["sync"] }
 moka = { version = "0.12.1", features = ["sync"] }
 uuid = {version = "1.2.2", features = ["v4"] }
-hyper = { version = "0.14.19", features = ["client", "http1", "http2", "tcp"] }
-hyper-rustls = { version = "0.24.1" , optional = true}
+http = "1.0"
+bytes = "1.11"
+hyper = { version = "1.0", features = ["client", "http1", "http2"] }
+hyper-util = { version = "0.1", features = ["client-legacy", "http1", "http2", "tokio"] }
+http-body-util = { version = "0.1" }
+hyper-rustls = { version = "0.27", default-features = false, features = ["http1", "http2", "native-tokio", "ring", "webpki-roots"], optional = true}
+tower = { version = "0.4" }
 rand = "0.9"
 flate2 = { version = "1.0.35", optional = true }
 aws-lc-rs = "1.14.1"
@@ -54,7 +60,7 @@ testing_logger = "0.1.1"
 
 [features]
 default = ["rustls"]
-rustls = ["hyper-rustls/http1", "hyper-rustls/http2", "eventsource-client/rustls"]
+rustls = ["hyper-rustls/http1", "hyper-rustls/http2", "eventsource-client/hyper-rustls"]
 event-compression = ["flate2"]
 
 [[example]]

launchdarkly-server-sdk/examples/print_flags.rs

@@ -27,7 +27,7 @@ async fn main() {
         } else if let ["str", name] = bits {
             str_flags.push(name.to_string());
         } else if let [flag_type, _] = bits {
-            error!("Unsupported flag type {} in {}", flag_type, flag);
+            error!("Unsupported flag type {flag_type} in {flag}");
             exit(2);
         } else if let [name] = bits {
             bool_flags.push(name.to_string());

launchdarkly-server-sdk/src/client.rs

@@ -294,10 +294,7 @@ impl Client {
         }
 
         let initialized = tokio::time::timeout(timeout, self.initialized_async_internal()).await;
-        match initialized {
-            Ok(result) => Some(result),
-            Err(_) => None,
-        }
+        initialized.ok()
     }
 
     async fn initialized_async_internal(&self) -> bool {
@@ -335,7 +332,7 @@ impl Client {
         // broadcast channel, so sending on it would always result in an error.
         if !self.offline && !self.daemon_mode {
             if let Err(e) = self.shutdown_broadcast.send(()) {
-                error!("Failed to shutdown client appropriately: {}", e);
+                error!("Failed to shutdown client appropriately: {e}");
             }
         }
 
@@ -380,8 +377,7 @@ impl Client {
             b
         } else {
             warn!(
-                "bool_variation called for a non-bool flag {:?} (got {:?})",
-                flag_key, val
+                "bool_variation called for a non-bool flag {flag_key:?} (got {val:?})"
             );
             default
         }
@@ -400,8 +396,7 @@ impl Client {
             s
         } else {
             warn!(
-                "str_variation called for a non-string flag {:?} (got {:?})",
-                flag_key, val
+                "str_variation called for a non-string flag {flag_key:?} (got {val:?})"
             );
             default
         }
@@ -420,8 +415,7 @@ impl Client {
             f
         } else {
             warn!(
-                "float_variation called for a non-float flag {:?} (got {:?})",
-                flag_key, val
+                "float_variation called for a non-float flag {flag_key:?} (got {val:?})"
             );
             default
         }
@@ -440,8 +434,7 @@ impl Client {
             f
         } else {
             warn!(
-                "int_variation called for a non-int flag {:?} (got {:?})",
-                flag_key, val
+                "int_variation called for a non-int flag {flag_key:?} (got {val:?})"
             );
             default
         }
@@ -761,14 +754,12 @@ impl Client {
                         );
                     }
                     Err(e) => error!(
-                        "Failed to build migration event, no event will be sent: {}",
-                        e
+                        "Failed to build migration event, no event will be sent: {e}"
                     ),
                 }
             }
             Err(e) => error!(
-                "Failed to lock migration tracker, no event will be sent: {}",
-                e
+                "Failed to lock migration tracker, no event will be sent: {e}"
             ),
         }
     }
@@ -849,7 +840,7 @@ mod tests {
     use crossbeam_channel::Receiver;
     use eval::{ContextBuilder, MultiContextBuilder};
     use futures::FutureExt;
-    use hyper::client::HttpConnector;
+    use hyper_util::client::legacy::connect::HttpConnector;
     use launchdarkly_server_sdk_evaluation::{Flag, Reason, Segment};
     use maplit::hashmap;
     use std::collections::HashMap;

launchdarkly-server-sdk/src/config.rs

@@ -6,6 +6,7 @@ use crate::events::processor_builders::{
 };
 use crate::stores::store_builders::{DataStoreFactory, InMemoryDataStoreBuilder};
 use crate::{ServiceEndpointsBuilder, StreamingDataSourceBuilder};
+use eventsource_client as es;
 
 use std::borrow::Borrow;
 
@@ -85,7 +86,7 @@ impl ApplicationInfo {
         match tag.is_valid() {
             Ok(_) => self.tags.push(tag),
             Err(e) => {
-                warn!("{}", e)
+                warn!("{e}")
             }
         }
 
@@ -301,9 +302,9 @@ impl ConfigBuilder {
                 }
                 Some(builder) => Ok(builder),
                 #[cfg(feature = "rustls")]
-                None => Ok(Box::new(StreamingDataSourceBuilder::<
-                    hyper_rustls::HttpsConnector<hyper::client::HttpConnector>,
-                >::new())),
+                None => Ok(Box::new(
+                    StreamingDataSourceBuilder::<es::HyperTransport>::new(),
+                )),
                 #[cfg(not(feature = "rustls"))]
                 None => Err(BuildError::InvalidConfig(
                     "data source builder required when rustls is disabled".into(),
@@ -321,7 +322,7 @@ impl ConfigBuilder {
                 Some(builder) => Ok(builder),
                 #[cfg(feature = "rustls")]
                 None => Ok(Box::new(EventProcessorBuilder::<
-                    hyper_rustls::HttpsConnector<hyper::client::HttpConnector>,
+                    hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>,
                 >::new())),
                 #[cfg(not(feature = "rustls"))]
                 None => Err(BuildError::InvalidConfig(