Type-erase the filter closure in EventStream

Instead of the filter closure being a type parameter we can erase its
type with `Arc<dyn Fn...>`. This means that the filter is executed with
dynamic dispatch instead of static but I doubt that it makes a
noticeable/significant difference. Erasing the type makes `EventStream`
easier to store on another struct since it's a hassle to specify
concrete type parameters for closures.

Also `EventReader::poll` and `read` are updated to use more specific
`FnMut` closures for the sake of flexibility. This is effectively the
same since `FnMut` is a supertrait of `Fn` but it enables mutating the
env of the closure.

Closes #3

src/event/reader.rs

@@ -41,7 +41,7 @@ impl EventReader {
 
     pub fn poll<F>(&self, timeout: Option<Duration>, filter: F) -> io::Result<bool>
     where
-        F: Fn(&Event) -> bool,
+        F: FnMut(&Event) -> bool,
     {
         let (mut reader, timeout) = if let Some(timeout) = timeout {
             let poll_timeout = PollTimeout::new(Some(timeout));
@@ -58,7 +58,7 @@ impl EventReader {
 
     pub fn read<F>(&self, filter: F) -> io::Result<Event>
     where
-        F: Fn(&Event) -> bool,
+        F: FnMut(&Event) -> bool,
     {
         let mut reader = self.shared.lock();
         reader.read(filter)
@@ -73,11 +73,11 @@ struct Shared {
 }
 
 impl Shared {
-    fn poll<F>(&mut self, timeout: Option<Duration>, filter: F) -> io::Result<bool>
+    fn poll<F>(&mut self, timeout: Option<Duration>, mut filter: F) -> io::Result<bool>
     where
-        F: Fn(&Event) -> bool,
+        F: FnMut(&Event) -> bool,
     {
-        if self.events.iter().any(&filter) {
+        if self.events.iter().any(&mut (filter)) {
             return Ok(true);
         }
 
@@ -111,9 +111,9 @@ impl Shared {
         }
     }
 
-    fn read<F>(&mut self, filter: F) -> io::Result<Event>
+    fn read<F>(&mut self, mut filter: F) -> io::Result<Event>
     where
-        F: Fn(&Event) -> bool,
+        F: FnMut(&Event) -> bool,
     {
         let mut skipped_events = VecDeque::new();
 
@@ -126,7 +126,7 @@ impl Shared {
                     skipped_events.push_back(event);
                 }
             }
-            let _ = self.poll(None, &filter)?;
+            let _ = self.poll(None, &mut filter)?;
         }
     }
 }

src/event/stream.rs

@@ -25,10 +25,9 @@ use super::{reader::EventReader, source::PlatformWaker, Event};
 ///
 /// Create an event stream for a terminal by passing the reader [crate::Terminal::event_reader]
 /// into [EventStream::new] with a filter.
-#[derive(Debug)]
-pub struct EventStream<F: Fn(&Event) -> bool> {
+pub struct EventStream {
     waker: PlatformWaker,
-    filter: F,
+    filter: Arc<dyn Fn(&Event) -> bool>,
     reader: EventReader,
     stream_wake_task_executed: Arc<AtomicBool>,
     stream_wake_task_should_shutdown: Arc<AtomicBool>,
@@ -42,11 +41,12 @@ struct Task {
     stream_wake_task_should_shutdown: Arc<AtomicBool>,
 }
 
-impl<F> EventStream<F>
-where
-    F: Fn(&Event) -> bool + Clone + Send + Sync + 'static,
-{
-    pub fn new(reader: EventReader, filter: F) -> Self {
+impl EventStream {
+    pub fn new<F>(reader: EventReader, filter: F) -> Self
+    where
+        F: Fn(&Event) -> bool + Send + Sync + 'static,
+    {
+        let filter = Arc::new(filter);
         let waker = reader.waker();
 
         let (task_sender, receiver) = mpsc::sync_channel::<Task>(1);
@@ -56,7 +56,7 @@ where
         thread::spawn(move || {
             while let Ok(task) = receiver.recv() {
                 loop {
-                    if let Ok(true) = task_reader.poll(None, &task_filter) {
+                    if let Ok(true) = task_reader.poll(None, &*task_filter) {
                         break;
                     }
                     if task.stream_wake_task_should_shutdown.load(Ordering::SeqCst) {
@@ -80,7 +80,7 @@ where
     }
 }
 
-impl<F: Fn(&Event) -> bool> Drop for EventStream<F> {
+impl Drop for EventStream {
     fn drop(&mut self) {
         self.stream_wake_task_should_shutdown
             .store(true, Ordering::SeqCst);
@@ -88,12 +88,15 @@ impl<F: Fn(&Event) -> bool> Drop for EventStream<F> {
     }
 }
 
-impl<F: Fn(&Event) -> bool> Stream for EventStream<F> {
+impl Stream for EventStream {
     type Item = io::Result<Event>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match self.reader.poll(Some(Duration::from_secs(0)), &self.filter) {
-            Ok(true) => match self.reader.read(&self.filter) {
+        match self
+            .reader
+            .poll(Some(Duration::from_secs(0)), &*self.filter)
+        {
+            Ok(true) => match self.reader.read(&*self.filter) {
                 Ok(event) => Poll::Ready(Some(Ok(event))),
                 Err(err) => Poll::Ready(Some(Err(err))),
             },