diff options
| author | Kit La Touche <kit@transneptune.net> | 2024-10-30 16:50:06 -0400 |
|---|---|---|
| committer | Kit La Touche <kit@transneptune.net> | 2024-10-30 16:50:06 -0400 |
| commit | 113096a2cca42008c0a19110abe322180dbdf66b (patch) | |
| tree | cb871dae060e60be7fd2114ee4741027ae38bd78 /src/event/app.rs | |
| parent | 610f6839d2e449d172aa6ac35e6c1de0677a0754 (diff) | |
| parent | 06c839436900ce07ec5c53175b01f3c5011e507c (diff) | |
Merge branch 'main' into wip/mobile
Diffstat (limited to 'src/event/app.rs')
| -rw-r--r-- | src/event/app.rs | 9 |
1 files changed, 4 insertions, 5 deletions
diff --git a/src/event/app.rs b/src/event/app.rs index c754388..b309245 100644 --- a/src/event/app.rs +++ b/src/event/app.rs @@ -6,7 +6,7 @@ use futures::{ use itertools::Itertools as _; use sqlx::sqlite::SqlitePool; -use super::{broadcaster::Broadcaster, Event, ResumePoint, Sequence, Sequenced}; +use super::{broadcaster::Broadcaster, Event, Sequence, Sequenced}; use crate::{ channel::{self, repo::Provider as _}, login::{self, repo::Provider as _}, @@ -26,9 +26,8 @@ impl<'a> Events<'a> { pub async fn subscribe( &self, - resume_at: impl Into<ResumePoint>, + resume_at: Sequence, ) -> Result<impl Stream<Item = Event> + std::fmt::Debug, Error> { - let resume_at = resume_at.into(); // Subscribe before retrieving, to catch messages broadcast while we're // querying the DB. We'll prune out duplicates later. let live_messages = self.events.subscribe(); @@ -63,7 +62,7 @@ impl<'a> Events<'a> { .merge_by(channel_events, Sequence::merge) .merge_by(message_events, Sequence::merge) .collect::<Vec<_>>(); - let resume_live_at = replay_events.last().map(Sequenced::sequence); + let resume_live_at = replay_events.last().map_or(resume_at, Sequenced::sequence); let replay = stream::iter(replay_events); @@ -77,7 +76,7 @@ impl<'a> Events<'a> { Ok(replay.chain(live_messages)) } - fn resume(resume_at: ResumePoint) -> impl for<'m> FnMut(&'m Event) -> future::Ready<bool> { + fn resume(resume_at: Sequence) -> impl for<'m> FnMut(&'m Event) -> future::Ready<bool> { let filter = Sequence::after(resume_at); move |event| future::ready(filter(event)) } |
