Struct ProjectionCheckpoint 

pub struct ProjectionCheckpoint<C> { /* private fields */ }

English:
Owned cursor identifying progress in a projection’s input stream.

The cursor type and meaning belong to the application. This container performs no ordering, monotonicity, or validity checks and does not store progress by itself.

Borrow a cursor for inspection, then recover its ownership without cloning.

日本語:
プロジェクションの入力ストリームにおける進捗を示す、所有されたカーソルです。

カーソルの型と意味はアプリケーションが定めます。このコンテナーは順序、単調性、妥当性を 検証せず、単体では進捗を保存しません。

カーソルを借用して確認した後、複製せずに所有権を取り出せます。

Example / 使用例

use ddd_toolkit_core::ProjectionCheckpoint;

let checkpoint = ProjectionCheckpoint::new(String::from("partition-2:17"));
assert_eq!(checkpoint.cursor(), "partition-2:17");
let cursor = checkpoint.into_cursor();
assert_eq!(cursor, "partition-2:17");

Implementations

impl<C> ProjectionCheckpoint<C>

pub fn new(cursor: C) -> ProjectionCheckpoint<C>

English:
Wraps cursor without validating its position or comparing previous progress.

日本語:
位置の検証や以前の進捗との比較をせずに cursor を包みます。

Examples found in repository

examples/reliable_delivery.rs (line 226)

async fn deliver(
    outbox: &mut Outbox,
    broker: &mut Broker,
    model: &mut ReadModel,
    checkpoints: &mut Checkpoints,
) -> Result<ProjectionApplication, &'static str> {
    let claim = outbox.claim().await?.ok_or("queue empty")?;
    let (id, event) = claim.into_parts();
    broker.publish(&event).await?;
    // This demo directly feeds a consumer after publishing; no real broker is used.
    // Message IDs are stable across retries; these cursors are not sortable offsets.
    let checkpoint = ProjectionCheckpoint::new(id.as_str().to_owned());
    let result = model.apply_once(&event, &checkpoint).await?;
    checkpoints.save(checkpoint).await?;
    outbox.acknowledge(&id).await?;
    Ok(result)
}

pub fn cursor(&self) -> &C

English:
Borrows the stored cursor without advancing it.

日本語:
進めることなく、保存されたカーソルを借用します。

Examples found in repository

examples/reliable_delivery.rs (line 179)

    fn apply_once<'a>(
        &'a mut self,
        event: &'a Credited,
        checkpoint: &'a ProjectionCheckpoint<String>,
    ) -> PortFuture<'a, ProjectionApplication> {
        Box::pin(async move {
            if self.applied.contains(checkpoint.cursor()) {
                return Ok(ProjectionApplication::AlreadyApplied);
            }
            let balance = self
                .balance
                .checked_add(event.0)
                .ok_or("balance overflow")?;
            // Deduplication marker + read model update must be atomic in production too.
            self.applied.insert(checkpoint.cursor().clone());
            self.balance = balance;
            Ok(ProjectionApplication::Applied)
        })
    }
}
#[derive(Default)]
struct Checkpoints {
    last: Option<ProjectionCheckpoint<String>>,
    fail_save: bool,
}
impl AsyncProjectionCheckpointStore for Checkpoints {
    type Cursor = String;
    type Error = &'static str;
    fn load<'a>(&'a self) -> PortFuture<'a, Option<ProjectionCheckpoint<String>>> {
        Box::pin(async move { Ok(self.last.clone()) })
    }
    fn save<'a>(&'a mut self, checkpoint: ProjectionCheckpoint<String>) -> PortFuture<'a, ()> {
        Box::pin(async move {
            if self.fail_save {
                return Err("checkpoint unavailable");
            }
            self.last = Some(checkpoint);
            Ok(())
        })
    }
}

