Provide a resume point to bridge clients from state snapshots to the event sequence.

1 files changed, 46 insertions, 21 deletions

diff --git a/src/events/routes/test.rs b/src/events/routes/test.rs
index 1cfca4f..11f01b8 100644
--- a/src/events/routes/test.rs
+++ b/src/events/routes/test.rs
@@ -1,4 +1,5 @@
 use axum::extract::State;
+use axum_extra::extract::Query;
 use futures::{
     future,
     stream::{self, StreamExt as _},
@@ -22,7 +23,7 @@ async fn includes_historical_message() {
 
     let subscriber_creds = fixtures::login::create_with_password(&app).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
-    let routes::Events(events) = routes::events(State(app), subscriber, None)
+    let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
         .expect("subscribe never fails");
 
@@ -49,9 +50,10 @@ async fn includes_live_message() {
 
     let subscriber_creds = fixtures::login::create_with_password(&app).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
-    let routes::Events(events) = routes::events(State(app.clone()), subscriber, None)
-        .await
-        .expect("subscribe never fails");
+    let routes::Events(events) =
+        routes::events(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
 
     // Verify the semantics
 
@@ -94,7 +96,7 @@ async fn includes_multiple_channels() {
 
     let subscriber_creds = fixtures::login::create_with_password(&app).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
-    let routes::Events(events) = routes::events(State(app), subscriber, None)
+    let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
         .expect("subscribe never fails");
 
@@ -130,7 +132,7 @@ async fn sequential_messages() {
 
     let subscriber_creds = fixtures::login::create_with_password(&app).await;
     let subscriber = fixtures::identity::identity(&app, &subscriber_creds, &fixtures::now()).await;
-    let routes::Events(events) = routes::events(State(app), subscriber, None)
+    let routes::Events(events) = routes::events(State(app), subscriber, None, Query::default())
         .await
         .expect("subscribe never fails");
 
@@ -172,9 +174,14 @@ async fn resumes_from() {
 
     let resume_at = {
         // First subscription
-        let routes::Events(events) = routes::events(State(app.clone()), subscriber.clone(), None)
-            .await
-            .expect("subscribe never fails");
+        let routes::Events(events) = routes::events(
+            State(app.clone()),
+            subscriber.clone(),
+            None,
+            Query::default(),
+        )
+        .await
+        .expect("subscribe never fails");
 
         let event = events
             .filter(fixtures::filter::messages())
@@ -189,9 +196,14 @@ async fn resumes_from() {
     };
 
     // Resume after disconnect
-    let routes::Events(resumed) = routes::events(State(app), subscriber, Some(resume_at.into()))
-        .await
-        .expect("subscribe never fails");
+    let routes::Events(resumed) = routes::events(
+        State(app),
+        subscriber,
+        Some(resume_at.into()),
+        Query::default(),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify the structure of the response.
 
@@ -242,9 +254,14 @@ async fn serial_resume() {
         ];
 
         // First subscription
-        let routes::Events(events) = routes::events(State(app.clone()), subscriber.clone(), None)
-            .await
-            .expect("subscribe never fails");
+        let routes::Events(events) = routes::events(
+            State(app.clone()),
+            subscriber.clone(),
+            None,
+            Query::default(),
+        )
+        .await
+        .expect("subscribe never fails");
 
         let events = events
             .filter(fixtures::filter::messages())
@@ -277,6 +294,7 @@ async fn serial_resume() {
             State(app.clone()),
             subscriber.clone(),
             Some(resume_at.into()),
+            Query::default(),
         )
         .await
         .expect("subscribe never fails");
@@ -312,6 +330,7 @@ async fn serial_resume() {
             State(app.clone()),
             subscriber.clone(),
             Some(resume_at.into()),
+            Query::default(),
         )
         .await
         .expect("subscribe never fails");
@@ -345,9 +364,10 @@ async fn terminates_on_token_expiry() {
     let subscriber =
         fixtures::identity::identity(&app, &subscriber_creds, &fixtures::ancient()).await;
 
-    let routes::Events(events) = routes::events(State(app.clone()), subscriber, None)
-        .await
-        .expect("subscribe never fails");
+    let routes::Events(events) =
+        routes::events(State(app.clone()), subscriber, None, Query::default())
+            .await
+            .expect("subscribe never fails");
 
     // Verify the resulting stream's behaviour
 
@@ -387,9 +407,14 @@ async fn terminates_on_logout() {
     let subscriber =
         fixtures::identity::from_token(&app, &subscriber_token, &fixtures::now()).await;
 
-    let routes::Events(events) = routes::events(State(app.clone()), subscriber.clone(), None)
-        .await
-        .expect("subscribe never fails");
+    let routes::Events(events) = routes::events(
+        State(app.clone()),
+        subscriber.clone(),
+        None,
+        Query::default(),
+    )
+    .await
+    .expect("subscribe never fails");
 
     // Verify the resulting stream's behaviour