pub fn start(self, path: &str) -> std::io::Result<Server> {
		let remote = self.remote.initialize()?;
		let rpc_handler = self.handler;
		let endpoint_addr = path.to_owned();
		let meta_extractor = self.meta_extractor;
		let session_stats = self.session_stats;
		let incoming_separator = self.incoming_separator;
		let outgoing_separator = self.outgoing_separator;
		let (stop_signal, stop_receiver) = oneshot::channel();
		let (start_signal, start_receiver) = oneshot::channel();

		remote.remote().spawn(move |handle| {
			use parity_tokio_ipc::Endpoint;

			if cfg!(unix) {
				// warn about existing file and remove it
				if ::std::fs::remove_file(&endpoint_addr).is_ok() {
					warn!("Removed existing file '{}'.", &endpoint_addr);
				}
			}

			let listener = match Endpoint::new(endpoint_addr, handle) {
				Ok(l) => l,
				Err(e) => {
					start_signal.send(Err(e)).expect("Cannot fail since receiver never dropped before receiving");
					return future::Either::A(future::ok(()));
				}
			};

			let remote = handle.remote().clone();
			let connections = listener.incoming();
			let mut id = 0u64;

			let server = connections.for_each(move |(io_stream, remote_id)| {
				id = id.wrapping_add(1);
				let session_id = id;
				let session_stats = session_stats.clone();
				trace!(target: "ipc", "Accepted incoming IPC connection: {}", session_id);
				session_stats.as_ref().map(|stats| stats.open_session(session_id));

				let (sender, receiver) = mpsc::channel(16);
				let meta = meta_extractor.extract(&RequestContext {
					endpoint_addr: &remote_id,
					session_id,
					sender,
				});
				let service = Service::new(rpc_handler.clone(), meta);
				let (writer, reader) = io_stream.framed(
					codecs::StreamCodec::new(
						incoming_separator.clone(),
						outgoing_separator.clone(),
					)
				).split();
				let responses = reader.and_then(move |req| {
					service.call(req).then(move |response| match response {
						Err(e) => {
							warn!(target: "ipc", "Error while processing request: {:?}", e);
							future::ok(None)
						},
						Ok(None) => {
							future::ok(None)
						},
						Ok(Some(response_data)) => {
							trace!(target: "ipc", "Sent response: {}", &response_data);
							future::ok(Some(response_data))
						}
					})
				})
				.filter_map(|x| x)
				// we use `select_with_weak` here, instead of `select`, to close the stream
				// as soon as the ipc pipe is closed
				.select_with_weak(receiver.map_err(|e| {
					warn!(target: "ipc", "Notification error: {:?}", e);
					std::io::ErrorKind::Other.into()
				}));

				let writer = writer.send_all(responses).then(move |_| {
					trace!(target: "ipc", "Peer: service finished");
					session_stats.as_ref().map(|stats| stats.close_session(session_id));
					Ok(())
				});

				remote.spawn(|_| writer);

				Ok(())
			});
			start_signal.send(Ok(())).expect("Cannot fail since receiver never dropped before receiving");

			let stop = stop_receiver.map_err(|_| std::io::ErrorKind::Interrupted.into());
			future::Either::B(
				server.select(stop)
					.map(|_| ())
					.map_err(|_| ())
			)
		});

		match start_receiver.wait().expect("Message should always be sent") {
			Ok(()) => Ok(Server { path: path.to_owned(), remote: Some(remote), stop: Some(stop_signal) }),
			Err(e) => Err(e)
		}
	}
/// Run server (in a separate thread)
Run server (in a separate thread)
Run server ( in a separate thread )
pub fn start(self, path: &str) -> std::io::Result<Server> {
		let remote = self.remote.initialize()?;
		let rpc_handler = self.handler;
		let endpoint_addr = path.to_owned();
		let meta_extractor = self.meta_extractor;
		let session_stats = self.session_stats;
		let incoming_separator = self.incoming_separator;
		let outgoing_separator = self.outgoing_separator;
		let (stop_signal, stop_receiver) = oneshot::channel();
		let (start_signal, start_receiver) = oneshot::channel();
		remote.remote().spawn(move |handle| {
			use parity_tokio_ipc::Endpoint;
			if cfg!(unix) {
				// warn about existing file and remove it
				if ::std::fs::remove_file(&endpoint_addr).is_ok() {
					warn!("Removed existing file '{}'.", &endpoint_addr);
				}
			}
			let listener = match Endpoint::new(endpoint_addr, handle) {
				Ok(l) => l,
				Err(e) => {
					start_signal.send(Err(e)).expect("Cannot fail since receiver never dropped before receiving");
					return future::Either::A(future::ok(()));
				}
			};
			let remote = handle.remote().clone();
			let connections = listener.incoming();
			let mut id = 0u64;
			let server = connections.for_each(move |(io_stream, remote_id)| {
				id = id.wrapping_add(1);
				let session_id = id;
				let session_stats = session_stats.clone();
				trace!(target: "ipc", "Accepted incoming IPC connection: {}", session_id);
				session_stats.as_ref().map(|stats| stats.open_session(session_id));
				let (sender, receiver) = mpsc::channel(16);
				let meta = meta_extractor.extract(&RequestContext {
					endpoint_addr: &remote_id,
					session_id,
					sender,
				});
				let service = Service::new(rpc_handler.clone(), meta);
				let (writer, reader) = io_stream.framed(
					codecs::StreamCodec::new(
						incoming_separator.clone(),
						outgoing_separator.clone(),
					)
				).split();
				let responses = reader.and_then(move |req| {
					service.call(req).then(move |response| match response {
						Err(e) => {
							warn!(target: "ipc", "Error while processing request: {:?}", e);
							future::ok(None)
						},
						Ok(None) => {
							future::ok(None)
						},
						Ok(Some(response_data)) => {
							trace!(target: "ipc", "Sent response: {}", &response_data);
							future::ok(Some(response_data))
						}
					})
				})
				.filter_map(|x| x)
				// we use `select_with_weak` here, instead of `select`, to close the stream
				// as soon as the ipc pipe is closed
				.select_with_weak(receiver.map_err(|e| {
					warn!(target: "ipc", "Notification error: {:?}", e);
					std::io::ErrorKind::Other.into()
				}));
				let writer = writer.send_all(responses).then(move |_| {
					trace!(target: "ipc", "Peer: service finished");
					session_stats.as_ref().map(|stats| stats.close_session(session_id));
					Ok(())
				});
				remote.spawn(|_| writer);
				Ok(())
			});
			start_signal.send(Ok(())).expect("Cannot fail since receiver never dropped before receiving");
			let stop = stop_receiver.map_err(|_| std::io::ErrorKind::Interrupted.into());
			future::Either::B(
				server.select(stop)
					.map(|_| ())
					.map_err(|_| ())
			)
		});
		match start_receiver.wait().expect("Message should always be sent") {
			Ok(()) => Ok(Server { path: path.to_owned(), remote: Some(remote), stop: Some(stop_signal) }),
			Err(e) => Err(e)
		}
	}
