use super::channel::{Message, body};
use super::listen_config::ListenConfig;
use liminal_sdk::DeliveredMessage;
use std::collections::BTreeMap;
use std::sync::{Condvar, Mutex};
struct Content {
text: Option<String>,
size: usize,
}
impl Content {
fn bytes(&self) -> usize {
self.text.as_ref().map_or(0, String::len)
}
fn render(self, message: &Message) -> String {
match self.text {
Some(text) => text,
None => format!(
"{} bytes of {} on {}",
self.size, message.declared_type, message.subject
),
}
}
}
#[derive(Default)]
struct State {
held: BTreeMap<(u64, u64), Content>,
bytes: usize,
boot: Option<u64>,
ended: Option<String>,
closed: bool,
blocked: bool,
}
#[derive(Default)]
pub(super) struct Join {
state: Mutex<State>,
changed: Condvar,
}
impl Join {
pub fn offer(&self, delivery: &DeliveredMessage, config: &ListenConfig) -> Result<(), String> {
let id = delivery
.message()
.ok_or("the channel delivery has no id to join to the tap")?;
let text = if delivery.payload().len() <= config.content_bytes {
std::str::from_utf8(delivery.payload())
.ok()
.map(str::to_owned)
} else {
None
};
let content = Content {
text,
size: delivery.payload().len(),
};
if content.bytes() > config.join_bytes {
return Err(format!(
"one delivery exceeds join_bytes {}",
config.join_bytes
));
}
let mut state = self.state.lock().map_err(|_| "the join lock is poisoned")?;
while !state.closed
&& (state.held.len() >= config.join_bound
|| state.bytes.saturating_add(content.bytes()) > config.join_bytes)
{
state.blocked = true;
self.changed.notify_all();
state = self
.changed
.wait(state)
.map_err(|_| "the join lock is poisoned")?;
}
state.blocked = false;
if state.closed {
return Err("the listen was closed".to_owned());
}
if state.boot.is_some_and(|boot| boot != id.boot) {
return Err("the delivery boot changed within one listen".to_owned());
}
state.boot = Some(id.boot);
let key = (id.boot, id.id);
if state.held.contains_key(&key) {
return Err("the channel repeated an unmatched delivery id".to_owned());
}
state.bytes = state
.bytes
.checked_add(content.bytes())
.ok_or("the join byte count overflowed")?;
state.held.insert(key, content);
drop(state);
self.changed.notify_all();
Ok(())
}
pub fn content(&self, message: &Message, config: &ListenConfig) -> Result<String, String> {
if let Some(text) = &message.payload {
if u64::try_from(text.len()).map_err(|_| "the tap size cannot fit u64")? != message.size
{
return Err("the tap payload size differs from its provenance".to_owned());
}
return Ok(body(text.as_bytes(), message, config.content_bytes));
}
let id = message
.id
.rsplit_once(':')
.and_then(|(_, id)| id.parse::<u64>().ok())
.ok_or("the tap id does not name a channel delivery")?;
let mut state = self.state.lock().map_err(|_| "the join lock is poisoned")?;
loop {
if state.closed {
return Err("the listen was closed".to_owned());
}
let matching = state
.held
.keys()
.find(|(_, candidate)| *candidate == id)
.copied();
if let Some(key) = matching {
let content = state
.held
.remove(&key)
.ok_or("the joined delivery disappeared")?;
state.bytes = state
.bytes
.checked_sub(content.bytes())
.ok_or("the join byte count underflowed")?;
drop(state);
self.changed.notify_all();
if u64::try_from(content.size).map_err(|_| "the delivery size cannot fit u64")?
!= message.size
{
return Err("the joined delivery size differs from the tap".to_owned());
}
return Ok(content.render(message));
}
if let Some(reason) = &state.ended {
return Err(reason.clone());
}
if state.blocked {
return Err(format!(
"delivery id {id} is absent before the join bound: {} messages, {} bytes",
config.join_bound, config.join_bytes
));
}
state = self
.changed
.wait(state)
.map_err(|_| "the join lock is poisoned")?;
}
}
pub fn end(&self, reason: String) -> Result<(), String> {
let mut state = self.state.lock().map_err(|_| "the join lock is poisoned")?;
if state.ended.is_none() {
state.ended = Some(reason);
}
drop(state);
self.changed.notify_all();
Ok(())
}
pub fn reason(&self) -> Result<Option<String>, String> {
self.state
.lock()
.map(|state| state.ended.clone())
.map_err(|_| "the join lock is poisoned".to_owned())
}
pub fn close(&self) -> Result<(), String> {
let mut state = self.state.lock().map_err(|_| "the join lock is poisoned")?;
state.closed = true;
drop(state);
self.changed.notify_all();
Ok(())
}
}