async fn deliver(
    outbox: &mut Outbox,
    broker: &mut Broker,
    model: &mut ReadModel,
    checkpoints: &mut Checkpoints,
) -> Result<ProjectionApplication, &'static str> {
    let claim = outbox.claim().await?.ok_or("queue empty")?;
    let (id, event) = claim.into_parts();
    broker.publish(&event).await?;
    // This demo directly feeds a consumer after publishing; no real broker is used.
    // Message IDs are stable across retries; these cursors are not sortable offsets.
    let checkpoint = ProjectionCheckpoint::new(id.as_str().to_owned());
    let result = model.apply_once(&event, &checkpoint).await?;
    checkpoints.save(checkpoint).await?;
    outbox.acknowledge(&id).await?;
    Ok(result)
}
fn main() {
    let mut account = Account {
        id: 1,
        balance: 10,
        events: vec![Credited(10)],
    };
    let mut outbox = Outbox::default();
    let mut context = AsyncTransactionalOutboxLifecycleContext::new(
        &mut account,
        &mut outbox,
        ExpectedRevision::NoStream(NoStream),
    );
    ready(context.save_and_enqueue()).unwrap();
    context.complete().unwrap();
    assert!(account.pull_events().is_empty());
    assert_eq!(outbox.saved.as_ref().unwrap().balance, 10);
    let mut broker = Broker {
        fail: true,
        ..Broker::default()
    };
    let mut model = ReadModel::default();
    let mut checkpoints = Checkpoints::default();
    assert_eq!(
        ready(deliver(
            &mut outbox,
            &mut broker,
            &mut model,
            &mut checkpoints
        )),
        Err("broker down")
    );
    assert_eq!(outbox.queue.len(), 1);
    assert_eq!(model.balance, 0);

    broker.fail = false;
    checkpoints.fail_save = true;
    assert_eq!(
        ready(deliver(
            &mut outbox,
            &mut broker,
            &mut model,
            &mut checkpoints
        )),
        Err("checkpoint unavailable")
    );
    assert_eq!(model.balance, 10);
    assert!(ready(checkpoints.load()).unwrap().is_none());
    checkpoints.fail_save = false;
    outbox.fail_ack = true;
    assert_eq!(
        ready(deliver(
            &mut outbox,
            &mut broker,
            &mut model,
            &mut checkpoints
        )),
        Err("acknowledgement lost")
    );
    assert_eq!(model.balance, 10);
    assert_eq!(outbox.queue.len(), 1);
    outbox.fail_ack = false;
    assert_eq!(
        ready(deliver(
            &mut outbox,
            &mut broker,
            &mut model,
            &mut checkpoints
        )),
        Ok(ProjectionApplication::AlreadyApplied)
    );
    assert_eq!(model.balance, 10);
    assert_eq!(broker.accepted.len(), 3); // At-least-once delivery, not exactly-once.
    assert_eq!(ready(checkpoints.load()).unwrap().unwrap().cursor(), "1:0");
    assert!(ready(outbox.claim()).unwrap().is_none());
    println!("reliable_delivery: retried broker/checkpoint/ack failures; 3 deliveries, balance=10");
}

pub fn into_cursor(self) -> C

English:
Consumes the checkpoint and returns its cursor by value.

日本語:
チェックポイントを消費し、カーソルを値として返します。

Trait Implementations

impl<C> Clone for ProjectionCheckpoint<C>

where C: Clone,

fn clone(&self) -> ProjectionCheckpoint<C>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<C> Debug for ProjectionCheckpoint<C>

where C: Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl<C> Eq for ProjectionCheckpoint<C>

where C: Eq,

impl<C> Hash for ProjectionCheckpoint<C>

where C: Hash,

fn hash<__H>(&self, state: &mut __H)

where __H: Hasher,

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl<C> PartialEq for ProjectionCheckpoint<C>

where C: PartialEq,

fn eq(&self, other: &ProjectionCheckpoint<C>) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl<C> StructuralPartialEq for ProjectionCheckpoint<C>

where C: PartialEq,