DGNum/rust-irc
6
0
Fork 0
Code Issues 1 Pull requests 3 Projects Releases Packages Wiki Activity Actions

Merge pull request #152 from DoumanAsh/remove_tokio_core

Browse source 
Clean up tokio-core
This commit is contained in:
Aaron Weiss 2018-09-22 17:23:42 -04:00 committed by GitHub
commit 0534ab061f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 28 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -39,8 +39,8 @@ serde = "1.0"
serde_derive = "1.0"
serde_json = { version = "1.0", optional = true }
serde_yaml = { version = "0.7", optional = true }
tokio = "0.1"
tokio-codec = "0.1"
tokio-core = "0.1"
tokio-io = "0.1"
tokio-mockstream = "1.1"
tokio-timer = "0.1"

13
src/client/conn.rs
View file

@ -8,8 +8,7 @@ use encoding::label::encoding_from_whatwg_label;
use futures::{Async, Poll, Future, Sink, StartSend, Stream};
use native_tls::{Certificate, TlsConnector, Identity};
use tokio_codec::Decoder;
use tokio_core::reactor::Handle;
use tokio_core::net::{TcpStream, TcpStreamNew};
use tokio::net::{TcpStream, ConnectFuture};
use tokio_mockstream::MockStream;
use tokio_tls::{self, TlsStream};
@ -48,7 +47,7 @@ type TlsFuture = Box<Future<Error = error::IrcError, Item = TlsStream<TcpStream>
/// A future representing an eventual `Connection`.
pub enum ConnectionFuture<'a> {
#[doc(hidden)]
Unsecured(&'a Config, TcpStreamNew),
Unsecured(&'a Config, ConnectFuture),
#[doc(hidden)]
Secured(&'a Config, TlsFuture),
#[doc(hidden)]
@ -122,8 +121,8 @@ impl<'a> Future for ConnectionFuture<'a> {
}
impl Connection {
/// Creates a new `Connection` using the specified `Config` and `Handle`.
pub fn new<'a>(config: &'a Config, handle: &Handle) -> error::Result<ConnectionFuture<'a>> {
/// Creates a new `Connection` using the specified `Config`
pub fn new<'a>(config: &'a Config) -> error::Result<ConnectionFuture<'a>> {
if config.use_mock_connection() {
Ok(ConnectionFuture::Mock(config))
} else if config.use_ssl() {
@ -148,7 +147,7 @@ impl Connection {
info!("Using {} for client certificate authentication.", client_cert_path);
}
let connector: tokio_tls::TlsConnector = builder.build()?.into();
let stream = Box::new(TcpStream::connect(&config.socket_addr()?, handle).map_err(|e| {
let stream = Box::new(TcpStream::connect(&config.socket_addr()?).map_err(|e| {
let res: error::IrcError = e.into();
res
}).and_then(move |socket| {
@ -161,7 +160,7 @@ impl Connection {
info!("Connecting to {}.", config.server()?);
Ok(ConnectionFuture::Unsecured(
config,
TcpStream::connect(&config.socket_addr()?, handle),
TcpStream::connect(&config.socket_addr()?),
))
}
}

27
src/client/mod.rs
View file

@ -57,7 +57,7 @@ use futures::stream::SplitStream;
use futures::sync::mpsc;
use futures::sync::oneshot;
use futures::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio_core::reactor::{Core, Handle};
use tokio::runtime::current_thread::Runtime;
use error;
use client::conn::{Connection, ConnectionFuture};
@ -694,10 +694,9 @@ impl IrcClient {
let cfg = config.clone();
let _ = thread::spawn(move || {
let mut reactor = Core::new().unwrap();
let mut reactor = Runtime::new().unwrap();
// Setting up internal processing stuffs.
let handle = reactor.handle();
let conn = reactor.run(Connection::new(&cfg, &handle).unwrap()).unwrap();
let conn = reactor.block_on(Connection::new(&cfg).unwrap()).unwrap();
tx_view.send(conn.log_view()).unwrap();
let (sink, stream) = conn.split();
@ -709,7 +708,7 @@ impl IrcClient {
// Send the stream half back to the original thread.
tx_incoming.send(stream).unwrap();
reactor.run(outgoing_future).unwrap();
reactor.block_on(outgoing_future).unwrap();
});
Ok(IrcClient {
@ -733,36 +732,35 @@ impl IrcClient {
/// # Example
/// ```no_run
/// # extern crate irc;
/// # extern crate tokio_core;
/// # extern crate tokio;
/// # use std::default::Default;
/// # use irc::client::prelude::*;
/// # use irc::client::PackedIrcClient;
/// # use irc::error;
/// # use tokio_core::reactor::Core;
/// # use tokio::runtime::current_thread::Runtime;
/// # fn main() {
/// # let config = Config {
/// # nickname: Some("example".to_owned()),
/// # server: Some("irc.example.com".to_owned()),
/// # .. Default::default()
/// # };
/// let mut reactor = Core::new().unwrap();
/// let future = IrcClient::new_future(reactor.handle(), &config).unwrap();
/// let mut reactor = Runtime::new().unwrap();
/// let future = IrcClient::new_future(&config).unwrap();
/// // immediate connection errors (like no internet) will turn up here...
/// let PackedIrcClient(client, future) = reactor.run(future).unwrap();
/// let PackedIrcClient(client, future) = reactor.block_on(future).unwrap();
/// // runtime errors (like disconnections and so forth) will turn up here...
/// reactor.run(client.stream().for_each(move |irc_msg| {
/// reactor.block_on(client.stream().for_each(move |irc_msg| {
/// // processing messages works like usual
/// process_msg(&client, irc_msg)
/// }).join(future)).unwrap();
/// # }
/// # fn process_msg(server: &IrcClient, message: Message) -> error::Result<()> { Ok(()) }
/// ```
pub fn new_future(handle: Handle, config: &Config) -> error::Result<IrcClientFuture> {
pub fn new_future(config: &Config) -> error::Result<IrcClientFuture> {
let (tx_outgoing, rx_outgoing) = mpsc::unbounded();
Ok(IrcClientFuture {
conn: Connection::new(config, &handle)?,
_handle: handle,
conn: Connection::new(config)?,
config: config,
tx_outgoing: Some(tx_outgoing),
rx_outgoing: Some(rx_outgoing),
@ -793,7 +791,6 @@ impl IrcClient {
#[derive(Debug)]
pub struct IrcClientFuture<'a> {
conn: ConnectionFuture<'a>,
_handle: Handle,
config: &'a Config,
tx_outgoing: Option<UnboundedSender<Message>>,
rx_outgoing: Option<UnboundedReceiver<Message>>,

14
src/client/reactor.rs
View file

@ -25,7 +25,7 @@
use futures::{Future, IntoFuture, Stream};
use futures::future;
use tokio_core::reactor::{Core, Handle};
use tokio::runtime::current_thread as tokio_rt;
use client::data::Config;
use client::{IrcClient, IrcClientFuture, PackedIrcClient, Client};
@ -40,7 +40,7 @@ use proto::Message;
///
/// For a full example usage, see [`irc::client::reactor`](./index.html).
pub struct IrcReactor {
inner: Core,
inner: tokio_rt::Runtime,
handlers: Vec<Box<Future<Item = (), Error = error::IrcError>>>,
}
@ -48,7 +48,7 @@ impl IrcReactor {
/// Creates a new reactor.
pub fn new() -> error::Result<IrcReactor> {
Ok(IrcReactor {
inner: Core::new()?,
inner: tokio_rt::Runtime::new()?,
handlers: Vec::new(),
})
}
@ -70,7 +70,7 @@ impl IrcReactor {
/// # }
/// ```
pub fn prepare_client<'a>(&mut self, config: &'a Config) -> error::Result<IrcClientFuture<'a>> {
IrcClient::new_future(self.inner_handle(), config)
IrcClient::new_future(config)
}
/// Runs an [`IrcClientFuture`](../struct.IrcClientFuture.html), such as one from
@ -91,7 +91,7 @@ impl IrcReactor {
/// # }
/// ```
pub fn connect_client(&mut self, future: IrcClientFuture) -> error::Result<IrcClient> {
self.inner.run(future).map(|PackedIrcClient(client, future)| {
self.inner.block_on(future).map(|PackedIrcClient(client, future)| {
self.register_future(future);
client
})
@ -159,7 +159,7 @@ impl IrcReactor {
/// Returns a handle to the internal event loop. This is a sort of escape hatch that allows you
/// to take more control over what runs on the reactor using `tokio`. This can be used for
/// sharing this reactor with some elements of other libraries.
pub fn inner_handle(&self) -> Handle {
pub fn inner_handle(&self) -> tokio_rt::Handle {
self.inner.handle()
}
@ -187,6 +187,6 @@ impl IrcReactor {
while let Some(handler) = self.handlers.pop() {
handlers.push(handler);
}
self.inner.run(future::join_all(handlers).map(|_| ()))
self.inner.block_on(future::join_all(handlers).map(|_| ()))
}
}

2
src/lib.rs
View file

@ -59,8 +59,8 @@ extern crate serde_derive;
extern crate serde_json;
#[cfg(feature = "yaml")]
extern crate serde_yaml;
extern crate tokio;
extern crate tokio_codec;
extern crate tokio_core;
extern crate tokio_io;
extern crate tokio_mockstream;
extern crate tokio_timer;