use {Async, Poll}; use stream::Stream; /// A stream combinator which returns a maximum number of elements. /// /// This structure is produced by the `Stream::take` method. #[derive(Debug)] #[must_use = "streams do nothing unless polled"] pub struct Take { stream: S, remaining: u64, } pub fn new(s: S, amt: u64) -> Take where S: Stream, { Take { stream: s, remaining: amt, } } impl Take { /// Acquires a reference to the underlying stream that this combinator is /// pulling from. pub fn get_ref(&self) -> &S { &self.stream } /// Acquires a mutable reference to the underlying stream that this /// combinator is pulling from. /// /// Note that care must be taken to avoid tampering with the state of the /// stream which may otherwise confuse this combinator. pub fn get_mut(&mut self) -> &mut S { &mut self.stream } /// Consumes this combinator, returning the underlying stream. /// /// Note that this may discard intermediate state of this combinator, so /// care should be taken to avoid losing resources when this is called. pub fn into_inner(self) -> S { self.stream } } // Forwarding impl of Sink from the underlying stream impl ::sink::Sink for Take where S: ::sink::Sink + Stream { type SinkItem = S::SinkItem; type SinkError = S::SinkError; fn start_send(&mut self, item: S::SinkItem) -> ::StartSend { self.stream.start_send(item) } fn poll_complete(&mut self) -> Poll<(), S::SinkError> { self.stream.poll_complete() } fn close(&mut self) -> Poll<(), S::SinkError> { self.stream.close() } } impl Stream for Take where S: Stream, { type Item = S::Item; type Error = S::Error; fn poll(&mut self) -> Poll, S::Error> { if self.remaining == 0 { Ok(Async::Ready(None)) } else { let next = try_ready!(self.stream.poll()); match next { Some(_) => self.remaining -= 1, None => self.remaining = 0, } Ok(Async::Ready(next)) } } }