pipecircuit 0.0.20

Correct. Robust. Fast. Portable. Pick four.
Documentation 
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

use super::*;
use crate::storage::warp_table::observed::AppliedViewChange;

#[must_use]
#[derive(Debug, PartialEq, Eq)]
pub enum SubscriptionChanges<'report, Id> {
    Applied {
        changes: &'report [AppliedViewChange<Id>],
    },
    Subscribed,
    Resynchronizing,
    Stale,
    BaselineRequired,
    Refused(SubscriptionRefusal),
}

pub struct ReceiveSubscriptionChanges<'message, 'report, Value, Outcome, Refusal, Id, Projection> {
    receive: ReceiveSubscription<'message, Value, Outcome, Refusal, Id, Projection>,
    report: &'report mut AppliedViewChanges<Id>,
}

impl<'message, Value, Outcome, Refusal, Id, Projection>
    ReceiveSubscription<'message, Value, Outcome, Refusal, Id, Projection>
{
    pub fn with_changes<'report>(
        self,
        report: &'report mut AppliedViewChanges<Id>,
    ) -> ReceiveSubscriptionChanges<'message, 'report, Value, Outcome, Refusal, Id, Projection>
    {
        ReceiveSubscriptionChanges {
            receive: self,
            report,
        }
    }
}

impl<'report, Value, Outcome, Refusal, Id: IndexKey, Projection: Clone + Eq>
    StepInputOf<SubscribedView<Id, Projection>>
    for ReceiveSubscriptionChanges<'_, 'report, Value, Outcome, Refusal, Id, Projection>
{
    type Outcome = Result<SubscriptionChanges<'report, Id>, SubscriptionReceiveError>;

    fn step(self, subscriber: &mut SubscribedView<Id, Projection>) -> Self::Outcome {
        self.report.clear();
        Ok(match self.receive.receive(subscriber, Some(self.report))? {
            SubscriptionProgress::Updated(ViewApplyOutcome::Applied) => {
                SubscriptionChanges::Applied {
                    changes: self.report.as_slice(),
                }
            }
            SubscriptionProgress::Updated(ViewApplyOutcome::Stale) => SubscriptionChanges::Stale,
            SubscriptionProgress::Updated(ViewApplyOutcome::BaselineRequired) => {
                SubscriptionChanges::BaselineRequired
            }
            SubscriptionProgress::Subscribed => SubscriptionChanges::Subscribed,
            SubscriptionProgress::Resynchronizing => SubscriptionChanges::Resynchronizing,
            SubscriptionProgress::Refused(reason) => SubscriptionChanges::Refused(reason),
        })
    }
}