From e6be82157fe718570aa13ab12803ee39083b8dff Mon Sep 17 00:00:00 2001 From: Owen Jacobson Date: Mon, 16 Sep 2024 11:15:23 -0400 Subject: Some code cleanup on events --- src/events.rs | 46 +++++++++------------------------------------- 1 file changed, 9 insertions(+), 37 deletions(-) (limited to 'src/events.rs') diff --git a/src/events.rs b/src/events.rs index 38d53fc..9b5901e 100644 --- a/src/events.rs +++ b/src/events.rs @@ -1,6 +1,5 @@ use axum::{ extract::State, - http::{HeaderName, HeaderValue}, response::{ sse::{self, Sse}, IntoResponse, @@ -8,7 +7,7 @@ use axum::{ routing::get, Router, }; -use axum_extra::{extract::Query, typed_header::TypedHeader}; +use axum_extra::extract::Query; use chrono::{format::SecondsFormat, DateTime}; use futures::{ future, @@ -19,6 +18,7 @@ use crate::{ app::App, channel::repo::broadcast, error::{BoxedError, InternalError}, + header::LastEventId, repo::{channel, login::Login}, }; @@ -35,11 +35,10 @@ struct EventsQuery { async fn on_events( State(app): State, _: Login, // requires auth, but doesn't actually care who you are - last_event_id: Option>, + last_event_id: Option, Query(query): Query, ) -> Result { let resume_at = last_event_id - .map(|TypedHeader(header)| header) .map(|LastEventId(header)| header) .map(|header| DateTime::parse_from_rfc3339(&header)) .transpose()? @@ -53,10 +52,7 @@ async fn on_events( .channels() .events(&channel, resume_at.as_ref()) .await? - .map_ok(move |message| ChannelEvent { - channel: channel.clone(), - message, - }); + .map_ok(ChannelEvent::wrap(channel)); Ok::<_, BoxedError>(events) } @@ -89,35 +85,11 @@ struct ChannelEvent { message: M, } -pub struct LastEventId(pub String); - -static LAST_EVENT_ID: HeaderName = HeaderName::from_static("last-event-id"); - -impl headers::Header for LastEventId { - fn name() -> &'static HeaderName { - &LAST_EVENT_ID - } - - fn decode<'i, I>(values: &mut I) -> Result - where - I: Iterator, - { - let value = values.next().ok_or_else(headers::Error::invalid)?; - if let Ok(value) = value.to_str() { - Ok(Self(value.into())) - } else { - Err(headers::Error::invalid()) +impl ChannelEvent { + fn wrap(channel: channel::Id) -> impl Fn(M) -> Self { + move |message| Self { + channel: channel.clone(), + message, } } - - fn encode(&self, values: &mut E) - where - E: Extend, - { - let Self(value) = self; - // Must panic or suppress; the trait provides no other options. - let value = HeaderValue::from_str(value).expect("LastEventId is a valid header value"); - - values.extend(std::iter::once(value)); - } } -- cgit v1.2.3