From 0075e52f7b548a949f9450be73b9b9253e1663fc Mon Sep 17 00:00:00 2001
From: Tangent 128 <Tangent128@gmail.com>
Date: Sat, 14 Apr 2018 18:44:53 -0400
Subject: [PATCH] Run filter under Tokio runtime so that Delay works

---
 src/commands/filter.rs | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/src/commands/filter.rs b/src/commands/filter.rs
index 5025afc..2062d01 100644
--- a/src/commands/filter.rs
+++ b/src/commands/filter.rs
@@ -6,6 +6,7 @@ use std::{
 
 use clap::{App, Arg, ArgMatches, SubCommand};
 use futures::prelude::*;
+use tokio;
 
 use super::stdin_stream;
 use webmetro::{
@@ -28,7 +29,7 @@ pub fn options() -> App<'static, 'static> {
 
 pub fn run(args: &ArgMatches) -> Result<(), Box<Error>> {
 
-    let mut chunk_stream: Box<Stream<Item = Chunk, Error = WebmetroError>> = Box::new(
+    let mut chunk_stream: Box<Stream<Item = Chunk, Error = WebmetroError> + Send> = Box::new(
         stdin_stream()
         .parse_ebml()
         .chunk_webm()
@@ -39,9 +40,12 @@ pub fn run(args: &ArgMatches) -> Result<(), Box<Error>> {
         chunk_stream = Box::new(chunk_stream.throttle());
     }
 
-    chunk_stream.fold((), |_, chunk| {
-        io::stdout().write_all(chunk.as_ref())
-    }).wait()?;
+    tokio::run(chunk_stream.for_each(|chunk| {
+        io::stdout().write_all(chunk.as_ref()).map_err(WebmetroError::IoError)
+    }).map_err(|err| {
+        println!("Error: {}", err);
+        ::std::process::exit(1);
+    }));
 
     Ok(())
 }