router: Use NewService instead of MakeService (#724)

There's no reason for `NewRouter` to operate over `MakeService`. This
change makes it generic over a `NewService` instead, and moves the
implementation into the stack subcrate.

Cargo.lock

@@ -827,7 +827,6 @@ dependencies = [
  "linkerd2-proxy-transport",
  "linkerd2-reconnect",
  "linkerd2-retry",
- "linkerd2-router",
  "linkerd2-service-profiles",
  "linkerd2-stack",
  "linkerd2-stack-metrics",
@@ -1426,18 +1425,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "linkerd2-router"
-version = "0.1.0"
-dependencies = [
- "futures 0.3.5",
- "linkerd2-error",
- "linkerd2-stack",
- "pin-project",
- "tower",
- "tracing",
-]
-
 [[package]]
 name = "linkerd2-service-profiles"
 version = "0.1.0"

Cargo.toml

@@ -42,7 +42,6 @@ members = [
     "linkerd/proxy/transport",
     "linkerd/reconnect",
     "linkerd/retry",
-    "linkerd/router",
     "linkerd/service-profiles",
     "linkerd/signal",
     "linkerd/stack",

linkerd/app/core/Cargo.toml

@@ -52,7 +52,6 @@ linkerd2-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd2-proxy-transport = { path = "../../proxy/transport" }
 linkerd2-reconnect = { path = "../../reconnect" }
 linkerd2-retry = { path = "../../retry" }
-linkerd2-router = { path = "../../router" }
 linkerd2-timeout = { path = "../../timeout" }
 linkerd2-tracing = { path = "../../tracing" }
 linkerd2-service-profiles = { path = "../../service-profiles" }

linkerd/app/core/src/lib.rs

@@ -19,7 +19,6 @@ pub use linkerd2_exp_backoff as exp_backoff;
 pub use linkerd2_http_metrics as http_metrics;
 pub use linkerd2_opencensus as opencensus;
 pub use linkerd2_reconnect as reconnect;
-pub use linkerd2_router as router;
 pub use linkerd2_service_profiles as profiles;
 pub use linkerd2_stack_metrics as stack_metrics;
 pub use linkerd2_stack_tracing as stack_tracing;

linkerd/app/inbound/src/endpoint.rs

@@ -3,7 +3,7 @@ use linkerd2_app_core::{
     classify, dst, http_request_authority_addr, http_request_host_addr,
     http_request_l5d_override_dst_addr, metrics, profiles,
     proxy::{http, identity, tap},
-    router, stack_tracing,
+    stack_tracing, svc,
     transport::{self, listen, tls},
     Addr, Conditional, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
 };
@@ -296,7 +296,7 @@ impl From<TcpAccept> for RequestTarget {
     }
 }
 
-impl<A> router::Recognize<http::Request<A>> for RequestTarget {
+impl<A> svc::stack::RecognizeRoute<http::Request<A>> for RequestTarget {
     type Key = Target;
 
     fn recognize(&self, req: &http::Request<A>) -> Self::Key {

linkerd/app/inbound/src/lib.rs

@@ -22,7 +22,7 @@ use linkerd2_app_core::{
         http::{self, orig_proto, strip_header},
         identity, tap, tcp,
     },
-    reconnect, router,
+    reconnect,
     spans::SpanConverter,
     svc::{self},
     transport::{self, io, listen, tls},
@@ -351,8 +351,9 @@ impl Config {
             // Routes each request to a target, obtains a service for that
             // target, and dispatches the request.
             .instrument_from_target()
-            .into_make_service()
-            .push(router::Layer::new(RequestTarget::from))
+            .push(svc::layer::mk(|inner| {
+                svc::stack::NewRouter::new(RequestTarget::from, inner)
+            }))
             // Used by tap.
             .push_http_insert_target()
             .check_new_service::<TcpAccept, http::Request<_>>()

linkerd/stack/src/lib.rs

@@ -15,6 +15,7 @@ mod oneshot;
 mod proxy;
 mod request_filter;
 mod result;
+pub mod router;
 mod switch;
 mod switch_ready;
 
@@ -30,6 +31,7 @@ pub use self::oneshot::{Oneshot, OneshotLayer};
 pub use self::proxy::{Proxy, ProxyService};
 pub use self::request_filter::{FilterRequest, RequestFilter};
 pub use self::result::ResultService;
+pub use self::router::{NewRouter, RecognizeRoute};
 pub use self::switch::{MakeSwitch, Switch};
 pub use self::switch_ready::{NewSwitchReady, SwitchReady};
 pub use tower::util::Either;

linkerd/stack/src/router.rs

@@ -0,0 +1,77 @@
+use crate::NewService;
+use std::task::{Context, Poll};
+use tower::util::{Oneshot, ServiceExt};
+
+pub trait RecognizeRoute<T> {
+    type Key: Clone;
+
+    fn recognize(&self, t: &T) -> Self::Key;
+}
+
+#[derive(Clone, Debug)]
+pub struct NewRouter<T, N> {
+    new_recgonize: T,
+    inner: N,
+}
+
+#[derive(Clone, Debug)]
+pub struct Router<T, N> {
+    recognize: T,
+    inner: N,
+}
+
+impl<K, N> NewRouter<K, N> {
+    pub fn new(new_recgonize: K, inner: N) -> Self {
+        Self {
+            new_recgonize,
+            inner,
+        }
+    }
+}
+
+impl<T, K, N> NewService<T> for NewRouter<K, N>
+where
+    K: NewService<T>,
+    N: Clone,
+{
+    type Service = Router<K::Service, N>;
+
+    fn new_service(&mut self, t: T) -> Self::Service {
+        Router {
+            recognize: self.new_recgonize.new_service(t),
+            inner: self.inner.clone(),
+        }
+    }
+}
+
+impl<K, N, S, Req> tower::Service<Req> for Router<K, N>
+where
+    K: RecognizeRoute<Req>,
+    N: NewService<K::Key, Service = S>,
+    S: tower::Service<Req>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = Oneshot<S, Req>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: Req) -> Self::Future {
+        let key = self.recognize.recognize(&req);
+        self.inner.new_service(key).oneshot(req)
+    }
+}
+
+impl<T, K, F> RecognizeRoute<T> for F
+where
+    K: Clone,
+    F: Fn(&T) -> K,
+{
+    type Key = K;
+
+    fn recognize(&self, t: &T) -> Self::Key {
+        (self)(t)
+    }
+}