Skip to main content

MessageQueueAdapter

tycho_collator::queue_adapter

Trait MessageQueueAdapter 

Source 
pub trait MessageQueueAdapter<V>: Send + Sync
where
    V: InternalMessageValue + Send + Sync,
{
    // Required methods
    fn create_iterator(
        &self,
        for_shard_id: ShardIdent,
        partition: QueuePartitionIdx,
        ranges: Vec<QueueShardBoundedRange>,
    ) -> Result<Box<dyn QueueIterator<V>>>;
    fn get_statistics(
        &self,
        partitions: &FastHashSet<QueuePartitionIdx>,
        ranges: &[QueueShardBoundedRange],
    ) -> Result<QueueStatistics>;
    fn prepare_diff(
        &self,
        diff: QueueDiffWithMessages<V>,
        block_id_short: BlockIdShort,
        diff_hash: &HashBytes,
        statistics: DiffStatistics,
        check_sequence: Option<DiffZone>,
    ) -> Result<Option<PendingQueueDiff>>;
    fn apply_diff(
        &self,
        diff: QueueDiffWithMessages<V>,
        block_id_short: BlockIdShort,
        diff_hash: &HashBytes,
        statistics: DiffStatistics,
        check_sequence: Option<DiffZone>,
    ) -> Result<()>;
    fn commit_diff(
        &self,
        mc_top_blocks: Vec<TopBlockIdUpdated>,
        partitions: &FastHashSet<QueuePartitionIdx>,
    ) -> Result<()>;
    fn clear_uncommitted_state(&self, top_shards: &[ShardIdent]) -> Result<()>;
    fn get_diff_info(
        &self,
        shard_ident: &ShardIdent,
        seqno: u32,
        zone: DiffZone,
    ) -> Result<Option<DiffInfo>>;
    fn is_diff_exists(&self, block_id_short: &BlockIdShort) -> Result<bool>;
    fn get_last_committed_mc_block_id(&self) -> Result<Option<BlockId>>;
    fn get_diffs_tail_len(
        &self,
        shard_ident: &ShardIdent,
        from: &QueueKey,
    ) -> u32;
    fn load_separated_diff_statistics(
        &self,
        partitions: &FastHashSet<QueuePartitionIdx>,
        range: &QueueShardBoundedRange,
    ) -> Result<SeparatedStatisticsByPartitions>;
    fn get_router_and_statistics(
        &self,
        block_id_short: &BlockIdShort,
        diff_info: DiffInfo,
        partition: QueuePartitionIdx,
    ) -> Result<(PartitionRouter, DiffStatistics)>;
}

Required Methods§

Source

fn create_iterator( &self, for_shard_id: ShardIdent, partition: QueuePartitionIdx, ranges: Vec<QueueShardBoundedRange>, ) -> Result<Box<dyn QueueIterator<V>>>

Create iterator for specified shard and return it

Source

fn get_statistics( &self, partitions: &FastHashSet<QueuePartitionIdx>, ranges: &[QueueShardBoundedRange], ) -> Result<QueueStatistics>

Returns statistics for the specified ranges by partition and source shards (equal to iterator ranges)

Source

fn prepare_diff( &self, diff: QueueDiffWithMessages<V>, block_id_short: BlockIdShort, diff_hash: &HashBytes, statistics: DiffStatistics, check_sequence: Option<DiffZone>, ) -> Result<Option<PendingQueueDiff>>

Prepare diff for applying to queue. Returns transaction that should be committed later. Returns None if diff is already applied (duplicate).

Source

fn apply_diff( &self, diff: QueueDiffWithMessages<V>, block_id_short: BlockIdShort, diff_hash: &HashBytes, statistics: DiffStatistics, check_sequence: Option<DiffZone>, ) -> Result<()>

Apply diff by storing it to the queue uncommitted zone (waiting for the operation to complete)

Source

fn commit_diff( &self, mc_top_blocks: Vec<TopBlockIdUpdated>, partitions: &FastHashSet<QueuePartitionIdx>, ) -> Result<()>

Commit previously applied diffs, updating commit pointers (waiting for the operation to complete)

Source

fn clear_uncommitted_state(&self, top_shards: &[ShardIdent]) -> Result<()>

Source

fn get_diff_info( &self, shard_ident: &ShardIdent, seqno: u32, zone: DiffZone, ) -> Result<Option<DiffInfo>>

Get diff for the given block from committed and/or uncommitted zone

Source

fn is_diff_exists(&self, block_id_short: &BlockIdShort) -> Result<bool>

Check if diff exists in state

Source

fn get_last_committed_mc_block_id(&self) -> Result<Option<BlockId>>

Get mc block id on which the queue was committed. Returns None if queue was not committed

Source

fn get_diffs_tail_len(&self, shard_ident: &ShardIdent, from: &QueueKey) -> u32

Get diffs tail len. from - start key for the tail. Diff with max_message == from will be excluded from the tail

Source

fn load_separated_diff_statistics( &self, partitions: &FastHashSet<QueuePartitionIdx>, range: &QueueShardBoundedRange, ) -> Result<SeparatedStatisticsByPartitions>

Load separated diff statistics for the specified partitions and range. range.from = diff with max_message == range.from will be excluded in statistics; range.to = diff with max_message == range.to will be included in statistics

Source

fn get_router_and_statistics( &self, block_id_short: &BlockIdShort, diff_info: DiffInfo, partition: QueuePartitionIdx, ) -> Result<(PartitionRouter, DiffStatistics)>

Get partition router and statistics for the specified block

Implementors§