diff --git a/agent/src/main.rs b/agent/src/main.rs
index 4e2c4ff25..e09569094 100644
--- a/agent/src/main.rs
+++ b/agent/src/main.rs
@@ -59,10 +59,8 @@ async fn run() -> Result<()> {
 }
 
 use protocol::Parser;
-use std::sync::Arc;
-use stream::{Backend, Builder, Request};
-type Endpoint = Arc<Backend<Request>>;
-type Topology = endpoint::TopologyProtocol<Builder<Parser, Request>, Endpoint, Request, Parser>;
+use stream::Request;
+type Topology = endpoint::TopologyProtocol<Request, Parser>;
 async fn discovery_init(
     ctx: &'static Context,
     rx: Receiver<TopologyWriteGuard<Topology>>,
diff --git a/agent/src/service.rs b/agent/src/service.rs
index 8604c7c1f..cc936fbfe 100644
--- a/agent/src/service.rs
+++ b/agent/src/service.rs
@@ -9,10 +9,9 @@ use ds::chan::Sender;
 use metrics::Path;
 use protocol::{Parser, Result};
 use stream::pipeline::copy_bidirectional;
-use stream::{Backend, Builder, CheckedTopology, Request, StreamMetrics};
+use stream::{CheckedTopology, Request, StreamMetrics};
 
-type Endpoint = Arc<Backend<Request>>;
-type Topology = endpoint::TopologyProtocol<Builder<Parser, Request>, Endpoint, Request, Parser>;
+type Topology = endpoint::TopologyProtocol<Request, Parser>;
 // 一直侦听，直到成功侦听或者取消侦听（当前尚未支持取消侦听）
 // 1. 尝试侦听之前，先确保服务配置信息已经更新完成
 pub(super) async fn process_one(
diff --git a/endpoint/Cargo.toml b/endpoint/Cargo.toml
index 8f57ec0e8..411f13873 100644
--- a/endpoint/Cargo.toml
+++ b/endpoint/Cargo.toml
@@ -15,6 +15,8 @@ sharding = { path = "../sharding" }
 log = { path = "../log" }
 rt = { path = "../rt" }
 context = { path = "../context" }
+noop-waker = "0.1.0"
+metrics = { path = "../metrics" }
 
 byteorder = "1.4.3"
 bytes = "1.0.1"
diff --git a/stream/src/builder.rs b/endpoint/src/builder.rs
similarity index 76%
rename from stream/src/builder.rs
rename to endpoint/src/builder.rs
index 5dc30bd32..e03bb6c19 100644
--- a/stream/src/builder.rs
+++ b/endpoint/src/builder.rs
@@ -9,7 +9,7 @@ use ds::chan::mpsc::{channel, Sender, TrySendError};
 use ds::Switcher;
 
 use crate::checker::BackendChecker;
-use endpoint::{Builder, Endpoint, Single, Timeout};
+use crate::{Endpoint, Single, Timeout};
 use metrics::Path;
 use protocol::{Error, Protocol, Request, ResOption, Resource};
 
@@ -18,15 +18,26 @@ pub struct BackendBuilder<P, R> {
     _marker: std::marker::PhantomData<(P, R)>,
 }
 
-impl<P: Protocol, R: Request> Builder<P, R, Arc<Backend<R>>> for BackendBuilder<P, R> {
-    fn auth_option_build(
+pub type Backend<R> = Arc<InnerBackend<R>>;
+
+impl<P: Protocol, R: Request> BackendBuilder<P, R> {
+    pub fn build(
+        addr: &str,
+        parser: P,
+        rsrc: Resource,
+        service: &str,
+        timeout: Timeout,
+    ) -> Backend<R> {
+        Self::auth_option_build(addr, parser, rsrc, service, timeout, Default::default())
+    }
+    pub fn auth_option_build(
         addr: &str,
         parser: P,
         rsrc: Resource,
         service: &str,
         timeout: Timeout,
         option: ResOption,
-    ) -> Arc<Backend<R>> {
+    ) -> Backend<R> {
         let (tx, rx) = channel(256);
         let finish: Switcher = false.into();
         let init: Switcher = false.into();
@@ -38,7 +49,7 @@ impl<P: Protocol, R: Request> Builder<P, R, Arc<Backend<R>>> for BackendBuilder<
         let s = single.clone();
         rt::spawn(async move { checker.start_check(s).await });
 
-        Backend {
+        InnerBackend {
             finish,
             init,
             tx,
@@ -48,7 +59,7 @@ impl<P: Protocol, R: Request> Builder<P, R, Arc<Backend<R>>> for BackendBuilder<
     }
 }
 
-pub struct Backend<R> {
+pub struct InnerBackend<R> {
     single: Arc<AtomicBool>,
     tx: Sender<R>,
     // 实例销毁时，设置该值，通知checker，会议上check.
@@ -57,7 +68,7 @@ pub struct Backend<R> {
     init: Switcher,
 }
 
-impl<R> discovery::Inited for Backend<R> {
+impl<R> discovery::Inited for InnerBackend<R> {
     // 已经连接上或者至少连接了一次
     #[inline]
     fn inited(&self) -> bool {
@@ -65,13 +76,13 @@ impl<R> discovery::Inited for Backend<R> {
     }
 }
 
-impl<R> Drop for Backend<R> {
+impl<R> Drop for InnerBackend<R> {
     fn drop(&mut self) {
         self.finish.on();
     }
 }
 
-impl<R: Request> Endpoint for Backend<R> {
+impl<R: Request> Endpoint for InnerBackend<R> {
     type Item = R;
     #[inline]
     fn send(&self, req: R) {
@@ -84,7 +95,7 @@ impl<R: Request> Endpoint for Backend<R> {
         }
     }
 }
-impl<R> Single for Backend<R> {
+impl<R> Single for InnerBackend<R> {
     fn single(&self) -> bool {
         self.single.load(Acquire)
     }
diff --git a/endpoint/src/cacheservice/topo.rs b/endpoint/src/cacheservice/topo.rs
index 9eeec78c9..9cc6435aa 100644
--- a/endpoint/src/cacheservice/topo.rs
+++ b/endpoint/src/cacheservice/topo.rs
@@ -1,4 +1,4 @@
-use crate::{Builder, Endpoint, Topology};
+use crate::{Backend, Endpoint, Topology};
 use discovery::TopologyWrite;
 use protocol::{Protocol, Request, Resource, TryNextType};
 use sharding::hash::{Hash, HashKey, Hasher};
@@ -12,20 +12,20 @@ use crate::Timeout;
 use protocol::Bit;
 
 #[derive(Clone)]
-pub struct CacheService<B, E, Req, P> {
+pub struct CacheService<Req, P> {
     // 一共有n组，每组1个连接。
     // 排列顺序： master, master l1, slave, slave l1
-    streams: Distance<Shards<E, Req>>,
+    streams: Distance<Shards<Backend<Req>, Req>>,
     // streams里面的前r_num个数据是提供读的(这个长度不包含slave l1, slave)。
     hasher: Hasher,
     parser: P,
     exp_sec: u32,
     force_write_all: bool, // 兼容已有业务逻辑，set master失败后，是否更新其他layer
     backend_no_storage: bool, // true：mc后面没有存储
-    _marker: std::marker::PhantomData<(B, Req)>,
+    _marker: std::marker::PhantomData<Req>,
 }
 
-impl<B, E, Req, P> From<P> for CacheService<B, E, Req, P> {
+impl<Req, P> From<P> for CacheService<Req, P> {
     #[inline]
     fn from(parser: P) -> Self {
         Self {
@@ -40,10 +40,7 @@ impl<B, E, Req, P> From<P> for CacheService<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> discovery::Inited for CacheService<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for CacheService<Req, P> {
     #[inline]
     fn inited(&self) -> bool {
         self.streams.len() > 0
@@ -54,12 +51,10 @@ where
     }
 }
 
-impl<B, E, Req, P> Hash for CacheService<B, E, Req, P>
+impl<Req, P> Hash for CacheService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, k: &K) -> i64 {
@@ -67,12 +62,10 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for CacheService<B, E, Req, P>
+impl<Req, P> Topology for CacheService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn exp_sec(&self) -> u32 {
@@ -80,9 +73,8 @@ where
     }
 }
 
-impl<B: Send + Sync, E, Req, P> Endpoint for CacheService<B, E, Req, P>
+impl<Req, P> Endpoint for CacheService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -122,10 +114,7 @@ where
         unsafe { self.streams.get_unchecked(idx).send(req) };
     }
 }
-impl<B: Send + Sync, E, Req: Request, P: Protocol> CacheService<B, E, Req, P>
-where
-    E: Endpoint<Item = Req>,
-{
+impl<Req: Request, P: Protocol> CacheService<Req, P> {
     #[inline]
     fn context_store(&self, ctx: &mut super::Context, req: &Req) -> (usize, bool, bool) {
         let (idx, try_next, write_back);
@@ -190,11 +179,10 @@ where
         (idx, try_next, write_back)
     }
 }
-impl<B, E, Req, P> TopologyWrite for CacheService<B, E, Req, P>
+impl<Req, P> TopologyWrite for CacheService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req>,
+    Req: Request,
 {
     #[inline]
     fn update(&mut self, namespace: &str, cfg: &str) {
@@ -212,7 +200,7 @@ where
             let old = &mut streams;
 
             for shards in old_streams {
-                let group: Vec<(E, String)> = shards.into();
+                let group: Vec<(Backend<Req>, String)> = shards.into();
                 for e in group {
                     old.insert(e.1, e.0);
                 }
@@ -256,30 +244,35 @@ where
         v
     }
 }
-impl<B, E, Req, P> CacheService<B, E, Req, P>
+impl<Req, P> CacheService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req>,
+    Req: Request,
 {
     fn build(
         &self,
-        old: &mut HashMap<String, E>,
+        old: &mut HashMap<String, Backend<Req>>,
         addrs: Vec<String>,
         dist: &str,
         name: &str,
         timeout: Timeout,
-    ) -> Shards<E, Req> {
+    ) -> Shards<Backend<Req>, Req> {
         Shards::from(dist, addrs, |addr| {
             old.remove(addr).map(|e| e).unwrap_or_else(|| {
-                B::build(addr, self.parser.clone(), Resource::Memcache, name, timeout)
+                crate::BackendBuilder::build(
+                    addr,
+                    self.parser.clone(),
+                    Resource::Memcache,
+                    name,
+                    timeout,
+                )
             })
         })
     }
 }
 
 use std::fmt::{self, Display, Formatter};
-impl<B, E, Req, P> Display for CacheService<B, E, Req, P> {
+impl<Req, P> Display for CacheService<Req, P> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         write!(
             f,
diff --git a/stream/src/checker.rs b/endpoint/src/checker.rs
similarity index 98%
rename from stream/src/checker.rs
rename to endpoint/src/checker.rs
index 10f08da0f..85594c6f4 100644
--- a/stream/src/checker.rs
+++ b/endpoint/src/checker.rs
@@ -23,7 +23,7 @@ pub struct BackendChecker<P, Req> {
     init: Switcher,
     parser: P,
     addr: String,
-    timeout: endpoint::Timeout,
+    timeout: crate::Timeout,
     path: Path,
     option: ResOption,
 }
@@ -36,7 +36,7 @@ impl<P, Req> BackendChecker<P, Req> {
         init: Switcher,
         parser: P,
         path: Path,
-        timeout: endpoint::Timeout,
+        timeout: crate::Timeout,
         option: ResOption,
     ) -> Self {
         Self {
diff --git a/stream/src/handler.rs b/endpoint/src/handler.rs
similarity index 100%
rename from stream/src/handler.rs
rename to endpoint/src/handler.rs
diff --git a/endpoint/src/kv/topo.rs b/endpoint/src/kv/topo.rs
index 724d47235..bab3f40cc 100644
--- a/endpoint/src/kv/topo.rs
+++ b/endpoint/src/kv/topo.rs
@@ -17,7 +17,7 @@ use sharding::hash::{Hash, HashKey};
 use sharding::Distance;
 
 use crate::dns::DnsConfig;
-use crate::Builder;
+use crate::Backend;
 use crate::Single;
 use crate::Timeout;
 use crate::{Endpoint, Topology};
@@ -27,16 +27,16 @@ use super::config::Years;
 use super::strategy::Strategist;
 use super::KVCtx;
 #[derive(Clone)]
-pub struct KvService<B, E, Req, P> {
-    shards: Shards<E>,
+pub struct KvService<Req, P> {
+    shards: Shards<Backend<Req>>,
     // selector: Selector,
     strategist: Strategist,
     parser: P,
     cfg: Box<DnsConfig<KvNamespace>>,
-    _mark: std::marker::PhantomData<(B, Req)>,
+    _mark: std::marker::PhantomData<Req>,
 }
 
-impl<B, E, Req, P> From<P> for KvService<B, E, Req, P> {
+impl<Req, P> From<P> for KvService<Req, P> {
     #[inline]
     fn from(parser: P) -> Self {
         Self {
@@ -50,12 +50,10 @@ impl<B, E, Req, P> From<P> for KvService<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> Hash for KvService<B, E, Req, P>
+impl<Req, P> Hash for KvService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, k: &K) -> i64 {
@@ -63,18 +61,15 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for KvService<B, E, Req, P>
+impl<Req, P> Topology for KvService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
 }
 
-impl<B: Send + Sync, E, Req, P> Endpoint for KvService<B, E, Req, P>
+impl<Req, P> Endpoint for KvService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -157,11 +152,10 @@ where
     }
 }
 
-impl<B, E, Req, P> TopologyWrite for KvService<B, E, Req, P>
+impl<Req, P> TopologyWrite for KvService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     fn need_load(&self) -> bool {
         self.shards.len() != self.cfg.shards_url.len() || self.cfg.need_load()
@@ -176,23 +170,22 @@ where
         }
     }
 }
-impl<B, E, Req, P> KvService<B, E, Req, P>
+impl<Req, P> KvService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     // #[inline]
     fn take_or_build(
         &self,
-        old: &mut HashMap<String, Vec<E>>,
+        old: &mut HashMap<String, Vec<Backend<Req>>>,
         addr: &str,
         timeout: Timeout,
         res: ResOption,
-    ) -> E {
+    ) -> Backend<Req> {
         match old.get_mut(addr).map(|endpoints| endpoints.pop()) {
             Some(Some(end)) => end,
-            _ => B::auth_option_build(
+            _ => crate::BackendBuilder::auth_option_build(
                 &addr,
                 self.parser.clone(),
                 Resource::Mysql,
@@ -314,10 +307,7 @@ where
         true
     }
 }
-impl<B, E, Req, P> discovery::Inited for KvService<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for KvService<Req, P> {
     // 每一个域名都有对应的endpoint，并且都初始化完成。
     #[inline]
     fn inited(&self) -> bool {
diff --git a/endpoint/src/lib.rs b/endpoint/src/lib.rs
index 07c6e8267..e7da12993 100644
--- a/endpoint/src/lib.rs
+++ b/endpoint/src/lib.rs
@@ -2,12 +2,17 @@ mod shards;
 mod topo;
 pub use topo::*;
 
+mod builder;
 pub mod cacheservice;
+pub(crate) mod checker;
+pub mod handler;
 pub mod kv;
 pub mod msgque;
 pub mod phantomservice;
+mod reconn;
 pub mod redisservice;
 pub mod uuid;
+pub use builder::*;
 
 pub(crate) mod dns;
 
diff --git a/endpoint/src/msgque/topo.rs b/endpoint/src/msgque/topo.rs
index 614e2dc7a..cc66882cc 100644
--- a/endpoint/src/msgque/topo.rs
+++ b/endpoint/src/msgque/topo.rs
@@ -11,7 +11,7 @@ use tokio::time::Instant;
 
 use std::collections::{BTreeMap, HashMap};
 
-use crate::{Builder, Endpoint, Timeout, Topology};
+use crate::{Backend, Endpoint, Timeout, Topology};
 use sharding::hash::{Hash, HashKey, Hasher, Padding};
 
 use crate::msgque::strategy::hitfirst::Node;
@@ -31,15 +31,15 @@ const OFFLINE_STOP_READ_SECONDS: u64 = 60 * 20;
 const OFFLINE_CLEAN_SECONDS: u64 = OFFLINE_STOP_READ_SECONDS + 60 * 2;
 
 #[derive(Clone)]
-pub struct MsgQue<B, E, Req, P> {
+pub struct MsgQue<Req, P> {
     service: String,
 
     // 读写stream需要分开，读会有大量的空读
-    streams_read: Vec<(String, E, usize)>,
-    streams_write: BTreeMap<usize, Vec<(String, E)>>,
+    streams_read: Vec<(String, Backend<Req>, usize)>,
+    streams_write: BTreeMap<usize, Vec<(String, Backend<Req>)>>,
 
     // 轮询访问，N分钟后下线
-    streams_offline: Vec<(String, E)>,
+    streams_offline: Vec<(String, Backend<Req>)>,
     offline_hits: Arc<AtomicUsize>,
     offline_time: Instant,
 
@@ -51,10 +51,10 @@ pub struct MsgQue<B, E, Req, P> {
 
     timeout_write: Timeout,
     timeout_read: Timeout,
-    _marker: std::marker::PhantomData<(B, Req)>,
+    _marker: std::marker::PhantomData<Req>,
 }
 
-impl<B, E, Req, P> From<P> for MsgQue<B, E, Req, P> {
+impl<Req, P> From<P> for MsgQue<Req, P> {
     #[inline]
     fn from(parser: P) -> Self {
         Self {
@@ -75,10 +75,7 @@ impl<B, E, Req, P> From<P> for MsgQue<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> discovery::Inited for MsgQue<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for MsgQue<Req, P> {
     #[inline]
     fn inited(&self) -> bool {
         // check read streams
@@ -109,12 +106,10 @@ where
 
 const PADDING: Hasher = Hasher::Padding(Padding);
 
-impl<B, E, Req, P> Hash for MsgQue<B, E, Req, P>
+impl<Req, P> Hash for MsgQue<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, k: &K) -> i64 {
@@ -122,10 +117,8 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for MsgQue<B, E, Req, P>
+impl<Req, P> Topology for MsgQue<Req, P>
 where
-    B: Send + Sync,
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -138,10 +131,8 @@ where
 }
 
 //TODO: 验证的时候需要考虑512字节这种边界msg
-impl<B, E, Req, P> Endpoint for MsgQue<B, E, Req, P>
+impl<Req, P> Endpoint for MsgQue<Req, P>
 where
-    B: Send + Sync,
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -212,10 +203,8 @@ where
 }
 
 //获得待读取的streams和qid，返回的bool指示是否从offline streams中读取，true为都offline stream
-impl<B, E, Req, P> MsgQue<B, E, Req, P>
+impl<Req, P> MsgQue<Req, P>
 where
-    B: Send + Sync,
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -249,14 +238,16 @@ where
     }
 }
 
-impl<B, E, Req, P> MsgQue<B, E, Req, P>
+impl<Req, P> MsgQue<Req, P>
 where
-    B: Builder<P, Req, E>,
-    E: Endpoint<Item = Req>,
     P: Protocol,
+    Req: Request,
 {
     // 构建下线的队列
-    fn build_offline(&mut self, sized_queue: &BTreeMap<usize, Vec<String>>) -> Vec<(String, E)> {
+    fn build_offline(
+        &mut self,
+        sized_queue: &BTreeMap<usize, Vec<String>>,
+    ) -> Vec<(String, Backend<Req>)> {
         let mut new_addrs = HashSet::with_capacity(self.streams_read.len());
         let _ = sized_queue
             .iter()
@@ -265,7 +256,7 @@ where
             if !new_addrs.contains(name) {
                 self.streams_offline.push((
                     name.clone(),
-                    B::build(
+                    crate::BackendBuilder::build(
                         name,
                         self.parser.clone(),
                         Resource::MsgQue,
@@ -290,11 +281,11 @@ where
 
     fn build_read_streams(
         &self,
-        old: &mut HashMap<String, E>,
+        old: &mut HashMap<String, Backend<Req>>,
         addrs: &BTreeMap<usize, Vec<String>>,
         name: &str,
         timeout: Timeout,
-    ) -> Vec<(String, E, usize)> {
+    ) -> Vec<(String, Backend<Req>, usize)> {
         let mut streams = Vec::with_capacity(addrs.len());
         for (size, servs) in addrs.iter() {
             for s in servs.iter() {
@@ -306,7 +297,7 @@ where
             .map(|(srv, size)| {
                 (
                     srv.clone(),
-                    old.remove(srv).unwrap_or(B::build(
+                    old.remove(srv).unwrap_or(crate::BackendBuilder::build(
                         srv,
                         self.parser.clone(),
                         Resource::MsgQue,
@@ -324,7 +315,7 @@ where
         addrs: &BTreeMap<usize, Vec<String>>,
         name: &str,
         timeout: Timeout,
-    ) -> BTreeMap<usize, Vec<(String, E)>> {
+    ) -> BTreeMap<usize, Vec<(String, Backend<Req>)>> {
         let mut old_streams = HashMap::with_capacity(self.streams_write.len() * 3);
         if self.streams_write.len() > 0 {
             let mut first_key = 512;
@@ -348,13 +339,15 @@ where
                     .map(|addr| {
                         (
                             addr.clone(),
-                            old_streams.remove(addr).unwrap_or(B::build(
-                                addr,
-                                self.parser.clone(),
-                                Resource::MsgQue,
-                                name,
-                                timeout,
-                            )),
+                            old_streams
+                                .remove(addr)
+                                .unwrap_or(crate::BackendBuilder::build(
+                                    addr,
+                                    self.parser.clone(),
+                                    Resource::MsgQue,
+                                    name,
+                                    timeout,
+                                )),
                         )
                     })
                     .collect(),
@@ -385,11 +378,10 @@ where
     // }
 }
 
-impl<B, E, Req, P> TopologyWrite for MsgQue<B, E, Req, P>
+impl<Req, P> TopologyWrite for MsgQue<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req>,
+    Req: Request,
 {
     #[inline]
     fn update(&mut self, name: &str, cfg: &str) {
@@ -403,7 +395,7 @@ where
             self.timeout_write.adjust(ns.timeout_write);
 
             let old_r = self.streams_read.split_off(0);
-            let mut old_streams_read: HashMap<String, E> =
+            let mut old_streams_read: HashMap<String, Backend<Req>> =
                 old_r.into_iter().map(|(addr, e, _)| (addr, e)).collect();
 
             // 首先构建offline stream，如果前一次下线ips已经超时，先清理
diff --git a/endpoint/src/phantomservice/topo.rs b/endpoint/src/phantomservice/topo.rs
index 56e963cf3..aeedfcfd8 100644
--- a/endpoint/src/phantomservice/topo.rs
+++ b/endpoint/src/phantomservice/topo.rs
@@ -1,6 +1,6 @@
 use std::{collections::HashMap, marker::PhantomData};
 
-use crate::{dns::DnsConfig, Builder, Endpoint, Timeout, Topology};
+use crate::{dns::DnsConfig, Backend, Endpoint, Timeout, Topology};
 use discovery::{
     dns::{self, IPPort},
     TopologyWrite,
@@ -15,17 +15,17 @@ use sharding::{
 use super::config::PhantomNamespace;
 
 #[derive(Clone)]
-pub struct PhantomService<B, E, Req, P> {
+pub struct PhantomService<Req, P> {
     // 一般有2组，相互做HA，每组是一个域名列表，域名下只有一个ip，但会变化
-    streams: Vec<Distance<(String, E)>>,
+    streams: Vec<Distance<(String, Backend<Req>)>>,
     hasher: Crc32,
     distribution: Range,
     parser: P,
     cfg: Box<DnsConfig<PhantomNamespace>>,
-    _mark: PhantomData<(B, Req)>,
+    _mark: PhantomData<Req>,
 }
 
-impl<B, E, Req, P> From<P> for PhantomService<B, E, Req, P> {
+impl<Req, P> From<P> for PhantomService<Req, P> {
     fn from(parser: P) -> Self {
         Self {
             parser,
@@ -38,12 +38,10 @@ impl<B, E, Req, P> From<P> for PhantomService<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> Hash for PhantomService<B, E, Req, P>
+impl<Req, P> Hash for PhantomService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, k: &K) -> i64 {
@@ -51,21 +49,17 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for PhantomService<B, E, Req, P>
+impl<Req, P> Topology for PhantomService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
 }
 
-impl<B, E, Req, P> Endpoint for PhantomService<B, E, Req, P>
+impl<Req, P> Endpoint for PhantomService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     type Item = Req;
     #[inline]
@@ -92,11 +86,10 @@ where
     }
 }
 
-impl<B, E, Req, P> TopologyWrite for PhantomService<B, E, Req, P>
+impl<Req, P> TopologyWrite for PhantomService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req>,
+    Req: Request,
 {
     #[inline]
     fn update(&mut self, namespace: &str, cfg: &str) {
@@ -129,17 +122,21 @@ where
     }
 }
 
-impl<B, E, Req, P> PhantomService<B, E, Req, P>
+impl<Req, P> PhantomService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req>,
+    Req: Request,
 {
     #[inline]
-    fn take_or_build(&self, old: &mut HashMap<String, Vec<E>>, addr: &str, timeout: Timeout) -> E {
+    fn take_or_build(
+        &self,
+        old: &mut HashMap<String, Vec<Backend<Req>>>,
+        addr: &str,
+        timeout: Timeout,
+    ) -> Backend<Req> {
         match old.get_mut(addr).map(|endpoints| endpoints.pop()) {
             Some(Some(end)) => end,
-            _ => B::build(
+            _ => crate::BackendBuilder::build(
                 &addr,
                 self.parser.clone(),
                 Resource::Redis,
@@ -200,10 +197,7 @@ where
     }
 }
 
-impl<B, E, Req, P> discovery::Inited for PhantomService<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for PhantomService<Req, P> {
     // 每一个域名都有对应的endpoint，并且都初始化完成。
     #[inline]
     fn inited(&self) -> bool {
@@ -219,7 +213,7 @@ where
             })
     }
 }
-impl<B, E, Req, P> std::fmt::Debug for PhantomService<B, E, Req, P> {
+impl<Req, P> std::fmt::Debug for PhantomService<Req, P> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{:?}", self.cfg)
     }
diff --git a/stream/src/reconn.rs b/endpoint/src/reconn.rs
similarity index 100%
rename from stream/src/reconn.rs
rename to endpoint/src/reconn.rs
diff --git a/endpoint/src/redisservice/topo.rs b/endpoint/src/redisservice/topo.rs
index 7d55386c3..59571dc11 100644
--- a/endpoint/src/redisservice/topo.rs
+++ b/endpoint/src/redisservice/topo.rs
@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use crate::{Builder, Endpoint, Single, Topology};
+use crate::{Backend, Endpoint, Single, Topology};
 use discovery::TopologyWrite;
 use protocol::{Protocol, RedisFlager, Request, Resource};
 use sharding::distribution::Distribute;
@@ -12,16 +12,16 @@ use crate::{dns::DnsConfig, Timeout};
 use discovery::dns::{self, IPPort};
 
 #[derive(Clone)]
-pub struct RedisService<B, E, Req, P> {
+pub struct RedisService<Req, P> {
     // 一共shards.len()个分片，每个分片 shard[0]是master, shard[1..]是slave
-    shards: Vec<Shard<E>>,
+    shards: Vec<Shard<Backend<Req>>>,
     hasher: Hasher,
     distribute: Distribute,
     parser: P,
     cfg: Box<DnsConfig<RedisNamespace>>,
-    _mark: std::marker::PhantomData<(B, Req)>,
+    _mark: std::marker::PhantomData<Req>,
 }
-impl<B, E, Req, P> From<P> for RedisService<B, E, Req, P> {
+impl<Req, P> From<P> for RedisService<Req, P> {
     #[inline]
     fn from(parser: P) -> Self {
         Self {
@@ -35,12 +35,10 @@ impl<B, E, Req, P> From<P> for RedisService<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> Hash for RedisService<B, E, Req, P>
+impl<Req, P> Hash for RedisService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, k: &K) -> i64 {
@@ -48,18 +46,15 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for RedisService<B, E, Req, P>
+impl<Req, P> Topology for RedisService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
 }
 
-impl<B: Send + Sync, E, Req, P> Endpoint for RedisService<B, E, Req, P>
+impl<Req, P> Endpoint for RedisService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -129,11 +124,10 @@ where
         self.distribute.index(hash)
     }
 }
-impl<B, E, Req, P> TopologyWrite for RedisService<B, E, Req, P>
+impl<Req, P> TopologyWrite for RedisService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     #[inline]
     fn update(&mut self, namespace: &str, cfg: &str) {
@@ -157,10 +151,7 @@ where
         self.cfg.load_guard().check_load(|| self.load_inner());
     }
 }
-impl<B, E, Req, P> discovery::Inited for RedisService<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for RedisService<Req, P> {
     // 每一个域名都有对应的endpoint，并且都初始化完成。
     #[inline]
     fn inited(&self) -> bool {
@@ -173,25 +164,29 @@ where
                 .fold(true, |inited, shard| inited && shard.inited())
     }
 }
-impl<B, E, Req, P> RedisService<B, E, Req, P> {
+impl<Req, P> RedisService<Req, P> {
     #[inline]
     fn len(&self) -> usize {
         self.shards.len()
     }
 }
 
-impl<B, E, Req, P> RedisService<B, E, Req, P>
+impl<Req, P> RedisService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     #[inline]
-    fn take_or_build(&self, old: &mut HashMap<String, Vec<E>>, addr: &str, timeout: Timeout) -> E {
+    fn take_or_build(
+        &self,
+        old: &mut HashMap<String, Vec<Backend<Req>>>,
+        addr: &str,
+        timeout: Timeout,
+    ) -> Backend<Req> {
         let service = &self.cfg.service;
         match old.get_mut(addr).map(|endpoints| endpoints.pop()) {
             Some(Some(end)) => end,
-            _ => B::build(
+            _ => crate::BackendBuilder::build(
                 &addr,
                 self.parser.clone(),
                 Resource::Redis,
@@ -342,9 +337,8 @@ impl<E: discovery::Inited> Shard<E> {
                 .fold(true, |inited, (_, e)| inited && e.inited())
     }
 }
-impl<B: Send + Sync, E, Req, P> std::fmt::Display for RedisService<B, E, Req, P>
+impl<Req, P> std::fmt::Display for RedisService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
diff --git a/endpoint/src/topo.rs b/endpoint/src/topo.rs
index 72223172d..15b3cd9c5 100644
--- a/endpoint/src/topo.rs
+++ b/endpoint/src/topo.rs
@@ -1,11 +1,10 @@
 use std::io::{Error, ErrorKind, Result};
 
 use discovery::Inited;
-use protocol::{Protocol, ResOption, Resource};
+use protocol::Protocol;
 use sharding::hash::{Hash, HashKey};
 
 use crate::msgque::topo::MsgQue;
-use crate::Timeout;
 
 use enum_dispatch::enum_dispatch;
 
@@ -69,31 +68,31 @@ where
     }
 }
 
-pub trait Builder<P, R, E> {
-    fn build(addr: &str, parser: P, rsrc: Resource, service: &str, timeout: Timeout) -> E {
-        Self::auth_option_build(addr, parser, rsrc, service, timeout, Default::default())
-    }
+// pub trait Builder<P, R, E> {
+//     fn build(addr: &str, parser: P, rsrc: Resource, service: &str, timeout: Timeout) -> E {
+//         Self::auth_option_build(addr, parser, rsrc, service, timeout, Default::default())
+//     }
 
-    // TODO: ResOption -> AuthOption
-    fn auth_option_build(
-        addr: &str,
-        parser: P,
-        rsrc: Resource,
-        service: &str,
-        timeout: Timeout,
-        option: ResOption,
-    ) -> E;
-}
+//     // TODO: ResOption -> AuthOption
+//     fn auth_option_build(
+//         addr: &str,
+//         parser: P,
+//         rsrc: Resource,
+//         service: &str,
+//         timeout: Timeout,
+//         option: ResOption,
+//     ) -> E;
+// }
 
 macro_rules! define_topology {
     ($($top:ty, $item:ident, $ep:expr);+) => {
 
  #[derive(Clone)]
- pub enum TopologyProtocol<B, E, R, P> {
+ pub enum TopologyProtocol<R, P> {
       $($item($top)),+
  }
 
- impl<B, E, R, P> TopologyProtocol<B, E, R, P>  {
+ impl<R, P> TopologyProtocol<R, P>  {
      pub fn try_from(parser:P, endpoint:&str) -> Result<Self> {
           match &endpoint[..]{
               $($ep => Ok(Self::$item(parser.into())),)+
@@ -101,7 +100,7 @@ macro_rules! define_topology {
           }
      }
  }
- impl<B, E, R, P> Inited for TopologyProtocol<B, E, R, P> where E:Inited {
+ impl<R, P> Inited for TopologyProtocol<R, P> {
      #[inline]
      fn inited(&self) -> bool {
           match self {
@@ -112,7 +111,7 @@ macro_rules! define_topology {
      }
  }
 
-impl<B, E, R, P> discovery::TopologyWrite for TopologyProtocol<B, E, R, P> where P:Sync+Send+Protocol, B:Builder<P, R, E>, E:Endpoint<Item = R>+Single{
+impl<R, P> discovery::TopologyWrite for TopologyProtocol<R, P> where P:Sync+Send+Protocol, R:protocol::Request{
     #[inline]
     fn update(&mut self, name: &str, cfg: &str) {
         match self {
@@ -139,8 +138,8 @@ impl<B, E, R, P> discovery::TopologyWrite for TopologyProtocol<B, E, R, P> where
     }
 }
 
-impl<B:Send+Sync, E, R, P> Hash for TopologyProtocol<B, E, R, P>
-where P:Sync+Send+Protocol, E:Endpoint<Item = R>, R:protocol::Request{
+impl<R, P> Hash for TopologyProtocol<R, P>
+where P:Sync+Send+Protocol, R:protocol::Request{
     #[inline]
     fn hash<K:HashKey>(&self, k:&K) -> i64 {
         match self {
@@ -151,8 +150,8 @@ where P:Sync+Send+Protocol, E:Endpoint<Item = R>, R:protocol::Request{
     }
 }
 
-impl<B:Send+Sync, E, R, P> Topology for TopologyProtocol<B, E, R, P>
-where P:Sync+Send+Protocol, E:Endpoint<Item = R>, R:protocol::Request{
+impl<R, P> Topology for TopologyProtocol<R, P>
+where P:Sync+Send+Protocol, R:protocol::Request{
     #[inline]
     fn exp_sec(&self) -> u32 {
         match self {
@@ -163,11 +162,11 @@ where P:Sync+Send+Protocol, E:Endpoint<Item = R>, R:protocol::Request{
     }
 }
 
-impl<B, E, R, P> Endpoint for TopologyProtocol<B, E, R, P>
-where P:Sync+Send+Protocol, E:Endpoint<Item = R>,
+impl<R, P> Endpoint for TopologyProtocol<R, P>
+where P:Sync+Send+Protocol,
     R: protocol::Request,
     P: Protocol,
-    B:Send+Sync,
+
 {
     type Item = R;
     #[inline]
@@ -203,12 +202,12 @@ use crate::redisservice::topo::RedisService;
 use crate::uuid::topo::UuidService;
 
 define_topology! {
-    MsgQue<B, E, R, P>, MsgQue, "mq";
-    RedisService<B, E, R, P>, RedisService, "rs";
-    CacheService<B, E, R, P>, CacheService, "cs";
-    PhantomService<B, E, R, P>, PhantomService, "pt";
-    KvService<B, E, R, P>, KvService, "kv";
-    UuidService<B, E, R, P>, UuidService, "uuid"
+    MsgQue<R, P>, MsgQue, "mq";
+    RedisService<R, P>, RedisService, "rs";
+    CacheService<R, P>, CacheService, "cs";
+    PhantomService<R, P>, PhantomService, "pt";
+    KvService<R, P>, KvService, "kv";
+    UuidService<R, P>, UuidService, "uuid"
 }
 
 // 从环境变量获取是否开启后端资源访问的性能模式
diff --git a/endpoint/src/uuid/topo.rs b/endpoint/src/uuid/topo.rs
index b1107ba2f..f68f2836c 100644
--- a/endpoint/src/uuid/topo.rs
+++ b/endpoint/src/uuid/topo.rs
@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use crate::{Builder, Endpoint, Single, Topology};
+use crate::{Backend, Endpoint, Topology};
 use discovery::TopologyWrite;
 use protocol::{Protocol, Request, Resource};
 use sharding::{
@@ -13,13 +13,13 @@ use crate::{dns::DnsConfig, Timeout};
 use discovery::dns::{self, IPPort};
 
 #[derive(Clone)]
-pub struct UuidService<B, E, Req, P> {
-    shard: Distance<(String, E)>,
+pub struct UuidService<Req, P> {
+    shard: Distance<(String, Backend<Req>)>,
     parser: P,
     cfg: Box<DnsConfig<UuidNamespace>>,
-    _mark: std::marker::PhantomData<(B, Req)>,
+    _mark: std::marker::PhantomData<Req>,
 }
-impl<B, E, Req, P> From<P> for UuidService<B, E, Req, P> {
+impl<Req, P> From<P> for UuidService<Req, P> {
     #[inline]
     fn from(parser: P) -> Self {
         Self {
@@ -31,12 +31,10 @@ impl<B, E, Req, P> From<P> for UuidService<B, E, Req, P> {
     }
 }
 
-impl<B, E, Req, P> Hash for UuidService<B, E, Req, P>
+impl<Req, P> Hash for UuidService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
     #[inline]
     fn hash<K: HashKey>(&self, _k: &K) -> i64 {
@@ -44,18 +42,15 @@ where
     }
 }
 
-impl<B, E, Req, P> Topology for UuidService<B, E, Req, P>
+impl<Req, P> Topology for UuidService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
-    B: Send + Sync,
 {
 }
 
-impl<B: Send + Sync, E, Req, P> Endpoint for UuidService<B, E, Req, P>
+impl<Req, P> Endpoint for UuidService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
@@ -96,11 +91,10 @@ where
         0
     }
 }
-impl<B, E, Req, P> TopologyWrite for UuidService<B, E, Req, P>
+impl<Req, P> TopologyWrite for UuidService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     #[inline]
     fn update(&mut self, namespace: &str, cfg: &str) {
@@ -118,10 +112,7 @@ where
         self.cfg.load_guard().check_load(|| self.load_inner());
     }
 }
-impl<B, E, Req, P> discovery::Inited for UuidService<B, E, Req, P>
-where
-    E: discovery::Inited,
-{
+impl<Req, P> discovery::Inited for UuidService<Req, P> {
     #[inline]
     fn inited(&self) -> bool {
         self.shard.len() > 0
@@ -132,18 +123,28 @@ where
     }
 }
 
-impl<B, E, Req, P> UuidService<B, E, Req, P>
+impl<Req, P> UuidService<Req, P>
 where
-    B: Builder<P, Req, E>,
     P: Protocol,
-    E: Endpoint<Item = Req> + Single,
+    Req: Request,
 {
     #[inline]
-    fn take_or_build(&self, old: &mut HashMap<String, Vec<E>>, addr: &str, timeout: Timeout) -> E {
+    fn take_or_build(
+        &self,
+        old: &mut HashMap<String, Vec<Backend<Req>>>,
+        addr: &str,
+        timeout: Timeout,
+    ) -> Backend<Req> {
         let service = &self.cfg.service;
         match old.get_mut(addr).map(|endpoints| endpoints.pop()) {
             Some(Some(end)) => end,
-            _ => B::build(&addr, self.parser.clone(), Resource::Uuid, service, timeout),
+            _ => crate::BackendBuilder::build(
+                &addr,
+                self.parser.clone(),
+                Resource::Uuid,
+                service,
+                timeout,
+            ),
         }
     }
 
@@ -198,9 +199,8 @@ where
     }
 }
 
-impl<B: Send + Sync, E, Req, P> std::fmt::Display for UuidService<B, E, Req, P>
+impl<Req, P> std::fmt::Display for UuidService<Req, P>
 where
-    E: Endpoint<Item = Req>,
     Req: Request,
     P: Protocol,
 {
diff --git a/stream/Cargo.toml b/stream/Cargo.toml
index 74cab0ae9..1247349b3 100644
--- a/stream/Cargo.toml
+++ b/stream/Cargo.toml
@@ -17,7 +17,6 @@ rt = { path = "../rt" }
 log = { path = "../log" }
 
 enum_dispatch = "0.3.8"
-noop-waker = "0.1.0"
 array-init = "2"
 ctor = "0.1.23"
 
diff --git a/stream/src/lib.rs b/stream/src/lib.rs
index 18e9e9433..53c8c4fa5 100644
--- a/stream/src/lib.rs
+++ b/stream/src/lib.rs
@@ -1,9 +1,7 @@
 //pub mod buffer;
-pub mod handler;
 pub mod pipeline;
 pub use protocol::callback::*;
 pub use protocol::request::*;
-mod reconn;
 
 mod context;
 
@@ -11,12 +9,6 @@ pub trait Read {
     fn consume<Out, C: Fn(&[u8]) -> (usize, Out)>(&mut self, c: C) -> Out;
 }
 
-mod builder;
-pub use builder::BackendBuilder as Builder;
-pub use builder::*;
-
-pub(crate) mod checker;
-
 mod metric;
 pub use metric::StreamMetrics;
 
diff --git a/tests/src/discovery.rs b/tests/src/discovery.rs
index c58341dc9..50a9b107b 100644
--- a/tests/src/discovery.rs
+++ b/tests/src/discovery.rs
@@ -7,7 +7,7 @@ use protocol::Parser;
 fn refresh() {
     use stream::{Backend, Builder, Request};
     type Endpoint = Arc<Backend<Request>>;
-    type Topology = endpoint::TopologyProtocol<Builder<Parser, Request>, Endpoint, Request, Parser>;
+    type Topology = endpoint::TopologyProtocol<Endpoint, Request, Parser>;
     let service = "redisservice";
     let p = Parser::try_from("redis").unwrap();
     let top: Topology = endpoint::TopologyProtocol::try_from(p.clone(), "rs").unwrap();
diff --git a/tests/src/layout.rs b/tests/src/layout.rs
index 3e904ab46..749a0c792 100644
--- a/tests/src/layout.rs
+++ b/tests/src/layout.rs
@@ -1,10 +1,10 @@
 // 不要轻易变更这里面的测试用例，除非你知道你在做什么。拉相关同学进行方案评审。
-use std::{mem::size_of, sync::Arc};
+use std::mem::size_of;
 
+use endpoint::InnerBackend;
 use protocol::{callback::CallbackContext, Parser};
-use stream::{Backend, Request};
-type Endpoint = Arc<Backend<Request>>;
-type Topology = endpoint::TopologyProtocol<Builder, Endpoint, Request, Parser>;
+use stream::Request;
+type Topology = endpoint::TopologyProtocol<Request, Parser>;
 //type RefreshTopology = endpoint::RefreshTopology<Topology>;
 
 type CheckedTopology = stream::CheckedTopology<Topology>;
@@ -12,14 +12,12 @@ type CheckedTopology = stream::CheckedTopology<Topology>;
 type CopyBidirectional = stream::pipeline::CopyBidirectional<Stream, Parser, CheckedTopology>;
 
 type Stream = rt::Stream<tokio::net::TcpStream>;
-type Handler<'r> = stream::handler::Handler<'r, Request, Parser, Stream>;
+type Handler<'r> = endpoint::handler::Handler<'r, Request, Parser, Stream>;
 
-type Builder = stream::Builder<Parser, Request>;
-type CacheService = endpoint::cacheservice::topo::CacheService<Builder, Endpoint, Request, Parser>;
-type RedisService = endpoint::redisservice::topo::RedisService<Builder, Endpoint, Request, Parser>;
-type PhantomService =
-    endpoint::phantomservice::topo::PhantomService<Builder, Endpoint, Request, Parser>;
-type MsgQue = endpoint::msgque::topo::MsgQue<Builder, Endpoint, Request, Parser>;
+type CacheService = endpoint::cacheservice::topo::CacheService<Request, Parser>;
+type RedisService = endpoint::redisservice::topo::RedisService<Request, Parser>;
+type PhantomService = endpoint::phantomservice::topo::PhantomService<Request, Parser>;
+type MsgQue = endpoint::msgque::topo::MsgQue<Request, Parser>;
 
 use rt::Entry;
 
@@ -39,8 +37,7 @@ fn checkout_basic() {
     assert_eq!(16, size_of::<metrics::Metric>());
     assert_eq!(64, size_of::<metrics::Item>());
     assert_eq!(1, size_of::<Parser>());
-    assert_eq!(48, size_of::<Backend<Request>>());
-    assert_eq!(0, size_of::<Builder>());
+    assert_eq!(48, size_of::<InnerBackend<Request>>());
     assert_eq!(40, size_of::<CheckedTopology>());
     assert_eq!(368, size_of::<stream::StreamMetrics>());
     assert_eq!(24, size_of::<sharding::hash::Hasher>());