1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
use axum::{
extract::State,
response::{
sse::{self, Sse},
IntoResponse,
},
routing::get,
Router,
};
use axum_extra::extract::Query;
use chrono::{format::SecondsFormat, DateTime};
use futures::{
future,
stream::{self, StreamExt as _, TryStreamExt as _},
};
use crate::{
app::App,
channel::repo::broadcast,
error::{BoxedError, InternalError},
header::LastEventId,
repo::{channel, login::Login},
};
pub fn router() -> Router<App> {
Router::new().route("/events", get(on_events))
}
#[derive(serde::Deserialize)]
struct EventsQuery {
#[serde(default, rename = "channel")]
channels: Vec<channel::Id>,
}
async fn on_events(
State(app): State<App>,
_: Login, // requires auth, but doesn't actually care who you are
last_event_id: Option<LastEventId>,
Query(query): Query<EventsQuery>,
) -> Result<impl IntoResponse, InternalError> {
let resume_at = last_event_id
.map(|LastEventId(header)| header)
.map(|header| DateTime::parse_from_rfc3339(&header))
.transpose()?
.map(|ts| ts.to_utc());
let streams = stream::iter(query.channels)
.then(|channel| {
let app = app.clone();
async move {
let events = app
.channels()
.events(&channel, resume_at.as_ref())
.await?
.map_ok(ChannelEvent::wrap(channel));
Ok::<_, BoxedError>(events)
}
})
.try_collect::<Vec<_>>()
.await?;
let stream = stream::select_all(streams).and_then(|msg| future::ready(to_sse_event(msg)));
let sse = Sse::new(stream).keep_alive(sse::KeepAlive::default());
Ok(sse)
}
fn to_sse_event(event: ChannelEvent<broadcast::Message>) -> Result<sse::Event, BoxedError> {
let data = serde_json::to_string(&event)?;
let event = sse::Event::default()
.id(event
.message
.sent_at
.to_rfc3339_opts(SecondsFormat::AutoSi, /* use_z */ true))
.data(&data);
Ok(event)
}
#[derive(serde::Serialize)]
struct ChannelEvent<M> {
channel: channel::Id,
#[serde(flatten)]
message: M,
}
impl<M> ChannelEvent<M> {
fn wrap(channel: channel::Id) -> impl Fn(M) -> Self {
move |message| Self {
channel: channel.clone(),
message,
}
}
}
|
