feat(channel): Add Change type to make tower internal dependency

hyperium · Sep 24, 2024 · 5568ea2 · 5568ea2
1 parent 6d93c1d
commit 5568ea2
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 11 deletions.
diff --git a/examples/src/dynamic_load_balance/client.rs b/examples/src/dynamic_load_balance/client.rs
@@ -3,15 +3,14 @@ pub mod pb {
 }
 
 use pb::{echo_client::EchoClient, EchoRequest};
+use tonic::transport::channel::Change;
 use tonic::transport::Channel;
-
 use tonic::transport::Endpoint;
 
 use std::sync::Arc;
 
 use std::sync::atomic::{AtomicBool, Ordering::SeqCst};
 use tokio::time::timeout;
-use tower::discover::Change;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {

diff --git a/tonic/Cargo.toml b/tonic/Cargo.toml
@@ -133,7 +133,6 @@ allowed_external_types = [
   "futures_core::stream::Stream",
   "h2::error::Error",
   "http_body_util::combinators::box_body::UnsyncBoxBody",
-  "tower::discover::Change",
   "tower_service::Service",
   "tower_layer::Layer",
   "tower_layer::stack::Stack",

diff --git a/tonic/src/transport/channel/mod.rs b/tonic/src/transport/channel/mod.rs
@@ -5,6 +5,7 @@ pub(crate) mod service;
 #[cfg(feature = "tls")]
 mod tls;
 
+pub use self::service::Change;
 pub use endpoint::Endpoint;
 #[cfg(feature = "tls")]
 pub use tls::ClientTlsConfig;
@@ -30,7 +31,7 @@ use hyper::rt;
 use tower::balance::p2c::Balance;
 use tower::{
     buffer::{self, Buffer},
-    discover::{Change, Discover},
+    discover::Discover,
     util::{BoxService, Either},
     Service,
 };

diff --git a/tonic/src/transport/channel/service/discover.rs b/tonic/src/transport/channel/service/discover.rs
@@ -7,11 +7,17 @@ use std::{
     task::{Context, Poll},
 };
 use tokio::sync::mpsc::Receiver;
-
 use tokio_stream::Stream;
-use tower::discover::Change;
-
-type DiscoverResult<K, S, E> = Result<Change<K, S>, E>;
+use tower::discover::Change as TowerChange;
+
+/// A change in the service set.
+#[derive(Debug, Clone)]
+pub enum Change<K, V> {
+    /// A new service identified by key `K` was identified.
+    Insert(K, V),
+    /// The service identified by key `K` disappeared.
+    Remove(K),
+}
 
 pub(crate) struct DynamicServiceStream<K: Hash + Eq + Clone> {
     changes: Receiver<Change<K, Endpoint>>,
@@ -24,7 +30,7 @@ impl<K: Hash + Eq + Clone> DynamicServiceStream<K> {
 }
 
 impl<K: Hash + Eq + Clone> Stream for DynamicServiceStream<K> {
-    type Item = DiscoverResult<K, Connection, crate::Error>;
+    type Item = Result<TowerChange<K, Connection>, crate::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let c = &mut self.changes;
@@ -39,10 +45,10 @@ impl<K: Hash + Eq + Clone> Stream for DynamicServiceStream<K> {
                     http.enforce_http(false);
 
                     let connection = Connection::lazy(endpoint.connector(http), endpoint);
-                    let change = Ok(Change::Insert(k, connection));
+                    let change = Ok(TowerChange::Insert(k, connection));
                     Poll::Ready(Some(change))
                 }
-                Change::Remove(k) => Poll::Ready(Some(Ok(Change::Remove(k)))),
+                Change::Remove(k) => Poll::Ready(Some(Ok(TowerChange::Remove(k)))),
             },
         }
     }

diff --git a/tonic/src/transport/channel/service/mod.rs b/tonic/src/transport/channel/service/mod.rs
@@ -11,6 +11,7 @@ mod connection;
 pub(super) use self::connection::Connection;
 
 mod discover;
+pub use self::discover::Change;
 pub(super) use self::discover::DynamicServiceStream;
 
 mod io;