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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
//! A durable, at-least-once delivery queue backed by a [`variable::Journal`](crate::journal::contiguous::variable).
//!
//! [Queue] provides a persistent message queue with at-least-once delivery semantics.
//! Items are durably stored in a journal and will survive crashes. [Queue::init] returns the
//! queue, which writes items, and its [Reader], which delivers committed items. The reader must
//! explicitly acknowledge each item after processing. On restart, all non-pruned items are
//! re-delivered (acknowledged or not).
//!
//! # Ownership
//!
//! Methods that write to storage (`append`, `enqueue`, `commit`, `sync`) take the queue by value
//! and return it on success. If one returns an error, or its future is dropped before it finishes,
//! the queue is gone: state that was not yet durable is discarded, but everything already on disk
//! stays recoverable. The reader then delivers every published item and returns `None`. Reader
//! methods borrow the reader. A failed read does not invalidate it.
//!
//! # Concurrency
//!
//! The queue and its reader can run in separate tasks. The reader never waits on the queue, and
//! [Reader::recv] integrates with `select!` for multiplexing with other futures.
//!
//! ```rust,ignore
//! use commonware_storage::queue::Queue;
//! use commonware_macros::select;
//!
//! let (queue, mut reader) = Queue::init(context, config).await?;
//!
//! // Writer task
//! let (queue, position) = queue.enqueue(&item).await?;
//!
//! // Reader task
//! loop {
//! select! {
//! result = reader.recv() => {
//! let Some((pos, item)) = result? else { break };
//! // Process item...
//! reader.ack(pos)?;
//! }
//! _ = shutdown => break,
//! }
//! }
//! ```
//!
//! # Example
//!
//! ```rust
//! use commonware_codec::RangeCfg;
//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
//! use commonware_storage::{queue::{Queue, Config}};
//! use std::num::{NonZeroU16, NonZeroU64, NonZeroUsize};
//!
//! let executor = deterministic::Runner::default();
//! executor.start(|context| async move {
//! // Create a page cache
//! let page_cache = CacheRef::new(
//! NonZeroU16::new(1024).unwrap(),
//! NonZeroUsize::new(10).unwrap(),
//! );
//!
//! // Create a queue and its reader
//! let (mut queue, mut reader) = Queue::<_, Vec<u8>>::init(context, Config {
//! partition: "my-queue".into(),
//! items_per_section: NonZeroU64::new(1000).unwrap(),
//! compression: None,
//! codec_config: ((0..).into(), ()), // RangeCfg for Vec length, () for u8
//! page_cache,
//! write_buffer: NonZeroUsize::new(4096).unwrap(),
//! replay_buffer: NonZeroUsize::new(4096).unwrap(),
//! }).await.unwrap();
//!
//! // Enqueue items
//! (queue, _) = queue.enqueue(&b"task1".to_vec()).await.unwrap();
//! (queue, _) = queue.enqueue(&b"task2".to_vec()).await.unwrap();
//!
//! // Receive and process items (can be acknowledged out of order)
//! while let Some((position, item)) = reader.try_recv().await.unwrap() {
//! // Process the item...
//! println!("Processing item at position {}", position);
//!
//! // Acknowledge after successful processing
//! reader.ack(position).unwrap();
//! }
//!
//! // Persist the queue and prune acknowledged items
//! queue.sync().await.unwrap();
//! });
//! ```
pub use ;
use Error;
/// Errors that can occur when interacting with [Queue].