mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-09-08 00:00:13 +02:00
Deprecate the channel module
The `futures::sync::mpsc` module should entirely supplant it.
This commit is contained in:
+7
-6
@@ -42,7 +42,7 @@ fn main() {
|
|||||||
// Create a channel for our stream, which other sockets will use to
|
// Create a channel for our stream, which other sockets will use to
|
||||||
// send us messages. Then register our address with the stream to send
|
// send us messages. Then register our address with the stream to send
|
||||||
// data to us.
|
// data to us.
|
||||||
let (tx, rx) = tokio_core::channel::channel(&handle).unwrap();
|
let (tx, rx) = futures::sync::mpsc::unbounded();
|
||||||
connections.borrow_mut().insert(addr, tx);
|
connections.borrow_mut().insert(addr, tx);
|
||||||
|
|
||||||
// Define here what we do for the actual I/O. That is, read a bunch of
|
// Define here what we do for the actual I/O. That is, read a bunch of
|
||||||
@@ -54,7 +54,7 @@ fn main() {
|
|||||||
// Model the read portion of this socket by mapping an infinite
|
// Model the read portion of this socket by mapping an infinite
|
||||||
// iterator to each line off the socket. This "loop" is then
|
// iterator to each line off the socket. This "loop" is then
|
||||||
// terminated with an error once we hit EOF on the socket.
|
// terminated with an error once we hit EOF on the socket.
|
||||||
let iter = stream::iter(iter::repeat(()).map(Ok));
|
let iter = stream::iter(iter::repeat(()).map(Ok::<(), Error>));
|
||||||
let socket_reader = iter.fold(reader, move |reader, _| {
|
let socket_reader = iter.fold(reader, move |reader, _| {
|
||||||
// Read a line off the socket, failing if we're at EOF
|
// Read a line off the socket, failing if we're at EOF
|
||||||
let line = io::read_until(reader, b'\n', Vec::new());
|
let line = io::read_until(reader, b'\n', Vec::new());
|
||||||
@@ -74,18 +74,18 @@ fn main() {
|
|||||||
let connections = connections_inner.clone();
|
let connections = connections_inner.clone();
|
||||||
line.map(move |(reader, message)| {
|
line.map(move |(reader, message)| {
|
||||||
println!("{}: {:?}", addr, message);
|
println!("{}: {:?}", addr, message);
|
||||||
let conns = connections.borrow_mut();
|
let mut conns = connections.borrow_mut();
|
||||||
if let Ok(msg) = message {
|
if let Ok(msg) = message {
|
||||||
// For each open connection except the sender, send the
|
// For each open connection except the sender, send the
|
||||||
// string via the channel.
|
// string via the channel.
|
||||||
let iter = conns.iter()
|
let iter = conns.iter_mut()
|
||||||
.filter(|&(&k, _)| k != addr)
|
.filter(|&(&k, _)| k != addr)
|
||||||
.map(|(_, v)| v);
|
.map(|(_, v)| v);
|
||||||
for tx in iter {
|
for tx in iter {
|
||||||
tx.send(format!("{}: {}", addr, msg)).unwrap();
|
tx.send(format!("{}: {}", addr, msg)).unwrap();
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
let tx = conns.get(&addr).unwrap();
|
let tx = conns.get_mut(&addr).unwrap();
|
||||||
tx.send("You didn't send valid UTF-8.".to_string()).unwrap();
|
tx.send("You didn't send valid UTF-8.".to_string()).unwrap();
|
||||||
}
|
}
|
||||||
reader
|
reader
|
||||||
@@ -97,13 +97,14 @@ fn main() {
|
|||||||
let socket_writer = rx.fold(writer, |writer, msg| {
|
let socket_writer = rx.fold(writer, |writer, msg| {
|
||||||
let amt = io::write_all(writer, msg.into_bytes());
|
let amt = io::write_all(writer, msg.into_bytes());
|
||||||
let amt = amt.map(|(writer, _)| writer);
|
let amt = amt.map(|(writer, _)| writer);
|
||||||
amt
|
amt.map_err(|_| ())
|
||||||
});
|
});
|
||||||
|
|
||||||
// Now that we've got futures representing each half of the socket, we
|
// Now that we've got futures representing each half of the socket, we
|
||||||
// use the `select` combinator to wait for either half to be done to
|
// use the `select` combinator to wait for either half to be done to
|
||||||
// tear down the other. Then we spawn off the result.
|
// tear down the other. Then we spawn off the result.
|
||||||
let connections = connections.clone();
|
let connections = connections.clone();
|
||||||
|
let socket_reader = socket_reader.map_err(|_| ());
|
||||||
let connection = socket_reader.map(|_| ()).select(socket_writer.map(|_| ()));
|
let connection = socket_reader.map(|_| ()).select(socket_writer.map(|_| ()));
|
||||||
handle.spawn(connection.then(move |_| {
|
handle.spawn(connection.then(move |_| {
|
||||||
connections.borrow_mut().remove(&addr);
|
connections.borrow_mut().remove(&addr);
|
||||||
|
|||||||
@@ -3,6 +3,9 @@
|
|||||||
//! This module contains a `Sender` and `Receiver` pair types which can be used
|
//! This module contains a `Sender` and `Receiver` pair types which can be used
|
||||||
//! to send messages between different future tasks.
|
//! to send messages between different future tasks.
|
||||||
|
|
||||||
|
#![deprecated(since = "0.1.1", note = "use `futures::sync::mpsc` instead")]
|
||||||
|
#![allow(deprecated)]
|
||||||
|
|
||||||
use std::io;
|
use std::io;
|
||||||
use std::sync::mpsc::TryRecvError;
|
use std::sync::mpsc::TryRecvError;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user