1 files changed, 19 insertions, 14 deletions

diff --git a/src/event/routes/test.rs b/src/event/routes/test.rs
index 249f5c2..e6a8b9d 100644
--- a/src/event/routes/test.rs
+++ b/src/event/routes/test.rs
@@ -31,7 +31,7 @@ async fn includes_historical_message() {
     // Verify the structure of the response.
 
     let event = events
-        .filter(fixtures::filter::messages())
+        .filter_map(fixtures::message::events)
         .next()
         .immediately()
         .await
@@ -62,7 +62,7 @@ async fn includes_live_message() {
     let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
 
     let event = events
-        .filter(fixtures::filter::messages())
+        .filter_map(fixtures::message::events)
         .next()
         .immediately()
         .await
@@ -104,7 +104,7 @@ async fn includes_multiple_channels() {
     // Verify the structure of the response.
 
     let events = events
-        .filter(fixtures::filter::messages())
+        .filter_map(fixtures::message::events)
         .take(messages.len())
         .collect::<Vec<_>>()
         .immediately()
@@ -141,13 +141,15 @@ async fn sequential_messages() {
 
     // Verify the structure of the response.
 
-    let mut events = events.filter(|event| {
-        future::ready(
-            messages
-                .iter()
-                .any(|message| fixtures::event::message_sent(event, message)),
-        )
-    });
+    let mut events = events
+        .filter_map(fixtures::message::events)
+        .filter(|event| {
+            future::ready(
+                messages
+                    .iter()
+                    .any(|message| fixtures::event::message_sent(event, message)),
+            )
+        });
 
     // Verify delivery in order
     for message in &messages {
@@ -193,7 +195,7 @@ async fn resumes_from() {
         .expect("subscribe never fails");
 
         let event = events
-            .filter(fixtures::filter::messages())
+            .filter_map(fixtures::message::events)
             .next()
             .immediately()
             .await
@@ -217,6 +219,7 @@ async fn resumes_from() {
     // Verify the structure of the response.
 
     let events = resumed
+        .filter_map(fixtures::message::events)
         .take(later_messages.len())
         .collect::<Vec<_>>()
         .immediately()
@@ -275,7 +278,7 @@ async fn serial_resume() {
         .expect("subscribe never fails");
 
         let events = events
-            .filter(fixtures::filter::messages())
+            .filter_map(fixtures::message::events)
             .take(initial_messages.len())
             .collect::<Vec<_>>()
             .immediately()
@@ -313,7 +316,7 @@ async fn serial_resume() {
         .expect("subscribe never fails");
 
         let events = events
-            .filter(fixtures::filter::messages())
+            .filter_map(fixtures::message::events)
             .take(resume_messages.len())
             .collect::<Vec<_>>()
             .immediately()
@@ -351,7 +354,7 @@ async fn serial_resume() {
         .expect("subscribe never fails");
 
         let events = events
-            .filter(fixtures::filter::messages())
+            .filter_map(fixtures::message::events)
             .take(final_messages.len())
             .collect::<Vec<_>>()
             .immediately()
@@ -401,6 +404,7 @@ async fn terminates_on_token_expiry() {
     ];
 
     assert!(events
+        .filter_map(fixtures::message::events)
         .filter(|event| future::ready(
             messages
                 .iter()
@@ -452,6 +456,7 @@ async fn terminates_on_logout() {
     ];
 
     assert!(events
+        .filter_map(fixtures::message::events)
         .filter(|event| future::ready(
             messages
                 .iter()