use std::net; use std::time::Duration; use futures::future::{err, ok, FutureResult}; use futures::{Future, Poll}; use tokio_current_thread::spawn; use tokio_reactor::Handle; use tokio_tcp::TcpStream; use counter::CounterGuard; use service::{NewService, Service}; /// Server message pub enum ServerMessage { /// New stream Connect(net::TcpStream), /// Gracefull shutdown Shutdown(Duration), /// Force shutdown ForceShutdown, } pub trait StreamServiceFactory: Send + Clone + 'static { type NewService: NewService; fn create(&self) -> Self::NewService; } pub trait ServiceFactory: Send + Clone + 'static { type NewService: NewService< Request = ServerMessage, Response = (), Error = (), InitError = (), >; fn create(&self) -> Self::NewService; } pub(crate) trait InternalServiceFactory: Send { fn name(&self) -> &str; fn clone_factory(&self) -> Box; fn create(&self) -> Box>; } pub(crate) type BoxedServerService = Box< Service< Request = (Option, ServerMessage), Response = (), Error = (), Future = FutureResult<(), ()>, >, >; pub(crate) struct StreamService { service: T, } impl StreamService { fn new(service: T) -> Self { StreamService { service } } } impl Service for StreamService where T: Service, T::Future: 'static, T::Error: 'static, { type Request = (Option, ServerMessage); type Response = (); type Error = (); type Future = FutureResult<(), ()>; fn poll_ready(&mut self) -> Poll<(), Self::Error> { self.service.poll_ready().map_err(|_| ()) } fn call(&mut self, (guard, req): (Option, ServerMessage)) -> Self::Future { match req { ServerMessage::Connect(stream) => { let stream = TcpStream::from_std(stream, &Handle::default()).map_err(|e| { error!("Can not convert to an async tcp stream: {}", e); }); if let Ok(stream) = stream { spawn(self.service.call(stream).map(move |val| { drop(guard); val })); ok(()) } else { err(()) } } _ => ok(()), } } } pub(crate) struct ServerService { service: T, } impl ServerService { fn new(service: T) -> Self { ServerService { service } } } impl Service for ServerService where T: Service, T::Future: 'static, T::Error: 'static, { type Request = (Option, ServerMessage); type Response = (); type Error = (); type Future = FutureResult<(), ()>; fn poll_ready(&mut self) -> Poll<(), Self::Error> { self.service.poll_ready().map_err(|_| ()) } fn call(&mut self, (guard, req): (Option, ServerMessage)) -> Self::Future { spawn(self.service.call(req).map(move |val| { drop(guard); val })); ok(()) } } pub(crate) struct ServiceNewService { name: String, inner: F, } impl ServiceNewService where F: ServiceFactory, { pub(crate) fn create(name: String, inner: F) -> Box { Box::new(Self { name, inner }) } } impl InternalServiceFactory for ServiceNewService where F: ServiceFactory, { fn name(&self) -> &str { &self.name } fn clone_factory(&self) -> Box { Box::new(Self { name: self.name.clone(), inner: self.inner.clone(), }) } fn create(&self) -> Box> { Box::new(self.inner.create().new_service().map(move |inner| { let service: BoxedServerService = Box::new(ServerService::new(inner)); service })) } } pub(crate) struct StreamNewService { name: String, inner: F, } impl StreamNewService where F: StreamServiceFactory, { pub(crate) fn create(name: String, inner: F) -> Box { Box::new(Self { name, inner }) } } impl InternalServiceFactory for StreamNewService where F: StreamServiceFactory, { fn name(&self) -> &str { &self.name } fn clone_factory(&self) -> Box { Box::new(Self { name: self.name.clone(), inner: self.inner.clone(), }) } fn create(&self) -> Box> { Box::new(self.inner.create().new_service().map(move |inner| { let service: BoxedServerService = Box::new(StreamService::new(inner)); service })) } } impl InternalServiceFactory for Box { fn name(&self) -> &str { self.as_ref().name() } fn clone_factory(&self) -> Box { self.as_ref().clone_factory() } fn create(&self) -> Box> { self.as_ref().create() } } impl ServiceFactory for F where F: Fn() -> T + Send + Clone + 'static, T: NewService, { type NewService = T; fn create(&self) -> T { (self)() } } impl StreamServiceFactory for F where F: Fn() -> T + Send + Clone + 'static, T: NewService, { type NewService = T; fn create(&self) -> T { (self)() } }