tangent128/webmetro
1
0
Fork 0
Code Issues 1 Pull requests Releases Packages Wiki Activity

normalize stream_parser poll to use modern future types

Browse source
This commit is contained in:
Tangent Wantwight 2019-11-17 01:19:30 -05:00
parent e6d8b58492
commit d563baaca8
2 changed files with 23 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
src/chunk.rs
View file

@ -1,5 +1,4 @@
use bytes::{Buf, Bytes}; use bytes::{Buf, Bytes};
use futures::{Async};
use futures3::prelude::*; use futures3::prelude::*;
use std::{ use std::{
io::Cursor, io::Cursor,
@ -141,10 +140,10 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> Stream for Webm
match chunker.state { match chunker.state {
ChunkerState::BuildingHeader(ref mut buffer) => { ChunkerState::BuildingHeader(ref mut buffer) => {
match chunker.source.poll_event(cx) { match chunker.source.poll_event(cx) {
Err(passthru) => return Ready(Some(Err(passthru))), Ready(Some(Err(passthru))) => return Ready(Some(Err(passthru))),
Ok(Async::NotReady) => return Pending, Pending => return Pending,
Ok(Async::Ready(None)) => return Ready(None), Ready(None) => return Ready(None),
Ok(Async::Ready(Some(element))) => match element { Ready(Some(Ok(element))) => match element {
WebmElement::Cluster => { WebmElement::Cluster => {
let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new())); let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new()));
let header_chunk = Chunk::Headers {bytes: Bytes::from(liberated_buffer.into_inner())}; let header_chunk = Chunk::Headers {bytes: Bytes::from(liberated_buffer.into_inner())};
@ -169,9 +168,9 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> Stream for Webm
}, },
ChunkerState::BuildingCluster(ref mut cluster_head, ref mut buffer) => { ChunkerState::BuildingCluster(ref mut cluster_head, ref mut buffer) => {
match chunker.source.poll_event(cx) { match chunker.source.poll_event(cx) {
Err(passthru) => return Ready(Some(Err(passthru))), Ready(Some(Err(passthru))) => return Ready(Some(Err(passthru))),
Ok(Async::NotReady) => return Pending, Pending => return Pending,
Ok(Async::Ready(Some(element))) => match element { Ready(Some(Ok(element))) => match element {
WebmElement::EbmlHead | WebmElement::Segment => { WebmElement::EbmlHead | WebmElement::Segment => {
let liberated_cluster_head = mem::replace(cluster_head, ClusterHead::new(0)); let liberated_cluster_head = mem::replace(cluster_head, ClusterHead::new(0));
let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new())); let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new()));
@ -222,7 +221,7 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> Stream for Webm
} }
}, },
}, },
Ok(Async::Ready(None)) => { Ready(None) => {
// flush final Cluster on end of stream // flush final Cluster on end of stream
let liberated_cluster_head = mem::replace(cluster_head, ClusterHead::new(0)); let liberated_cluster_head = mem::replace(cluster_head, ClusterHead::new(0));
let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new())); let liberated_buffer = mem::replace(buffer, Cursor::new(Vec::new()));

31
src/stream_parser.rs
View file

@ -1,5 +1,4 @@
use bytes::{Buf, BufMut, Bytes, BytesMut}; use bytes::{Buf, BufMut, Bytes, BytesMut};
use futures::Async;
use futures3::stream::{Stream, StreamExt, TryStream}; use futures3::stream::{Stream, StreamExt, TryStream};
use std::task::{Context, Poll}; use std::task::{Context, Poll};
@ -44,7 +43,7 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> EbmlStreamingPa
pub fn poll_event<'a, T: FromEbml<'a>>( pub fn poll_event<'a, T: FromEbml<'a>>(
&'a mut self, &'a mut self,
cx: &mut Context, cx: &mut Context,
) -> Result<Async<Option<T>>, WebmetroError> { ) -> Poll<Option<Result<T, WebmetroError>>> {
loop { loop {
match T::check_space(&self.buffer)? { match T::check_space(&self.buffer)? {
None => { None => {
@ -54,16 +53,16 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> EbmlStreamingPa
let mut bytes = self.buffer.split_to(info.element_len).freeze(); let mut bytes = self.buffer.split_to(info.element_len).freeze();
bytes.advance(info.body_offset); bytes.advance(info.body_offset);
self.borrowed = bytes; self.borrowed = bytes;
return Ok(Async::Ready(Some(T::decode( return Poll::Ready(Some(T::decode(
info.element_id, info.element_id,
&self.borrowed, &self.borrowed,
)?))); ).map_err(Into::into)));
} }
} }
if let Some(limit) = self.buffer_size_limit { if let Some(limit) = self.buffer_size_limit {
if limit <= self.buffer.len() { if limit <= self.buffer.len() {
return Err(WebmetroError::ResourcesExceeded); return Poll::Ready(Some(Err(WebmetroError::ResourcesExceeded)));
} }
} }
@ -73,8 +72,8 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> EbmlStreamingPa
self.buffer.put(buf); self.buffer.put(buf);
// ok can retry decoding now // ok can retry decoding now
} }
Poll::Ready(None) => return Ok(Async::Ready(None)), Poll::Ready(None) => return Poll::Ready(None),
Poll::Pending => return Ok(Async::NotReady), Poll::Pending => return Poll::Pending,
} }
} }
} }
@ -114,9 +113,9 @@ impl<I: Buf, S: Stream<Item = Result<I, WebmetroError>> + Unpin> EbmlStreamingPa
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use bytes::IntoBuf; use bytes::IntoBuf;
use futures::Async::*;
use futures3::{future::poll_fn, stream::StreamExt, FutureExt}; use futures3::{future::poll_fn, stream::StreamExt, FutureExt};
use matches::assert_matches; use matches::assert_matches;
use std::task::Poll::*;
use crate::stream_parser::*; use crate::stream_parser::*;
use crate::tests::ENCODE_WEBM_TEST_FILE; use crate::tests::ENCODE_WEBM_TEST_FILE;
@ -137,35 +136,35 @@ mod tests {
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::EbmlHead))) Ready(Some(Ok(WebmElement::EbmlHead)))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Segment))) Ready(Some(Ok(WebmElement::Segment)))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Tracks(_)))) Ready(Some(Ok(WebmElement::Tracks(_))))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Cluster))) Ready(Some(Ok(WebmElement::Cluster)))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Timecode(0)))) Ready(Some(Ok(WebmElement::Timecode(0))))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::SimpleBlock(_)))) Ready(Some(Ok(WebmElement::SimpleBlock(_))))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Cluster))) Ready(Some(Ok(WebmElement::Cluster)))
); );
assert_matches!( assert_matches!(
stream_parser.poll_event(cx), stream_parser.poll_event(cx),
Ok(Ready(Some(WebmElement::Timecode(1000)))) Ready(Some(Ok(WebmElement::Timecode(1000))))
); );
std::task::Poll::Ready(()) std::task::Poll::Ready(())