use crate::storage::warp_table::{
Capacity, WarpId, WarpTable, WarpTableCreationError, WarpTableInsertOutcome,
WarpTableRemoveOutcome, WarpTableRow,
};
use crate::{TimerArmDecision, TimerArmOutcome, TimerArmRefusal, TimerIn, TimerToken, WakeAfter};
use super::Delayed;
#[derive(WarpTableRow)]
struct PendingDelayedMessage<Message> {
message: Message,
}
type PendingDelayedMessageCreationError<Message> = WarpTableCreationError<
<PendingDelayedMessage<Message> as WarpTableRow>::ColumnsCreationError,
<PendingDelayedMessage<Message> as WarpTableRow>::IndexesCreationError,
>;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DelayLedgerCreationFailure {
StorageUnavailable,
}
#[derive(Debug)]
pub struct DelayLedgerCreationError<Message> {
failure: DelayLedgerCreationFailure,
_source: PendingDelayedMessageCreationError<Message>,
}
impl<Message> DelayLedgerCreationError<Message> {
pub const fn failure(&self) -> DelayLedgerCreationFailure {
self.failure
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DelayUnavailable {
AtCapacity { maximum_pending: u32 },
IdentifiersExhausted,
WakeRefused { reason: TimerArmRefusal },
}
pub struct DelayLedger<Message> {
pending: WarpTable<PendingDelayedMessage<Message>>,
}
impl<Message> DelayLedger<Message> {
pub fn try_new(capacity: Capacity) -> Result<Self, DelayLedgerCreationError<Message>> {
let pending = WarpTable::try_new(capacity).map_err(|source| DelayLedgerCreationError {
failure: DelayLedgerCreationFailure::StorageUnavailable,
_source: source,
})?;
Ok(Self { pending })
}
pub const fn capacity(&self) -> Capacity {
self.pending.capacity()
}
pub const fn len(&self) -> usize {
self.pending.len()
}
pub const fn is_empty(&self) -> bool {
self.pending.is_empty()
}
pub fn admit(&mut self, delayed: Delayed<Message>) -> DelayAdmissionOutcome<Message> {
let (message, wake_after) = delayed.into_parts();
let inserted = self.pending.insert(PendingDelayedMessage { message });
self.admission_outcome(inserted, wake_after)
}
fn admission_outcome(
&self,
inserted: WarpTableInsertOutcome<
WarpId<PendingDelayedMessage<Message>>,
PendingDelayedMessage<Message>,
>,
wake_after: WakeAfter,
) -> DelayAdmissionOutcome<Message> {
match inserted {
WarpTableInsertOutcome::Inserted { id, .. } => DelayAdmissionOutcome::Admitted {
admission: DelayAdmission {
token: TimerToken::from_nonzero(id.value()),
wake_after,
},
},
WarpTableInsertOutcome::Full {
rejected: PendingDelayedMessage { message },
} => DelayAdmissionOutcome::Refused {
reason: DelayUnavailable::AtCapacity {
maximum_pending: self.capacity().value().get(),
},
rejected: Delayed::new(message, wake_after),
},
WarpTableInsertOutcome::IdentifiersExhausted {
rejected: PendingDelayedMessage { message },
} => DelayAdmissionOutcome::Refused {
reason: DelayUnavailable::IdentifiersExhausted,
rejected: Delayed::new(message, wake_after),
},
WarpTableInsertOutcome::UniqueConflict {
conflict,
rejected: _,
} => match conflict {},
}
}
pub fn refuse(&mut self, refusal: DelayArmRefusal) -> DelayRefusalOutcome<Message> {
match self.pending.remove(WarpId::new(refusal.token.value())) {
WarpTableRemoveOutcome::Removed {
removed: PendingDelayedMessage { message },
..
} => DelayRefusalOutcome::Refused {
reason: DelayUnavailable::WakeRefused {
reason: refusal.reason,
},
rejected: Delayed::new(message, refusal.wake_after),
},
WarpTableRemoveOutcome::Absent { .. } => DelayRefusalOutcome::Stale { refusal },
}
}
pub fn withdraw(&mut self, admission: DelayAdmission) -> DelayWithdrawalOutcome<Message> {
match self.pending.remove(WarpId::new(admission.token.value())) {
WarpTableRemoveOutcome::Removed {
removed: PendingDelayedMessage { message },
..
} => DelayWithdrawalOutcome::Withdrawn {
rejected: Delayed::new(message, admission.wake_after),
},
WarpTableRemoveOutcome::Absent { .. } => DelayWithdrawalOutcome::Absent { admission },
}
}
pub fn release(&mut self, token: TimerToken) -> DelayReleaseOutcome<Message> {
match self.pending.remove(WarpId::new(token.value())) {
WarpTableRemoveOutcome::Removed {
removed: PendingDelayedMessage { message },
..
} => DelayReleaseOutcome::Released { message },
WarpTableRemoveOutcome::Absent { .. } => DelayReleaseOutcome::Stale { token },
}
}
}
#[cfg(test)]
mod tests;
#[must_use = "admitted delayed work or exact refusal must be handled by the fitting"]
pub enum DelayAdmissionOutcome<Message> {
Admitted {
admission: DelayAdmission,
},
Refused {
reason: DelayUnavailable,
rejected: Delayed<Message>,
},
}
#[must_use = "delayed-message admission must be armed before returning"]
pub struct DelayAdmission {
token: TimerToken,
wake_after: WakeAfter,
}
impl DelayAdmission {
pub const fn timer_input(&self) -> TimerIn<TimerToken> {
TimerIn::Arm {
token: self.token,
after: self.wake_after,
}
}
pub fn settle_timer_arm(
self,
outcome: TimerArmOutcome<TimerToken>,
) -> Result<DelayArmOutcome, DelayArmMismatch> {
let decision = outcome.into_decision();
let matches_admission = match decision {
TimerArmDecision::Armed { token } => token == self.token,
TimerArmDecision::Refused { token, after, .. } => {
token == self.token && after == self.wake_after
}
};
if !matches_admission {
return Err(DelayArmMismatch {
admission: self,
decision,
});
}
match decision {
TimerArmDecision::Armed { .. } => Ok(DelayArmOutcome::Armed),
TimerArmDecision::Refused { reason, .. } => Ok(DelayArmOutcome::Refused {
refusal: DelayArmRefusal {
token: self.token,
wake_after: self.wake_after,
reason,
},
}),
}
}
}
#[must_use = "armed delayed work or exact timer refusal must be handled by the fitting"]
pub enum DelayArmOutcome {
Armed,
Refused { refusal: DelayArmRefusal },
}
pub struct DelayArmMismatch {
admission: DelayAdmission,
decision: TimerArmDecision<TimerToken>,
}
impl DelayArmMismatch {
pub fn into_parts(self) -> (DelayAdmission, TimerArmDecision<TimerToken>) {
(self.admission, self.decision)
}
}
#[must_use = "timer refusal must be returned to the delay ledger"]
pub struct DelayArmRefusal {
token: TimerToken,
wake_after: WakeAfter,
reason: TimerArmRefusal,
}
impl DelayArmRefusal {
pub const fn reason(&self) -> TimerArmRefusal {
self.reason
}
}
#[must_use = "resolved or stale timer refusal must be handled by the fitting"]
pub enum DelayRefusalOutcome<Message> {
Refused {
reason: DelayUnavailable,
rejected: Delayed<Message>,
},
Stale {
refusal: DelayArmRefusal,
},
}
#[must_use = "released or stale timer tokens must be handled by the fitting"]
pub enum DelayReleaseOutcome<Message> {
Released { message: Message },
Stale { token: TimerToken },
}
#[must_use = "withdrawn delayed work or an absent admission must be handled by the fitting"]
pub enum DelayWithdrawalOutcome<Message> {
Withdrawn { rejected: Delayed<Message> },
Absent { admission: DelayAdmission },
}