Struct AsyncTransactionalOutboxLifecycleContext 

pub struct AsyncTransactionalOutboxLifecycleContext<'storage, A, O>
where
    A: AggregateRoot + Send + Sync,
    <A as AggregateRoot>::Event: Send,
    O: AsyncTransactionalOutbox<Aggregate = A> + Send,
    <O as AsyncTransactionalOutbox>::Error: Send,
{ /* private fields */ }

English:
Single-attempt asynchronous enqueue tracking with exclusive aggregate and outbox borrows.

Calling Self::save_and_enqueue consumes the attempt and pulls events immediately, before its future is polled. Failure or dropping that future does not restore events or enable retry. Only a successful adapter result records success. Self::complete checks this record; it does not commit a transaction. Dropping the context performs no check or rollback. Transaction and cancellation effects on storage belong to the outbox adapter.

The aggregate must be Send + Sync, its event must be Send, and the outbox and error must be Send to support the returned Send future. No executor or locking is provided.

Await a successful enqueue before consuming the context to check completion. The assertion follows from recording success after the adapter returns Ok.

日本語:
集約とアウトボックスを排他的に借用し、非同期登録の一回の試行を追跡します。

Self::save_and_enqueue の呼び出し時に、Future のポーリング前に試行を消費して イベントを取り出します。失敗や Future の破棄でもイベントは復元せず、再試行もできません。 アダプターの成功結果だけが成功を記録します。Self::complete はこの記録を検査するだけで、 コミットしません。コンテキストの破棄でも検査やロールバックは行いません。 トランザクションとキャンセルがストレージに及ぼす効果はアダプターが定めます。

Send な Future のため、集約には Send + Sync、イベント、アウトボックス、エラーには Send が必要です。実行器やロックは提供しません。

登録成功を待ってからコンテキストを消費して完了確認します。 アダプターが Ok を返した後に成功を記録するため、以下のアサーションが成立します。

Example / 使用例

use ddd_toolkit_core::{
    AggregateRoot, AppendResult, AsyncTransactionalOutbox,
    AsyncTransactionalOutboxLifecycleContext, AsyncTransactionalOutboxOperationError,
    ExpectedRevision,
};

async fn enqueue<A, O>(
    aggregate: &mut A,
    outbox: &mut O,
    expected: ExpectedRevision,
) -> Result<AppendResult, AsyncTransactionalOutboxOperationError<O::Error>>
where
    A: AggregateRoot + Send + Sync,
    A::Event: Send,
    O: AsyncTransactionalOutbox<Aggregate = A> + Send,
{
    let mut context = AsyncTransactionalOutboxLifecycleContext::new(aggregate, outbox, expected);
    let result = context.save_and_enqueue().await?;
    assert_eq!(context.complete(), Ok(()));
    Ok(result)
}

Implementations

impl<'storage, A, O> AsyncTransactionalOutboxLifecycleContext<'storage, A, O>

where A: AggregateRoot + Send + Sync, <A as AggregateRoot>::Event: Send, O: AsyncTransactionalOutbox<Aggregate = A> + Send, <O as AsyncTransactionalOutbox>::Error: Send,

pub fn new( aggregate: &'storage mut A, outbox: &'storage mut O, expected: ExpectedRevision, ) -> AsyncTransactionalOutboxLifecycleContext<'storage, A, O>

English:
Borrows the aggregate and outbox for 'storage and stores expected unchanged. The context starts pending; construction does not pull events or call the adapter.

日本語:
集約とアウトボックスを 'storage の間借用し、expected を変更せず保持します。 未試行で開始し、構築時にはイベントを取り出さず、アダプターも呼び出しません。

Examples found in repository

examples/reliable_delivery.rs (lines 239-243)

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 save_and_enqueue<'ctx>( &'ctx mut self, ) -> Pin<Box<dyn Future<Output = Result<AppendResult, AsyncTransactionalOutboxOperationError<<O as AsyncTransactionalOutbox>::Error>>> + Send + 'ctx>>

where 'storage: 'ctx, A: 'ctx, <A as AggregateRoot>::Event: 'ctx, O: 'ctx, <O as AsyncTransactionalOutbox>::Error: 'ctx,

English:
Immediately reserves the attempt and pulls events, then returns a boxed Send future.

The future borrows this context for 'ctx. Polling it calls the adapter with the borrowed aggregate, stored revision, and owned batch, including an empty batch. It returns the AppendResult or wraps the unchanged AppendError. Dropping the future before its first poll discards the pulled batch without calling the adapter; dropping it while pending leaves the attempt unsuccessful, with storage effects adapter-defined. Neither path restores events. A later call returns a future yielding AsyncTransactionalOutboxOperationError::AlreadyEnqueued.

Panics from pulling events or invoking/polling the adapter propagate. State is marked unsuccessful before pulling; catching a panic does not enable another attempt.

日本語:
呼び出し時に試行を確保してイベントを取り出し、ボックス化された Send Future を返します。

Future はコンテキストを 'ctx の間借用します。ポーリング時に、借用した集約、保存済み リビジョン、所有されたバッチをアダプターへ渡します。空バッチも渡します。 AppendResult を返すか、変更していない AppendError を包んで返します。 初回ポーリング前の破棄ではアダプターを呼ばずにバッチを破棄します。待機中の破棄では 試行が未成功のままとなり、ストレージへの効果はアダプターに依存します。 どちらもイベントを復元しません。後続の呼び出しは AsyncTransactionalOutboxOperationError::AlreadyEnqueued を返す Future を返します。

イベントの取り出しやアダプターの呼び出し・ポーリングのパニックは伝播します。 取り出し前に未成功と記録するため、パニックを捕捉しても再試行できません。

Examples found in repository

examples/reliable_delivery.rs (line 244)

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 complete(self) -> Result<(), AsyncTransactionalOutboxCompletionError>

English:
Consumes the context and returns Ok(()) only after recorded adapter success. No enqueue call yields AsyncTransactionalOutboxCompletionError::MissingEnqueue; a call without recorded success yields AsyncTransactionalOutboxCompletionError::EnqueueFailedConsumed. This makes no storage call. Rejection of a repeated attempt does not erase an earlier success. Any returned enqueue future must release its mutable borrow before this method is called.

日本語:
コンテキストを消費し、アダプターの成功を記録済みの場合のみ Ok(()) を返します。 登録未呼び出しなら AsyncTransactionalOutboxCompletionError::MissingEnqueue、呼び出し済みで 未成功なら AsyncTransactionalOutboxCompletionError::EnqueueFailedConsumed を返します。 ストレージは呼び出しません。再試行の拒否は以前の成功を消しません。 呼び出す前に、登録の Future が保持する可変借用を解放する必要があります。

Examples found in repository

examples/reliable_delivery.rs (line 245)

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");
}