#![deprecated(since = "0.1.1", note = "use `futures::sync::mpsc` instead")]
#![allow(deprecated)]
#![cfg(feature = "with-deprecated")]
use std::io;
use std::sync::mpsc::TryRecvError;
use futures::{Poll, Async, Sink, AsyncSink, StartSend, Stream};
use mio::channel;
use reactor::{Handle, PollEvented};
#[must_use = "sinks do nothing unless polled"]
pub struct Sender<T> {
tx: channel::Sender<T>,
}
#[must_use = "streams do nothing unless polled"]
pub struct Receiver<T> {
rx: PollEvented<channel::Receiver<T>>,
}
pub fn channel<T>(handle: &Handle) -> io::Result<(Sender<T>, Receiver<T>)>
where T: Send + 'static,
{
let (tx, rx) = channel::channel();
let rx = try!(PollEvented::new(rx, handle));
Ok((Sender { tx: tx }, Receiver { rx: rx }))
}
impl<T> Sender<T> {
pub fn send(&self, t: T) -> io::Result<()> {
self.tx.send(t).map_err(|e| {
match e {
channel::SendError::Io(e) => e,
channel::SendError::Disconnected(_) => {
io::Error::new(io::ErrorKind::Other,
"channel has been disconnected")
}
}
})
}
}
impl<T> Sink for Sender<T> {
type SinkItem = T;
type SinkError = io::Error;
fn start_send(&mut self, t: T) -> StartSend<T, io::Error> {
Sender::send(self, t).map(|()| AsyncSink::Ready)
}
fn poll_complete(&mut self) -> Poll<(), io::Error> {
Ok(().into())
}
fn close(&mut self) -> Poll<(), io::Error> {
Ok(().into())
}
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Sender<T> {
Sender { tx: self.tx.clone() }
}
}
impl<T> Stream for Receiver<T> {
type Item = T;
type Error = io::Error;
fn poll(&mut self) -> Poll<Option<T>, io::Error> {
if let Async::NotReady = self.rx.poll_read() {
return Ok(Async::NotReady)
}
match self.rx.get_ref().try_recv() {
Ok(t) => Ok(Async::Ready(Some(t))),
Err(TryRecvError::Empty) => {
self.rx.need_read();
Ok(Async::NotReady)
}
Err(TryRecvError::Disconnected) => Ok(Async::Ready(None)),
}
}
}