Move ProxyConnector to its own crate

Cargo.toml

@@ -63,6 +63,7 @@ restate_common = { path = "src/common" }
 restate_consensus = { path = "src/consensus" }
 restate_errors = { path = "src/errors" }
 restate_futures_util = { path = "src/futures_util" }
+restate_hyper_util = { path = "src/hyper_util" }
 restate_ingress_grpc = { path = "src/ingress_grpc" }
 restate_queue = { path = "src/queue"}
 restate_invoker = { path = "src/invoker" }

src/common/Cargo.toml

@@ -22,12 +22,10 @@ schemars = { workspace = true, optional = true }
 serde = { workspace = true, optional = true }
 serde_with = { workspace = true, optional = true }
 humantime = { workspace = true }
-hyper = { workspace = true }
 uuid = { workspace = true }
 opentelemetry_api = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, default-features = false, features = ["time", "sync"]}
-tower = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 xxhash-rust = { version = "0.8", features = ["xxh3"] }

src/common/src/lib.rs

@@ -1,6 +1,5 @@
 pub mod errors;
 pub mod partitioner;
-pub mod proxy_connector;
 pub mod retry_policy;
 pub mod traits;
 pub mod types;

src/hyper_util/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "restate_hyper_util"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+publish = false
+
+[features]
+default = []
+
+[dependencies]
+hyper = { workspace = true }
+thiserror = { workspace = true }

src/hyper_util/src/lib.rs

@@ -0,0 +1 @@
+pub mod proxy_connector;

src/common/src/proxy_connector.rs → src/hyper_util/src/proxy_connector.rs

@@ -4,8 +4,8 @@ use std::str::FromStr;
 use std::task::{Context, Poll};
 
 use hyper::http::uri::{InvalidUri, Parts, Scheme};
+use hyper::service::Service;
 use hyper::Uri;
-use tower::Service;
 
 #[derive(Clone, Debug, thiserror::Error)]
 #[error("invalid proxy Uri (must have scheme, authority, and path): {0}")]
@@ -23,15 +23,15 @@ impl Display for Proxy {
 }
 
 #[derive(Debug, thiserror::Error)]
-pub enum ProxyFromStrErr {
+pub enum ProxyFromStrError {
     #[error(transparent)]
     InvalidUri(#[from] InvalidUri),
     #[error(transparent)]
     InvalidProxyUri(#[from] InvalidProxyUri),
 }
 
 impl FromStr for Proxy {
-    type Err = ProxyFromStrErr;
+    type Err = ProxyFromStrError;
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         Ok(Self::new(Uri::from_str(s)?)?)
     }

src/invoker/Cargo.toml

@@ -15,6 +15,7 @@ options_schema = ["dep:schemars", "restate_common/options_schema"]
 bytes = { workspace = true }
 bytes-utils = { workspace = true }
 restate_common = { workspace = true, features = ["serde"] }
+restate_hyper_util = { workspace = true }
 drain = { workspace = true }
 futures = { workspace = true }
 hyper = { workspace = true, features = ["http1", "http2", "client", "tcp", "stream", "runtime"] }

src/invoker/src/invoker.rs

@@ -7,8 +7,8 @@ use std::{cmp, panic};
 
 use futures::stream;
 use futures::stream::{PollNext, StreamExt};
-use restate_common::proxy_connector::{Proxy, ProxyConnector};
 use restate_common::types::PartitionLeaderEpoch;
+use restate_hyper_util::proxy_connector::{Proxy, ProxyConnector};
 use restate_journal::raw::{PlainRawEntry, RawEntryCodec};
 use restate_service_metadata::ServiceEndpointRegistry;
 use restate_timer_queue::TimerQueue;

src/meta/Cargo.toml

@@ -33,6 +33,7 @@ schemars = { workspace = true }
 
 # Service discovery and metadata
 restate_common = { workspace = true, features = ["serde"] }
+restate_hyper_util = { workspace = true }
 restate_service_key_extractor = { workspace = true, features = ["json"] }
 restate_service_metadata = { workspace = true, features = ["serde", "serde_schema"] }
 restate_service_protocol = { workspace = true, features = ["discovery"] }

src/meta/src/lib.rs

@@ -4,9 +4,9 @@ mod storage;
 
 use codederror::CodedError;
 use rest_api::MetaRestEndpoint;
-use restate_common::proxy_connector::Proxy;
 use restate_common::retry_policy::RetryPolicy;
 use restate_common::worker_command::WorkerCommandSender;
+use restate_hyper_util::proxy_connector::Proxy;
 use restate_ingress_grpc::ReflectionRegistry;
 use restate_service_key_extractor::KeyExtractorsRegistry;
 use restate_service_metadata::{InMemoryMethodDescriptorRegistry, InMemoryServiceEndpointRegistry};

src/meta/src/service.rs

@@ -7,10 +7,10 @@ use std::future::Future;
 use hyper::http::{HeaderName, HeaderValue};
 use hyper::Uri;
 use prost_reflect::DescriptorPool;
-use restate_common::proxy_connector::Proxy;
 use restate_common::retry_policy::RetryPolicy;
 use restate_errors::{error_it, warn_it};
 use restate_futures_util::command::{Command, UnboundedCommandReceiver, UnboundedCommandSender};
+use restate_hyper_util::proxy_connector::Proxy;
 use restate_ingress_grpc::{ReflectionRegistry, RegistrationError};
 use restate_service_key_extractor::KeyExtractorsRegistry;
 use restate_service_metadata::{

src/service_protocol/Cargo.toml

@@ -12,7 +12,7 @@ default = []
 protocol = []
 message = ["protocol", "dep:restate_common", "dep:bytes-utils", "dep:codederror", "dep:restate_errors", "dep:size", "dep:tracing"]
 codec = ["protocol", "dep:restate_common", "dep:thiserror", "dep:restate_journal", "dep:paste"]
-discovery = ["dep:thiserror", "dep:codederror", "dep:restate_errors", "dep:restate_service_key_extractor", "dep:hyper", "dep:hyper-rustls", "dep:prost-reflect", "dep:restate_common", "dep:restate_service_metadata"]
+discovery = ["dep:thiserror", "dep:codederror", "dep:restate_errors", "dep:restate_service_key_extractor", "dep:hyper", "dep:hyper-rustls", "dep:prost-reflect", "dep:restate_common", "dep:restate_hyper_util", "dep:restate_service_metadata"]
 
 [dependencies]
 bytes = { workspace = true }
@@ -34,6 +34,7 @@ prost-reflect = { workspace = true, optional = true }
 restate_service_metadata = { workspace = true, optional = true }
 codederror = { workspace = true, optional = true }
 restate_errors = { workspace = true, optional = true }
+restate_hyper_util = { workspace = true, optional = true }
 
 size = { version = "0.4.1", optional = true }
 

src/service_protocol/src/discovery.rs

@@ -11,9 +11,9 @@ use prost_reflect::{
     DescriptorError, DescriptorPool, ExtensionDescriptor, FieldDescriptor, Kind, MethodDescriptor,
     ServiceDescriptor,
 };
-use restate_common::proxy_connector::{Proxy, ProxyConnector};
 use restate_common::retry_policy::RetryPolicy;
 use restate_errors::{META0001, META0002, META0003};
+use restate_hyper_util::proxy_connector::{Proxy, ProxyConnector};
 use restate_service_key_extractor::{KeyStructure, ServiceInstanceType};
 use restate_service_metadata::ProtocolType;