Struct TxnsSubscribeTask

struct TxnsSubscribeTask<T, C: TxnsCodec = TxnsCodecDefault> {
    txns_subscribe: Subscribe<C::Key, C::Val, T, i64>,
    buf: Vec<(TxnsEntry, T, i64)>,
    tx: UnboundedSender<TxnsReadCmd<T>>,
    only_data_id: Option<ShardId>,
}

Reads txn updates from a Subscribe and forwards them to a TxnsReadTask when receiving a progress update.

Fields

txns_subscribe: Subscribe<C::Key, C::Val, T, i64>

buf: Vec<(TxnsEntry, T, i64)>

Staged update that we will consume and forward to the TxnsReadTask when we receive a progress update.

tx: UnboundedSender<TxnsReadCmd<T>>

For sending updates to the main TxnsReadTask.

only_data_id: Option<ShardId>

If Some, this cache only tracks the indicated data shard as a performance optimization. When used, only some methods (in particular, the ones necessary for the txns_progress operator) are supported.

TODO: It’d be nice to make this a compile time thing. I have some ideas, but they’re decently invasive, so leave it for a followup.

Implementations

impl<T, C> TxnsSubscribeTask<T, C>

where T: Timestamp + Lattice + TotalOrder + StepForward + Codec64 + Sync, C: TxnsCodec,

pub async fn open( client: &PersistClient, txns_id: ShardId, only_data_id: Option<ShardId>, tx: UnboundedSender<TxnsReadCmd<T>>, ) -> (Self, TxnsCacheState<T>)

Creates a TxnsSubscribeTask reading from the given txn shard that forwards updates (entries and progress) to the given tx.

This returns both the created task and a TxnsCacheState that can be used to interact with the txn system and into which the updates should be funneled.

NOTE: We create both the TxnsSubscribeTask and the TxnsCacheState at the same time because the cache is initialized with a since_ts, which we get from the same ReadHandle that we use to initialize the Subscribe.

async fn run(&mut self)

Trait Implementations

impl<T: Debug, C: Debug + TxnsCodec> Debug for TxnsSubscribeTask<T, C>

where C::Key: Debug, C::Val: Debug,

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

Formats the value using the given formatter.