Discover

pub trait Discover: Sealed<Change<(), ()>> {
    /// A unique identifier for each active service.
    ///
    /// An identifier can be re-used once a [`Change::Remove`] has been yielded for its service.
    type Key: Eq;

    /// The type of [`Service`] yielded by this [`Discover`].
    ///
    /// [`Service`]: crate::Service
    type Service;

    /// Error produced during discovery
    type Error;

    /// Yields the next discovery change set.
    fn poll_discover(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Change<Self::Key, Self::Service>, Self::Error>>>;
}

impl<K, S, E, D: ?Sized> Discover for D
where
    D: TryStream<Ok = Change<K, S>, Error = E>,
    K: Eq,
{
    type Key = K;
    type Service = S;
    type Error = E;

    fn poll_discover(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<D::Ok, D::Error>>> {
        TryStream::try_poll_next(self, cx)
    }
}

#[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),
}

#[allow(unreachable_pub)]
#[cfg(any(feature = "balance", feature = "discover", feature = "make"))]
mod sealed {
    pub trait Sealed<T> {}
}

tower = { version = "=0.5.3", features = ["discover"] }

use std::{
  pin::Pin,
  task::{Context, Poll},
};
use tower::discover::{Change, Discover};

struct MyDiscover;

impl Discover for MyDiscover {
  type Key = u32;
  type Service = ();
  type Error = ();

  fn poll_discover(
    self: Pin<&mut Self>,
    _cx: &mut Context<'_>,
  ) -> Poll<Option<Result<Change<u32, ()>, ()>>> {
    Poll::Ready(None)
  }
}

fn main() {}

error[E0277]: the trait bound `MyDiscover: tower::sealed::Sealed<Change<(), ()>>` is not satisfied
  --> src/bin/sealed.rs:9:19
   |
 9 | impl Discover for MyDiscover {
   |                   ^^^^^^^^^^ unsatisfied trait bound
   |
help: the trait `futures_core::stream::Stream` is not implemented for `MyDiscover`
  --> src/bin/sealed.rs:7:1
   |
 7 | struct MyDiscover;
   | ^^^^^^^^^^^^^^^^^
   = help: the following other types implement trait `futures_core::stream::Stream`:
             &mut S
             AssertUnwindSafe<S>
             Box<S>
             CallAll<Svc, S>
             CallAllUnordered<Svc, S>
             Constant<D, M>
             PeakEwmaDiscover<D, C>
             PendingRequestsDiscover<D, C>
           and 84 others
   = note: required for `MyDiscover` to implement `futures_core::stream::TryStream`
   = note: required for `MyDiscover` to implement `tower::sealed::Sealed<Change<(), ()>>`
note: required by a bound in `tower::discover::Discover`
  --> ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tower-0.5.3/src/discover/mod.rs:55:21
   |
55 | pub trait Discover: Sealed<Change<(), ()>> {
   |                     ^^^^^^^^^^^^^^^^^^^^^^ required by this bound in `Discover`
   = note: `Discover` is a "sealed trait", because to implement it you also need to implement `tower::sealed::Sealed`, which is not accessible; this is usually done to force you to use one of the provided types that already implement it
   = help: the following types implement the trait:
             D
             S
             M

For more information about this error, try `rustc --explain E0277`.
error: could not compile `exp-tower-bala` (bin "sealed") due to 1 previous error

trait Stream { type Item; }
trait Service<R> { type Response; }
trait Sealed {}
impl<D: Stream> Sealed for D {}
impl<S: Service<u8>> Sealed for S {}
fn main() {}

error[E0119]: conflicting implementations of trait `Sealed`
 --> nogeneric.rs:5:1
  |
4 | impl<D: Stream> Sealed for D {}
  | ---------------------------- first implementation here
5 | impl<S: Service<u8>> Sealed for S {}
  | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ conflicting implementation

error: aborting due to 1 previous error

For more information about this error, try `rustc --explain E0119`.

    pub struct ServiceList<T>
    where
        T: IntoIterator,
    {
        inner: Enumerate<T::IntoIter>,
    }

tower = { version = "=0.5.3", features = ["balance"] }
tokio = { version = "=1.54.1", features = ["macros", "rt"] }
futures = "0.3"

use std::{
  convert::Infallible,
  future::poll_fn,
  pin::Pin,
  task::{Context, Poll, Waker},
};
use futures::channel::mpsc;
use tower::{
  balance::p2c::Balance,
  discover::{Change, Discover, ServiceList},
  load::Constant,
  service_fn, Service,
};

fn show<D>(name: &str, d: Pin<&mut D>)
where
  D: Discover,
  D::Key: std::fmt::Debug,
{
  let mut cx = Context::from_waker(Waker::noop());
  let s = match d.poll_discover(&mut cx) {
    Poll::Pending => "Pending".to_string(),
    Poll::Ready(None) => "Ready(None)".to_string(),
    Poll::Ready(Some(Ok(Change::Insert(k, _)))) => format!("Insert({k:?})"),
    Poll::Ready(Some(Ok(Change::Remove(k)))) => format!("Remove({k:?})"),
    Poll::Ready(Some(Err(_))) => "Err".to_string(),
  };
  println!("{name}: {s}");
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  // 1. ServiceList: a fixed list, keys are positions
  let svc = service_fn(|x: u32| async move { Ok::<_, Infallible>(x) });
  let mut list = ServiceList::new(vec![svc, svc, svc]);
  for _ in 0..5 {
    show("list", Pin::new(&mut list));
  }

  // 2. any TryStream of Change is a Discover: an mpsc receiver
  let (tx, mut rx) = mpsc::unbounded::<Result<Change<&str, _>, Infallible>>();
  tx.unbounded_send(Ok(Change::Insert("a", svc))).unwrap();
  tx.unbounded_send(Ok(Change::Insert("b", svc))).unwrap();
  show("chan", Pin::new(&mut rx));
  show("chan", Pin::new(&mut rx));
  show("chan", Pin::new(&mut rx));
  tx.unbounded_send(Ok(Change::Remove("a"))).unwrap();
  show("chan", Pin::new(&mut rx));
  drop(tx);
  show("chan", Pin::new(&mut rx));

  // 3. Balance consumes a Discover inside poll_ready
  let (tx, rx) = mpsc::unbounded::<Result<Change<&str, _>, Infallible>>();
  let mut lb = Balance::new(rx);
  for k in ["a", "b", "c"] {
    tx.unbounded_send(Ok(Change::Insert(k, Constant::new(svc, 0u8)))).unwrap();
  }
  println!("lb before poll_ready: len = {}", lb.len());
  poll_fn(|cx| lb.poll_ready(cx)).await.unwrap();
  println!("lb after poll_ready:  len = {}", lb.len());
  tx.unbounded_send(Ok(Change::Remove("b"))).unwrap();
  println!("lb Remove(b) sent:    len = {}", lb.len());
  poll_fn(|cx| lb.poll_ready(cx)).await.unwrap();
  println!("lb after poll_ready:  len = {}", lb.len());
  println!("lb.call(7) = {:?}", lb.call(7).await);
}

list: Insert(0)
list: Insert(1)
list: Insert(2)
list: Ready(None)
list: Ready(None)
chan: Insert("a")
chan: Insert("b")
chan: Pending
chan: Remove("a")
chan: Ready(None)
lb before poll_ready: len = 0
lb after poll_ready:  len = 3
lb Remove(b) sent:    len = 3
lb after poll_ready:  len = 2
lb.call(7) = Ok(7)