pub fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) where
		G: FnOnce(Sink<T, E>) -> R,
		R: future::IntoFuture<Future=F, Item=(), Error=()>,
		F: future::Future<Item=(), Error=()> + Send + 'static,
	{
		let id = self.next_id.next_id();
		if let Ok(sink) = subscriber.assign_id(id.into()) {
			let (tx, rx) = oneshot::channel();
			let future = into_future(sink)
				.into_future()
				.select(rx.map_err(|e| warn!("Error timeing out: {:?}", e)))
				.then(|_| Ok(()));

			self.active_subscriptions.lock().insert(id, tx);
			self.executor.spawn(future);
		}
	}
/// Creates new subscription for given subscriber.
///
/// Second parameter is a function that converts Subscriber sink into a future.
/// This future will be driven to completion bu underlying event loop
/// or will be cancelled in case #cancel is invoked.
Creates new subscription for given subscriber.
Second parameter is a function that converts Subscriber sink into a future.
This future will be driven to completion bu underlying event loop
or will be cancelled in case #cancel is invoked.
Creates new subscription for given subscriber . Second parameter is a function that converts Subscriber sink into a future . This future will be driven to completion bu underlying event loop or will be cancelled in case #cancel is invoked .
pub fn add<T, E, G, R, F>(&self, subscriber: Subscriber<T, E>, into_future: G) where
		G: FnOnce(Sink<T, E>) -> R,
		R: future::IntoFuture<Future=F, Item=(), Error=()>,
		F: future::Future<Item=(), Error=()> + Send + 'static,
	{
		let id = self.next_id.next_id();
		if let Ok(sink) = subscriber.assign_id(id.into()) {
			let (tx, rx) = oneshot::channel();
			let future = into_future(sink)
				.into_future()
				.select(rx.map_err(|e| warn!("Error timeing out: {:?}", e)))
				.then(|_| Ok(()));
			self.active_subscriptions.lock().insert(id, tx);
			self.executor.spawn(future);
		}
	}
pub fn add < T , E , G , R , F > ( & self , subscriber : Subscriber < T , E > , into_future : G ) where G : FnOnce ( Sink < T , E > ) -> R , R : future :: IntoFuture < Future = F , Item = ( ) , Error = ( ) > , F : future :: Future < Item = ( ) , Error = ( ) > + Send + ' static , { let id = self . next_id . next_id ( ) ; if let Ok ( sink ) = subscriber . assign_id ( id . into ( ) ) { let ( tx , rx ) = oneshot :: channel ( ) ; let future = into_future ( sink ) . into_future ( ) . select ( rx . map_err ( | e | warn ! ( " " e ) ) ) . then ( | _ | Ok ( ( ) ) ) ; self . active_subscriptions . lock ( ) . insert ( id , tx ) ; self . executor . spawn ( future ) ; } }
Creates new subscription for given subscriber.
Creates new subscription for given subscriber .
&self
subscriber
Subscriber<T, E>
into_future
G
&self
subscriber
Subscriber<T, E>
into_future
G