Rust Futures Weirdness

cargo-run.txt

   Compiling futures-demo v0.1.0 (/Users/robbiemckinstry/rust-playground/futures-demo)
error: the `for_each` method cannot be invoked on a trait object
  --> src/main.rs:63:24
   |
63 |         stream.wrapped.for_each(|c| {
   |                        ^^^^^^^^
help: another candidate was found in the following trait, perhaps add a `use` for it:
   |
3  | use crate::futures::StreamExt;
   |

error: aborting due to previous error

error: Could not compile `futures-demo`.

To learn more, run the command again with --verbose.

Cargo.toml

[package]
name = "futures-demo"
version = "0.1.0"
authors = ["Robbie"]
edition = "2018"

[dependencies]
futures-preview = "0.3.0-alpha.17"
tokio = "=0.2.0-alpha.1"

main.rs

extern crate futures; // 0.3.0-alpha.17
extern crate tokio; // =0.2.0-alpha
use futures::prelude::Stream;
use futures::Poll;
use futures::future;
use tokio::runtime::Runtime;
use std::pin::Pin;
use crate::futures::StreamExt;

pub trait CharacterStream: Stream<Item=char> {}

struct StreamWrapper {
    wrapped: Box<dyn CharacterStream>,
}

impl StreamWrapper {
    fn new<T: 'static + CharacterStream>(stream: T) -> Self {
        let wrapped = Box::new(stream.clone());
        StreamWrapper{wrapped}
    }
}

#[derive(Clone)]
struct CharVector {
    my_stream: Vec<char>,
}

impl dyn CharacterStream<Item = char> {
    pub fn new(my_string: String) -> impl CharacterStream {
        CharVector::new(my_string)
    }
}

impl CharVector {
    fn new(my_string: String) -> Self {
        let mut chars: Vec<char> = my_string.chars().collect();
        chars.reverse();
        CharVector{my_stream: chars}
    }
}

impl Stream for CharVector {
    type Item = char;
    fn poll_next(mut self: Pin<&mut Self>, _ctx: &mut std::task::Context) -> Poll<Option<Self::Item>> {
        let c = self.my_stream.pop();
        Poll::Ready(c)
    }
}

impl CharacterStream for CharVector {}

fn main() {
    println!("{}", "Starting Rust example!");
    let stream = StreamWrapper::new(
        CharacterStream::new(
            "abcd".to_owned()
        )
    );

    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        println!("{}", "Executing from inside the future!");
        stream.wrapped.for_each(|c| {
            print!("{}", c);
            future::ready(())
        }).await;
    });
}