From bc081b2f8a9563ee93698511c21343c73b931339 Mon Sep 17 00:00:00 2001 From: Natsuki Ikeguchi Date: Mon, 1 Jan 2024 02:31:52 +0900 Subject: [PATCH] wip --- Cargo.toml | 12 +++--- src/axum_support/mod.rs | 2 +- src/axum_support/slack_events_middleware.rs | 9 ++-- src/axum_support/slack_oauth_routes.rs | 24 +++++++---- src/hyper_tokio/connector.rs | 27 ++++++------ src/hyper_tokio/hyper_ext.rs | 25 +++++------ src/hyper_tokio/listener/command_events.rs | 36 ++++++++++------ .../listener/interaction_events.rs | 41 +++++++++++++------ src/hyper_tokio/listener/mod.rs | 41 +++++++++++++------ src/hyper_tokio/listener/oauth.rs | 31 +++++++++----- src/hyper_tokio/listener/push_events.rs | 34 +++++++++------ src/hyper_tokio/mod.rs | 3 +- src/hyper_tokio/socket_mode/mod.rs | 2 +- .../socket_mode/tokio_clients_manager.rs | 2 +- 14 files changed, 182 insertions(+), 107 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 64ab40de..fcf315a9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ path = "src/lib.rs" [features] default = [] -hyper = ["dep:tokio","dep:hyper", "dep:hyper-rustls", "dep:tokio-stream","dep:tokio-tungstenite", "dep:tokio-tungstenite", "dep:signal-hook", "dep:signal-hook-tokio"] +hyper = ["dep:tokio", "dep:http-body-util", "dep:hyper", "dep:hyper-rustls", "dep:hyper-util", "dep:tokio-stream","dep:tokio-tungstenite", "dep:tokio-tungstenite", "dep:signal-hook", "dep:signal-hook-tokio"] axum = ["hyper", "dep:axum", "dep:tower"] [dependencies] @@ -39,7 +39,7 @@ hex = "0.4" tracing = "0.1" ring = "0.17" lazy_static = "1.4" -http = "0.2" +http = "1.0" async-trait = "0.1" bytes = "1" rand = "0.8" @@ -47,12 +47,14 @@ async-recursion="1.0" mime = "0.3" chrono = { version = "0.4", default-features = false, features = ["clock", "std", "serde"] } url = { version = "2.5", features = ["serde"]} -hyper = { version ="0.14", features = ["http2","server", "client", "h2", "stream"], optional = true } +http-body-util = { version = "0.1", optional = true } +hyper = { version ="1.0", features = ["http2","server", "client"], optional = true } +hyper-util = { version = "0.1", features = ["client", "client-legacy", "server"], optional = true } tokio = { version = "1", features = ["bytes","rt-multi-thread","signal","tracing"], optional = true } tokio-stream = { version = "0.1", optional = true } hyper-rustls = { version="0.24", features = ["rustls-native-certs", "http2"], optional = true } tokio-tungstenite = { version = "0.21.0", features = ["rustls-tls-native-roots"], optional = true } -axum = { version = "0.6", optional = true } +axum = { version = "0.7", optional = true } tower = { version = "0.4", optional = true } serde_urlencoded = "0.7.1" @@ -68,7 +70,7 @@ cargo-husky = { version = "1", default-features = false, features = ["run-for-al cargo-audit = "0.18" tracing-subscriber = { version ="0.3", features = ["env-filter"] } hyper-proxy = "0.9" -hyper = { version ="0.14", features = ["full"] } +hyper = { version ="1.0", features = ["full"] } tokio = { version = "1", features = ["full"] } [package.metadata.release] diff --git a/src/axum_support/mod.rs b/src/axum_support/mod.rs index 733b4f24..dd21e490 100644 --- a/src/axum_support/mod.rs +++ b/src/axum_support/mod.rs @@ -1,6 +1,6 @@ use crate::hyper_tokio::SlackClientHyperConnector; use crate::listener::SlackClientEventsListenerEnvironment; -use hyper::client::connect::Connect; +use hyper_util::client::legacy::connect::Connect; use std::sync::Arc; mod slack_events_middleware; diff --git a/src/axum_support/slack_events_middleware.rs b/src/axum_support/slack_events_middleware.rs index c0401f80..a80dc26c 100644 --- a/src/axum_support/slack_events_middleware.rs +++ b/src/axum_support/slack_events_middleware.rs @@ -7,11 +7,10 @@ use crate::listener::SlackClientEventsListenerEnvironment; use crate::prelude::hyper_ext::HyperExtensions; use crate::signature_verifier::SlackEventSignatureVerifier; use crate::{SlackClientHttpConnector, SlackSigningSecret}; -use axum::body::BoxBody; use axum::response::IntoResponse; use axum::{body::Body, http::Request, response::Response}; use futures_util::future::BoxFuture; -use hyper::client::connect::Connect; +use hyper_util::client::legacy::connect::Connect; use std::convert::Infallible; use std::marker::PhantomData; use std::sync::Arc; @@ -104,7 +103,7 @@ where ); Ok(Response::builder() .status(http_status) - .body(BoxBody::default()) + .body(Body::default()) .unwrap()) } else { *verified_request.body_mut() = Body::from(verified_body); @@ -126,7 +125,7 @@ where ); Ok(Response::builder() .status(http_status) - .body(BoxBody::default()) + .body(Body::default()) .unwrap()) } } @@ -141,7 +140,7 @@ where ); Ok(Response::builder() .status(http_status) - .body(BoxBody::default()) + .body(Body::default()) .unwrap()) } } diff --git a/src/axum_support/slack_oauth_routes.rs b/src/axum_support/slack_oauth_routes.rs index cba3aff9..2bd8d8a6 100644 --- a/src/axum_support/slack_oauth_routes.rs +++ b/src/axum_support/slack_oauth_routes.rs @@ -2,12 +2,13 @@ use crate::axum_support::SlackEventsAxumListener; use crate::hyper_tokio::hyper_ext::HyperExtensions; use crate::listener::{SlackClientEventsListenerEnvironment, UserCallbackFunction}; use crate::prelude::SlackOAuthListenerConfig; -use axum::response::Response; +use axum::body::Body; +use axum::response::{IntoResponse, Response}; use futures_util::future::BoxFuture; use futures_util::FutureExt; use http::Request; -use hyper::client::connect::Connect; -use hyper::Body; +use http_body_util::BodyExt; +use hyper_util::client::legacy::connect::Connect; use rvstruct::ValueStruct; use std::future::Future; use std::sync::Arc; @@ -22,7 +23,7 @@ impl SlackEventsAxumListener { pub fn slack_oauth_install( &self, config: &SlackOAuthListenerConfig, - ) -> impl Fn(Request) -> BoxFuture<'static, Response> + 'static + Send + Clone { + ) -> impl Fn(Request) -> BoxFuture<'static, Response> + 'static + Send + Clone { let environment = self.environment.clone(); let config = config.clone(); move |_| { @@ -41,7 +42,7 @@ impl SlackEventsAxumListener { ], ); debug!("Redirecting to Slack OAuth authorize: {}", &full_uri); - HyperExtensions::hyper_redirect_to(full_uri.as_ref()) + HyperExtensions::hyper_redirect_to(full_uri.as_ref()).map(|r| r.into_response()) } .map(|res| Self::handle_error(environment, res)) .boxed() @@ -66,7 +67,7 @@ impl SlackEventsAxumListener { let err_config = config.clone(); async move { - let params = HyperExtensions::parse_query_params(&req); + let params = HyperExtensions::parse_query_params(req.uri()); debug!("Received Slack OAuth callback: {:?}", ¶ms); match (params.get("code"), params.get("error")) { @@ -105,6 +106,7 @@ impl SlackEventsAxumListener { ) .await; HyperExtensions::hyper_redirect_to(&config.redirect_installed_url) + .map(|r| r.into_response()) } Err(err) => { error!("Slack OAuth error: {}", &err); @@ -116,6 +118,7 @@ impl SlackEventsAxumListener { HyperExtensions::hyper_redirect_to( &config.redirect_error_redirect_url, ) + .map(|r| r.into_response()) } } } @@ -134,6 +137,7 @@ impl SlackEventsAxumListener { req.uri().query().map_or("".into(), |q| format!("?{}", &q)) ); HyperExtensions::hyper_redirect_to(&redirect_error_url) + .map(|r| r.into_response()) } _ => { error!("Slack OAuth cancelled with unknown reason"); @@ -146,6 +150,7 @@ impl SlackEventsAxumListener { environment.user_state.clone(), ); HyperExtensions::hyper_redirect_to(&config.redirect_error_redirect_url) + .map(|r| r.into_response()) } } } @@ -163,6 +168,7 @@ impl SlackEventsAxumListener { ); HyperExtensions::hyper_redirect_to(&err_config.redirect_error_redirect_url) .unwrap() + .into_response() } }) .boxed() @@ -195,8 +201,8 @@ impl SlackEventsAxumListener { fn handle_error( environment: Arc>>, - result: AnyStdResult>, - ) -> Response { + result: AnyStdResult, + ) -> Response { match result { Err(err) => { let http_status = (environment.error_handler)( @@ -206,7 +212,7 @@ impl SlackEventsAxumListener { ); Response::builder() .status(http_status) - .body(hyper::Body::empty()) + .body(Body::empty()) .unwrap() } Ok(result) => result, diff --git a/src/hyper_tokio/connector.rs b/src/hyper_tokio/connector.rs index a047ef0c..24b036f8 100644 --- a/src/hyper_tokio/connector.rs +++ b/src/hyper_tokio/connector.rs @@ -4,10 +4,12 @@ use crate::models::{SlackClientId, SlackClientSecret}; use crate::*; use async_recursion::async_recursion; use futures::future::{BoxFuture, FutureExt}; -use hyper::client::*; +use hyper::body::{Body, Incoming}; use hyper::http::StatusCode; -use hyper::{Body, Request}; +use hyper::Request; use hyper_rustls::HttpsConnector; +use hyper_util::client::legacy::*; +use hyper_util::rt::TokioExecutor; use rvstruct::ValueStruct; use crate::prelude::hyper_ext::HyperExtensions; @@ -21,13 +23,14 @@ use url::Url; #[derive(Clone, Debug)] pub struct SlackClientHyperConnector { - hyper_connector: Client, + hyper_connector: Client, tokio_rate_controller: Option>, } -pub type SlackClientHyperHttpsConnector = SlackClientHyperConnector>; +pub type SlackClientHyperHttpsConnector = + SlackClientHyperConnector>; -impl SlackClientHyperConnector> { +impl SlackClientHyperConnector> { pub fn new() -> Self { let https_connector = hyper_rustls::HttpsConnectorBuilder::new() .with_native_roots() @@ -38,10 +41,10 @@ impl SlackClientHyperConnector> { } } -impl From> - for SlackClientHyperConnector> +impl From> + for SlackClientHyperConnector> { - fn from(https_connector: hyper_rustls::HttpsConnector) -> Self { + fn from(https_connector: HttpsConnector) -> Self { Self::with_connector(https_connector) } } @@ -49,7 +52,7 @@ impl From> impl SlackClientHyperConnector { pub fn with_connector(connector: H) -> Self { Self { - hyper_connector: Client::builder().build::<_, hyper::Body>(connector), + hyper_connector: Client::builder(TokioExecutor::new()).build::<_, Incoming>(connector), tokio_rate_controller: None, } } @@ -65,7 +68,7 @@ impl SlackClientHyperConnec async fn send_http_request<'a, RS>( &'a self, - request: Request, + request: Request, context: SlackClientApiCallContext<'a>, ) -> ClientResult where @@ -185,7 +188,7 @@ impl SlackClientHyperConnec retried: usize, ) -> ClientResult where - R: Fn() -> ClientResult> + Send + Sync, + R: Fn() -> ClientResult> + Send + Sync, RS: for<'de> serde::de::Deserialize<'de> + Send, { match ( @@ -223,7 +226,7 @@ impl SlackClientHyperConnec context: SlackClientApiCallContext<'a>, ) -> ClientResult where - R: Fn() -> ClientResult> + Send + Sync, + R: Fn() -> ClientResult> + Send + Sync, RS: for<'de> serde::de::Deserialize<'de> + Send, { match result { diff --git a/src/hyper_tokio/hyper_ext.rs b/src/hyper_tokio/hyper_ext.rs index 2d9bbe81..b739e053 100644 --- a/src/hyper_tokio/hyper_ext.rs +++ b/src/hyper_tokio/hyper_ext.rs @@ -1,25 +1,25 @@ use crate::signature_verifier::*; use crate::{AnyStdResult, SlackApiToken}; use base64::prelude::*; -use bytes::Buf; +use bytes::{Buf, Bytes}; use futures_util::TryFutureExt; use http::request::Parts; use http::{Request, Response, Uri}; -use hyper::body::HttpBody; -use hyper::Body; +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyExt, Empty}; +use hyper::body::{Body, Incoming}; use mime::Mime; use rvstruct::ValueStruct; use std::collections::HashMap; +use std::convert::Infallible; use std::io::Read; use url::Url; pub struct HyperExtensions; impl HyperExtensions { - pub fn parse_query_params(request: &Request) -> HashMap { - request - .uri() - .query() + pub fn parse_query_params(uri: &Uri) -> HashMap { + uri.query() .map(|v| { url::form_urlencoded::parse(v.as_bytes()) .into_owned() @@ -30,11 +30,12 @@ impl HyperExtensions { pub fn hyper_redirect_to( url: &str, - ) -> Result, Box> { + ) -> Result>, Box> + { Response::builder() .status(hyper::http::StatusCode::FOUND) .header(hyper::header::LOCATION, url) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } @@ -75,10 +76,10 @@ impl HyperExtensions { pub async fn http_body_to_string(body: T) -> AnyStdResult where - T: HttpBody, + T: Body, T::Error: std::error::Error + Sync + Send + 'static, { - let http_body = hyper::body::aggregate(body).await?; + let http_body = body.collect().await?.aggregate(); let mut http_reader = http_body.reader(); let mut http_body_str = String::new(); http_reader.read_to_string(&mut http_body_str)?; @@ -94,7 +95,7 @@ impl HyperExtensions { } pub async fn decode_signed_response( - req: Request, + req: Request, signature_verifier: &SlackEventSignatureVerifier, ) -> AnyStdResult<(String, Parts)> { match ( diff --git a/src/hyper_tokio/listener/command_events.rs b/src/hyper_tokio/listener/command_events.rs index c4374739..7ebb6f05 100644 --- a/src/hyper_tokio/listener/command_events.rs +++ b/src/hyper_tokio/listener/command_events.rs @@ -8,10 +8,14 @@ use crate::hyper_tokio::*; pub use crate::models::events::*; pub use crate::models::SlackResponseUrl; use futures::future::{BoxFuture, FutureExt, TryFutureExt}; +use http_body_util::combinators::BoxBody; +use http_body_util::Empty; +use http_body_util::{BodyExt, Full}; use hyper::body::*; -use hyper::client::connect::Connect; use hyper::{Method, Request, Response, StatusCode}; +use hyper_util::client::legacy::connect::Connect; use std::collections::HashMap; +use std::convert::Infallible; use std::future::Future; use std::sync::Arc; @@ -25,18 +29,23 @@ impl SlackClientEventsHyperListener< SlackClientHyperConnector, >, ) -> impl Fn( - Request, + Request, D, ) -> BoxFuture< 'a, - Result, Box>, + Result>, Box>, > + 'a + Send + Clone where - D: Fn(Request) -> F + 'a + Send + Sync + Clone, - F: Future, Box>> + D: Fn(Request) -> F + 'a + Send + Sync + Clone, + F: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, R: Into>, @@ -48,7 +57,7 @@ impl SlackClientEventsHyperListener< let error_handler = self.environment.error_handler.clone(); let user_state_storage = self.environment.user_state.clone(); - move |req: Request, chain: D| { + move |req: Request, chain: D| { let cfg = config.clone(); let sign_verifier = signature_verifier.clone(); let sc = client.clone(); @@ -117,13 +126,16 @@ impl SlackClientEventsHyperListener< "application/json; charset=utf-8", ) .body( - serde_json::to_string(&cresp) - .unwrap() - .into(), + Full::new( + serde_json::to_string(&cresp) + .unwrap() + .into(), + ) + .boxed(), ), None => Response::builder() .status(StatusCode::OK) - .body(Body::empty()), + .body(Empty::new().boxed()), } .map_err(|e| e.into()), Err(err) => { @@ -134,7 +146,7 @@ impl SlackClientEventsHyperListener< ); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -147,7 +159,7 @@ impl SlackClientEventsHyperListener< ); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } diff --git a/src/hyper_tokio/listener/interaction_events.rs b/src/hyper_tokio/listener/interaction_events.rs index a5cfe753..36e4bc70 100644 --- a/src/hyper_tokio/listener/interaction_events.rs +++ b/src/hyper_tokio/listener/interaction_events.rs @@ -8,10 +8,13 @@ use crate::blocks::SlackViewSubmissionResponse; use crate::hyper_tokio::hyper_ext::HyperExtensions; use crate::hyper_tokio::*; use futures::future::{BoxFuture, FutureExt, TryFutureExt}; +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyExt, Empty}; use hyper::body::*; -use hyper::client::connect::Connect; use hyper::{Method, Request, Response, StatusCode}; +use hyper_util::client::legacy::connect::Connect; use std::collections::HashMap; +use std::convert::Infallible; use std::future::Future; use std::sync::Arc; @@ -25,18 +28,23 @@ impl SlackClientEventsHyperListener< SlackClientHyperConnector, >, ) -> impl Fn( - Request, + Request, D, ) -> BoxFuture< 'a, - Result, Box>, + Result>, Box>, > + 'a + Send + Clone where - D: Fn(Request) -> F + 'a + Send + Sync + Clone, - F: Future, Box>> + D: Fn(Request) -> F + 'a + Send + Sync + Clone, + F: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, R: SlackInteractionEventResponse, @@ -48,7 +56,7 @@ impl SlackClientEventsHyperListener< let error_handler = self.environment.error_handler.clone(); let user_state_storage = self.environment.user_state.clone(); - move |req: Request, chain: D| { + move |req: Request, chain: D| { let cfg = config.clone(); let sign_verifier = signature_verifier.clone(); let sc = client.clone(); @@ -90,7 +98,7 @@ impl SlackClientEventsHyperListener< let status_code = thread_error_handler(err, sc, thread_user_state_storage); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -103,7 +111,7 @@ impl SlackClientEventsHyperListener< let status_code = thread_error_handler(err, sc, thread_user_state_storage); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -112,7 +120,7 @@ impl SlackClientEventsHyperListener< let status_code = thread_error_handler(event_err, sc, thread_user_state_storage); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -128,11 +136,17 @@ impl SlackClientEventsHyperListener< } pub trait SlackInteractionEventResponse { - fn to_http_response(&self, event: &SlackInteractionEvent) -> AnyStdResult>; + fn to_http_response( + &self, + event: &SlackInteractionEvent, + ) -> AnyStdResult>>; } impl SlackInteractionEventResponse for () { - fn to_http_response(&self, event: &SlackInteractionEvent) -> AnyStdResult> { + fn to_http_response( + &self, + event: &SlackInteractionEvent, + ) -> AnyStdResult>> { match event { SlackInteractionEvent::ViewSubmission(_) => Response::builder() .status(StatusCode::OK) @@ -147,7 +161,10 @@ impl SlackInteractionEventResponse for () { } impl SlackInteractionEventResponse for SlackViewSubmissionResponse { - fn to_http_response(&self, _event: &SlackInteractionEvent) -> AnyStdResult> { + fn to_http_response( + &self, + _event: &SlackInteractionEvent, + ) -> AnyStdResult>> { let json_str = serde_json::to_string(&self)?; Ok(Response::builder() .status(StatusCode::OK) diff --git a/src/hyper_tokio/listener/mod.rs b/src/hyper_tokio/listener/mod.rs index b2c247bd..9d151b9f 100644 --- a/src/hyper_tokio/listener/mod.rs +++ b/src/hyper_tokio/listener/mod.rs @@ -1,17 +1,21 @@ -use crate::hyper_tokio::connector::SlackClientHyperConnector; - +use std::convert::Infallible; use std::future::Future; +use std::sync::Arc; +use bytes::Bytes; use futures::future::{BoxFuture, FutureExt}; -use hyper::client::connect::Connect; -use hyper::{Body, Request, Response}; +use http_body_util::combinators::BoxBody; +use hyper::body::Incoming; +use hyper::{Request, Response}; +use hyper_util::client::legacy::connect::Connect; +use crate::hyper_tokio::connector::SlackClientHyperConnector; use crate::listener::SlackClientEventsListenerEnvironment; + pub use command_events::*; pub use interaction_events::*; pub use oauth::*; pub use push_events::*; -use std::sync::Arc; mod command_events; mod interaction_events; @@ -34,20 +38,33 @@ pub fn chain_service_routes_fn<'a, R, D, FR, FD>( route: R, default: D, ) -> impl Fn( - Request, -) -> BoxFuture<'a, Result, Box>> + Request, +) -> BoxFuture< + 'a, + Result>, Box>, +> + 'a + Send + Clone where - R: Fn(Request, D) -> FR + 'a + Clone + Send, - D: Fn(Request) -> FD + 'a + Clone + Send, - FR: Future, Box>> + R: Fn(Request, D) -> FR + 'a + Clone + Send, + D: Fn(Request) -> FD + 'a + Clone + Send, + FR: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, - FD: Future, Box>> + FD: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, { - move |req: Request| route(req, default.clone()).boxed() + move |req: Request| route(req, default.clone()).boxed() } diff --git a/src/hyper_tokio/listener/oauth.rs b/src/hyper_tokio/listener/oauth.rs index ecd2dfea..1f62cbb7 100644 --- a/src/hyper_tokio/listener/oauth.rs +++ b/src/hyper_tokio/listener/oauth.rs @@ -1,5 +1,6 @@ use crate::hyper_tokio::connector::SlackClientHyperConnector; use crate::hyper_tokio::SlackClientEventsHyperListener; +use std::convert::Infallible; use crate::api::*; use crate::errors::*; @@ -8,9 +9,10 @@ use crate::{SlackClient, SlackClientHttpApiUri}; use crate::hyper_tokio::hyper_ext::HyperExtensions; use futures::future::{BoxFuture, FutureExt}; +use http_body_util::combinators::BoxBody; use hyper::body::*; -use hyper::client::connect::Connect; use hyper::{Method, Request, Response}; +use hyper_util::client::legacy::connect::Connect; use rvstruct::*; use std::future::Future; use std::sync::Arc; @@ -18,9 +20,10 @@ use tracing::*; impl SlackClientEventsHyperListener { pub(crate) async fn slack_oauth_install_service( - _: Request, + _: Request, config: &SlackOAuthListenerConfig, - ) -> Result, Box> { + ) -> Result>, Box> + { let full_uri = SlackClientHttpApiUri::create_url_with_params( SlackOAuthListenerConfig::OAUTH_AUTHORIZE_URL_VALUE, &vec![ @@ -37,7 +40,7 @@ impl SlackClientEventsHyperListener< } pub(crate) async fn slack_oauth_callback_service( - req: Request, + req: Request, config: &SlackOAuthListenerConfig, client: Arc>>, user_state_storage: SlackClientEventsUserState, @@ -47,8 +50,9 @@ impl SlackClientEventsHyperListener< SlackClientHyperConnector, >, error_handler: BoxedErrorHandler>, - ) -> Result, Box> { - let params = HyperExtensions::parse_query_params(&req); + ) -> Result>, Box> + { + let params = HyperExtensions::parse_query_params(req.uri()); debug!("Received Slack OAuth callback: {:?}", ¶ms); match (params.get("code"), params.get("error")) { @@ -127,18 +131,23 @@ impl SlackClientEventsHyperListener< SlackClientHyperConnector, >, ) -> impl Fn( - Request, + Request, D, ) -> BoxFuture< 'a, - Result, Box>, + Result>, Box>, > + 'a + Send + Clone where - D: Fn(Request) -> F + 'a + Send + Sync + Clone, - F: Future, Box>> + D: Fn(Request) -> F + 'a + Send + Sync + Clone, + F: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, { @@ -146,7 +155,7 @@ impl SlackClientEventsHyperListener< let listener_error_handler = self.environment.error_handler.clone(); let user_state_storage = self.environment.user_state.clone(); - move |req: Request, chain: D| { + move |req: Request, chain: D| { let cfg = config.clone(); let sc = client.clone(); let error_handler = listener_error_handler.clone(); diff --git a/src/hyper_tokio/listener/push_events.rs b/src/hyper_tokio/listener/push_events.rs index 30365846..ba886b99 100644 --- a/src/hyper_tokio/listener/push_events.rs +++ b/src/hyper_tokio/listener/push_events.rs @@ -6,9 +6,12 @@ use crate::listener::*; pub use crate::models::events::*; use crate::signature_verifier::SlackEventSignatureVerifier; use futures::future::{BoxFuture, FutureExt, TryFutureExt}; +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyExt, Empty, Full}; use hyper::body::*; -use hyper::client::connect::Connect; use hyper::{Method, Request, Response}; +use hyper_util::client::legacy::connect::Connect; +use std::convert::Infallible; use std::future::Future; use std::sync::Arc; use tracing::*; @@ -23,18 +26,23 @@ impl SlackClientEventsHyperListener< SlackClientHyperConnector, >, ) -> impl Fn( - Request, + Request, D, ) -> BoxFuture< 'a, - Result, Box>, + Result>, Box>, > + 'a + Send + Clone where - D: Fn(Request) -> F + 'a + Send + Sync + Clone, - F: Future, Box>> + D: Fn(Request) -> F + 'a + Send + Sync + Clone, + F: Future< + Output = Result< + Response>, + Box, + >, + > + 'a + Send, { @@ -45,7 +53,7 @@ impl SlackClientEventsHyperListener< let error_handler = self.environment.error_handler.clone(); let user_state_storage = self.environment.user_state.clone(); - move |req: Request, chain: D| { + move |req: Request, chain: D| { let cfg = config.clone(); let sign_verifier = signature_verifier.clone(); let sc = client.clone(); @@ -76,9 +84,9 @@ impl SlackClientEventsHyperListener< ) .await { - Ok(_) => { - Ok(Response::new(Body::from(url_ver.challenge))) - } + Ok(_) => Ok(Response::new( + Full::new(url_ver.challenge.into()).boxed(), + )), Err(err) => { let status_code = thread_error_handler( err, @@ -87,7 +95,7 @@ impl SlackClientEventsHyperListener< ); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -101,7 +109,7 @@ impl SlackClientEventsHyperListener< ) .await { - Ok(_) => Ok(Response::new(Body::empty())), + Ok(_) => Ok(Response::new(Empty::new().boxed())), Err(err) => { let status_code = thread_error_handler( err, @@ -110,7 +118,7 @@ impl SlackClientEventsHyperListener< ); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } } @@ -123,7 +131,7 @@ impl SlackClientEventsHyperListener< ); Response::builder() .status(status_code) - .body(Body::empty()) + .body(Empty::new().boxed()) .map_err(|e| e.into()) } }, diff --git a/src/hyper_tokio/mod.rs b/src/hyper_tokio/mod.rs index 3310ef14..342348b1 100644 --- a/src/hyper_tokio/mod.rs +++ b/src/hyper_tokio/mod.rs @@ -29,4 +29,5 @@ pub type SlackHyperClient = SlackClient; pub type SlackHyperListenerEnvironment = SlackClientEventsListenerEnvironment; -pub type SlackHyperHttpsConnector = hyper_rustls::HttpsConnector; +pub type SlackHyperHttpsConnector = + hyper_rustls::HttpsConnector; diff --git a/src/hyper_tokio/socket_mode/mod.rs b/src/hyper_tokio/socket_mode/mod.rs index 82dae6c4..f6543104 100644 --- a/src/hyper_tokio/socket_mode/mod.rs +++ b/src/hyper_tokio/socket_mode/mod.rs @@ -2,7 +2,7 @@ use crate::clients_manager::{SlackSocketModeClientsManager, SlackSocketModeClien use crate::hyper_tokio::connector::SlackClientHyperConnector; use crate::hyper_tokio::socket_mode::tokio_clients_manager::SlackSocketModeTokioClientsManager; use crate::listener::SlackClientEventsListenerEnvironment; -use hyper::client::connect::Connect; +use hyper_util::client::legacy::connect::Connect; use std::sync::Arc; mod tokio_clients_manager; diff --git a/src/hyper_tokio/socket_mode/tokio_clients_manager.rs b/src/hyper_tokio/socket_mode/tokio_clients_manager.rs index 46293afd..e5126036 100644 --- a/src/hyper_tokio/socket_mode/tokio_clients_manager.rs +++ b/src/hyper_tokio/socket_mode/tokio_clients_manager.rs @@ -1,6 +1,6 @@ use crate::*; use async_trait::async_trait; -use hyper::client::connect::Connect; +use hyper_util::client::legacy::connect::Connect; use std::sync::Arc; use crate::hyper_tokio::socket_mode::tungstenite_wss_client::SlackTungsteniteWssClient;