lace_tokio/operations/
bounded.rs

1//! The owning bounded interlace: driving to the bounded terminal survives a
2//! dropped borrowing future, and consuming close waits for started work.
3
4use super::*;
5
6/// One bounded interlace. [`run`](Self::run) drives it to its terminal;
7/// dropping that future leaves the operation owned, its observations kept, and
8/// closeable. [`close`](Self::close) cancels it and settles started work.
9#[must_use = "run or close the bounded interlace"]
10pub struct BoundedInterlace {
11    pub(crate) driver: DriverCore,
12    observations: Vec<InterlaceEvent>,
13    /// Route settlement after fixed point started; it cannot resume.
14    settling: bool,
15    outcome: Option<BoundedInterlaceOutcome>,
16}
17
18impl BoundedInterlace {
19    pub fn new(
20        first: impl Into<Participant>,
21        second: impl Into<Participant>,
22        policy: impl IntoPolicy,
23        operation: InterlaceOperationOptions,
24    ) -> Result<Self> {
25        let module = policy.into_policy()?;
26        let operands = [module.clone(), module];
27        Ok(Self::owning(route(
28            first.into(),
29            second.into(),
30            operands,
31            operation.limits,
32        )?))
33    }
34
35    pub(super) fn owning(open: OpenInterlace) -> Self {
36        Self {
37            driver: open.driver,
38            observations: Vec::new(),
39            settling: false,
40            outcome: None,
41        }
42    }
43
44    /// Drive to the bounded terminal, after required final output and route
45    /// settlement. A later call returns the same outcome.
46    pub async fn run(&mut self) -> BoundedInterlaceOutcome {
47        if let Some(outcome) = &self.outcome {
48            return outcome.clone();
49        }
50        let collected = self.collect(drop).await;
51        let outcome = finish_bounded(collected, None, self.driver.peer_diagnostics());
52        self.outcome = Some(outcome.clone());
53        outcome
54    }
55
56    /// Cancel the operation, retire its transport and settle started store
57    /// work through durable position publication. `Ok` is settled local
58    /// cancellation, not convergence; a retained operation failure and any
59    /// found while closing are returned together.
60    pub async fn close(self) -> Result<()> {
61        self.driver.close_cancelled().await.map(drop)
62    }
63
64    /// Drive to the bounded terminal, keeping public observations and handing
65    /// each `First` admitted record to `saved`; return them with the primary
66    /// failure.
67    pub(super) async fn collect(
68        &mut self,
69        mut saved: impl FnMut(Record),
70    ) -> (Vec<InterlaceEvent>, Option<Error>) {
71        let error = loop {
72            if self.settling {
73                let error = Error::integration("bounded route settlement was interrupted");
74                self.driver.fail(error.clone());
75                break Some(error);
76            }
77            match self.driver.next_event().await {
78                Ok(Some(OpenEvent::FixedPoint)) => {
79                    self.settling = true;
80                    let settled = self.driver.settle_route().await;
81                    self.settling = false;
82                    match settled {
83                        Ok(()) => {
84                            self.observations.push(InterlaceEvent::FixedPoint);
85                            break None;
86                        }
87                        Err(error) => break Some(error),
88                    }
89                }
90                Ok(Some(OpenEvent::Closed { .. })) | Ok(None) => break Some(exhausted()),
91                Ok(Some(OpenEvent::RecordSaved {
92                    position,
93                    record_hash,
94                    record,
95                })) => {
96                    self.observations.push(InterlaceEvent::RecordSaved {
97                        position,
98                        record_hash,
99                    });
100                    if position == ParticipantPosition::First {
101                        saved(record);
102                    }
103                }
104                Ok(Some(event)) => self.observations.push(event.into()),
105                Err(error) => break Some(error),
106            }
107        };
108        (std::mem::take(&mut self.observations), error)
109    }
110}
lace_tokio/operations/bounded.rs

lace_tokio/operations/
bounded.rs
