Struct Queue 

pub struct Queue<E: TitoEngine> {
    pub engine: E,
    pub config: QueueConfig,
}

Fields

engine: E

config: QueueConfig

Implementations

impl<E: TitoEngine> Queue<E>

pub fn new(engine: E, config: QueueConfig) -> Self

Examples found in repository

examples/queue_fifo.rs (lines 36-39)

async fn main() -> Result<(), TitoError> {
    println!("Testing FIFO Queue\n");

    let tito_db = TiKV::connect(vec!["127.0.0.1:2379"]).await?;

    let queue = Arc::new(TitoQueue::new(
        tito_db.clone(),
        QueueConfig::with_partitions(1),
    ));

    println!("Publishing 5 events...\n");

    for i in 1..=5 {
        let user_id = DBUuid::new_v4().to_string();
        let event = QueueEvent::new(
            format!("user:{}", user_id),
            UserEvent {
                user_id,
                name: format!("User {}", i),
                email: format!("user{}@example.com", i),
                action: "created".to_string(),
            },
        );

        queue.publish(event).await?;
        println!("Published event for User {}", i);
    }

    println!("\nStarting worker...\n");

    let events_processed = Arc::new(std::sync::atomic::AtomicU32::new(0));
    let events_processed_clone = events_processed.clone();

    let (shutdown_tx, shutdown_rx) = broadcast::channel(1);

    let handler = move |event: QueueEvent<UserEvent>| {
        let counter = events_processed_clone.clone();
        Box::pin(async move {
            let count = counter.fetch_add(1, Ordering::SeqCst) + 1;
            println!(
                "[{}] {} - {} ({})",
                count,
                event.payload.action,
                event.payload.name,
                event.payload.email,
            );
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            Ok::<_, TitoError>(())
        }) as BoxFuture<'static, Result<(), TitoError>>
    };

    let worker_handle = run_worker(
        queue.clone(),
        WorkerConfig {
            event_type: String::from("user"),
            consumer: String::from("example-consumer"),
            partition_range: 0..1,
        },
        handler,
        shutdown_rx,
    )
    .await;

    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;

    println!("\nShutting down...");
    let _ = shutdown_tx.send(());
    let _ = worker_handle.await;

    let total = events_processed.load(Ordering::SeqCst);
    println!("\nProcessed {} events", total);

    Ok(())
}

pub async fn publish_in_tx<T: EventType + Serialize>( &self, event: QueueEvent<T>, tx: &E::Transaction, ) -> Result<(), TitoError>

pub async fn publish<T: EventType + Serialize>( &self, event: QueueEvent<T>, ) -> Result<(), TitoError>

Examples found in repository

examples/queue_fifo.rs (line 55)

async fn main() -> Result<(), TitoError> {
    println!("Testing FIFO Queue\n");

    let tito_db = TiKV::connect(vec!["127.0.0.1:2379"]).await?;

    let queue = Arc::new(TitoQueue::new(
        tito_db.clone(),
        QueueConfig::with_partitions(1),
    ));

    println!("Publishing 5 events...\n");

    for i in 1..=5 {
        let user_id = DBUuid::new_v4().to_string();
        let event = QueueEvent::new(
            format!("user:{}", user_id),
            UserEvent {
                user_id,
                name: format!("User {}", i),
                email: format!("user{}@example.com", i),
                action: "created".to_string(),
            },
        );

        queue.publish(event).await?;
        println!("Published event for User {}", i);
    }

    println!("\nStarting worker...\n");

    let events_processed = Arc::new(std::sync::atomic::AtomicU32::new(0));
    let events_processed_clone = events_processed.clone();

    let (shutdown_tx, shutdown_rx) = broadcast::channel(1);

    let handler = move |event: QueueEvent<UserEvent>| {
        let counter = events_processed_clone.clone();
        Box::pin(async move {
            let count = counter.fetch_add(1, Ordering::SeqCst) + 1;
            println!(
                "[{}] {} - {} ({})",
                count,
                event.payload.action,
                event.payload.name,
                event.payload.email,
            );
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            Ok::<_, TitoError>(())
        }) as BoxFuture<'static, Result<(), TitoError>>
    };

    let worker_handle = run_worker(
        queue.clone(),
        WorkerConfig {
            event_type: String::from("user"),
            consumer: String::from("example-consumer"),
            partition_range: 0..1,
        },
        handler,
        shutdown_rx,
    )
    .await;

    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;

    println!("\nShutting down...");
    let _ = shutdown_tx.send(());
    let _ = worker_handle.await;

    let total = events_processed.load(Ordering::SeqCst);
    println!("\nProcessed {} events", total);

    Ok(())
}

pub async fn pull<T: EventType + DeserializeOwned>( &self, event_type: &str, partition: u32, limit: u32, ) -> Result<Vec<(String, QueueEvent<T>)>, TitoError>

pub async fn ack(&self, key: &str) -> Result<(), TitoError>

pub async fn reschedule<T: EventType + Serialize>( &self, event: QueueEvent<T>, storage_key: &str, new_scheduled_at: i64, ) -> Result<(), TitoError>

pub async fn clear(&self) -> Result<(), TitoError>

pub async fn move_to_dlq<T: EventType + Serialize>( &self, event: QueueEvent<T>, storage_key: &str, ) -> Result<(), TitoError>

Trait Implementations

impl<E: Clone + TitoEngine> Clone for Queue<E>

fn clone(&self) -> Queue<E>

Returns a duplicate of the value.

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

Performs copy-assignment from source.