use super::{
Authority, Change, Error, Guarantee, OperationId, PersistentRow, ReadOutcome, ReadRequest,
Revision,
};
use crate::StepInputOf;
use crate::storage::warp_table::WarpTable;
use alloc::vec::Vec;
pub trait TableProtocol<Row: PersistentRow> {
const SCHEMA: [u8; 16];
const VERSION: u32;
type Query;
type Value;
type Mutation;
type Outcome;
type Refusal;
fn query(query: &Self::Query, table: &WarpTable<Row>) -> Result<Self::Value, Self::Refusal>;
fn prepare(
command: &Self::Mutation,
table: &WarpTable<Row>,
changes: &mut ChangeSet<Row>,
) -> Result<Self::Outcome, Self::Refusal>;
}
pub struct PrepareCommand<'a, Row: PersistentRow, P: TableProtocol<Row>> {
pub operation: OperationId,
pub guarantee: Guarantee,
pub command: &'a P::Mutation,
}
pub struct PreparedCommand<Row: PersistentRow, Outcome, Refusal> {
pub expected: Revision,
pub changes: Vec<Change<Row, Row::Id>>,
pub outcome: Result<Outcome, Refusal>,
}
impl<Row: PersistentRow, P: TableProtocol<Row>> StepInputOf<Authority<Row>>
for PrepareCommand<'_, Row, P>
{
type Outcome = Result<PreparedCommand<Row, P::Outcome, P::Refusal>, Error>;
fn step(self, machine: &mut Authority<Row>) -> Self::Outcome {
machine.admit_command(self.operation, self.guarantee)?;
let mut changes = ChangeSet::new(machine.limits().changes)?;
let table = match machine.read(ReadRequest {
guarantee: Guarantee::Accepted,
}) {
ReadOutcome::Ready { table, .. } => table,
ReadOutcome::Pending => return Err(Error::Busy),
ReadOutcome::Refused(error) => return Err(error),
};
let outcome = P::prepare(self.command, table, &mut changes);
let mut changes = changes.finish()?;
if outcome.is_err() {
changes.clear();
}
Ok(PreparedCommand {
expected: machine.revision(),
changes,
outcome,
})
}
}
pub struct ChangeSet<Row: PersistentRow> {
changes: Vec<Change<Row, Row::Id>>,
limit: usize,
refused: bool,
}
impl<Row: PersistentRow> ChangeSet<Row> {
pub fn new(limit: usize) -> Result<Self, Error> {
if limit == 0 {
return Err(Error::InvalidBounds);
}
let mut changes = Vec::new();
changes
.try_reserve_exact(limit)
.map_err(|_| Error::Allocation)?;
Ok(Self {
changes,
limit,
refused: false,
})
}
pub fn push(&mut self, change: Change<Row, Row::Id>) -> Result<(), Change<Row, Row::Id>> {
if self.changes.len() == self.limit {
self.refused = true;
return Err(change);
}
self.changes.push(change);
Ok(())
}
pub fn finish(self) -> Result<Vec<Change<Row, Row::Id>>, Error> {
if self.refused {
Err(Error::Full)
} else {
Ok(self.changes)
}
}
}