pub fn start ( self , path : & str ) -> std :: io :: Result < Server > { let remote = self . remote . initialize ( ) ? ; let rpc_handler = self . handler ; let endpoint_addr = path . to_owned ( ) ; let meta_extractor = self . meta_extractor ; let session_stats = self . session_stats ; let incoming_separator = self . incoming_separator ; let outgoing_separator = self . outgoing_separator ; let ( stop_signal , stop_receiver ) = oneshot :: channel ( ) ; let ( start_signal , start_receiver ) = oneshot :: channel ( ) ; remote . remote ( ) . spawn ( move | handle | { use parity_tokio_ipc :: Endpoint ; if cfg ! ( unix ) { // warn about existing file and remove it if :: std :: fs :: remove_file ( & endpoint_addr ) . is_ok ( ) { warn ! ( " " endpoint_addr ) ; } } let listener = match Endpoint :: new ( endpoint_addr , handle ) { Ok ( l ) => l , Err ( e ) => { start_signal . send ( Err ( e ) ) . expect ( " " ) ; return future :: Either :: A ( future :: ok ( ( ) ) ) ; } } ; let remote = handle . remote ( ) . clone ( ) ; let connections = listener . incoming ( ) ; let mut id = 0u64 ; let server = connections . for_each ( move | ( io_stream , remote_id ) | { id = id . wrapping_add ( 1 ) ; let session_id = id ; let session_stats = session_stats . clone ( ) ; trace ! ( target " " " " session_id ) ; session_stats . as_ref ( ) . map ( | stats | stats . open_session ( session_id ) ) ; let ( sender , receiver ) = mpsc :: channel ( 16 ) ; let meta = meta_extractor . extract ( & RequestContext { endpoint_addr : & remote_id , session_id , sender , } ) ; let service = Service :: new ( rpc_handler . clone ( ) , meta ) ; let ( writer , reader ) = io_stream . framed ( codecs :: StreamCodec :: new ( incoming_separator . clone ( ) , outgoing_separator . clone ( ) , ) ) . split ( ) ; let responses = reader . and_then ( move | req | { service . call ( req ) . then ( move | response | match response { Err ( e ) => { warn ! ( target " " " " e ) ; future :: ok ( None ) } , Ok ( None ) => { future :: ok ( None ) } , Ok ( Some ( response_data ) ) => { trace ! ( target " " " " response_data ) ; future :: ok ( Some ( response_data ) ) } } ) } ) . filter_map ( | x | x ) // we use `select_with_weak` here, instead of `select`, to close the stream // as soon as the ipc pipe is closed . select_with_weak ( receiver . map_err ( | e | { warn ! ( target " " " " e ) ; std :: io :: ErrorKind :: Other . into ( ) } ) ) ; let writer = writer . send_all ( responses ) . then ( move | _ | { trace ! ( target " " " " ) ; session_stats . as_ref ( ) . map ( | stats | stats . close_session ( session_id ) ) ; Ok ( ( ) ) } ) ; remote . spawn ( | _ | writer ) ; Ok ( ( ) ) } ) ; start_signal . send ( Ok ( ( ) ) ) . expect ( " " ) ; let stop = stop_receiver . map_err ( | _ | std :: io :: ErrorKind :: Interrupted . into ( ) ) ; future :: Either :: B ( server . select ( stop ) . map ( | _ | ( ) ) . map_err ( | _ | ( ) ) ) } ) ; match start_receiver . wait ( ) . expect ( " " ) { Ok ( ( ) ) => Ok ( Server { path : path . to_owned ( ) , remote : Some ( remote ) , stop : Some ( stop_signal ) } ) , Err ( e ) => Err ( e ) } }
Run server (in a separate thread)
Run server ( in a separate thread )
self
path
&str
self
path
&str