Struct SessionLockManager

Source

pub struct SessionLockManager { /* private fields */ }

Expand description

Manages session locks for concurrent message processing.

The lock manager coordinates exclusive access to sessions, ensuring that only one consumer processes messages from a session at a time. It handles lock acquisition, renewal, release, and automatic expiration cleanup.

§Thread Safety

This type is thread-safe and can be shared across async tasks using Arc.

§Example

use queue_runtime::sessions::SessionLockManager;
use queue_runtime::message::SessionId;
use std::time::Duration;

let manager = SessionLockManager::new(Duration::from_secs(30));
let session_id = SessionId::new("order-456".to_string()).unwrap();

// Acquire lock
let lock = manager.acquire_lock(session_id.clone(), "consumer-1".to_string()).await?;
assert_eq!(lock.owner(), "consumer-1");

// Try to acquire same session with different consumer - should fail
let result = manager.try_acquire_lock(session_id.clone(), "consumer-2".to_string()).await;
assert!(result.is_err());

// Release lock
manager.release_lock(&session_id, "consumer-1").await?;

Implementations§

Source §

impl SessionLockManager

Source

pub fn new(default_lock_duration: Duration) -> Self

Create a new session lock manager.

§Arguments

default_lock_duration - Default duration for session locks

§Example

use queue_runtime::sessions::SessionLockManager;
use std::time::Duration;

let manager = SessionLockManager::new(Duration::from_secs(60));

Source

pub async fn try_acquire_lock( &self, session_id: SessionId, owner: String, ) -> Result<SessionLock, QueueError>

Try to acquire a lock on a session (non-blocking).

Returns immediately with an error if the session is already locked by another consumer.

§Arguments

session_id - The session to lock
owner - Identifier of the consumer requesting the lock

§Returns

The acquired lock if successful, or an error if the session is locked

§Errors

Returns QueueError::SessionLocked if the session is already locked by another consumer and the lock has not expired.

Source

pub async fn acquire_lock( &self, session_id: SessionId, owner: String, ) -> Result<SessionLock, QueueError>

Acquire a lock on a session (blocking with timeout).

Waits for the lock to become available if it’s currently held by another consumer, up to the specified timeout.

§Arguments

session_id - The session to lock
owner - Identifier of the consumer requesting the lock

§Returns

The acquired lock if successful within the timeout period

§Errors

Returns QueueError::SessionLocked if unable to acquire the lock within the timeout period.

Source

pub async fn renew_lock( &self, session_id: &SessionId, owner: &str, extension: Option<Duration>, ) -> Result<SessionLock, QueueError>

Renew an existing session lock.

Extends the lock’s expiration time, allowing the consumer to continue processing messages from the session.

§Arguments

session_id - The session whose lock to renew
owner - Identifier of the consumer owning the lock
extension - How long to extend the lock by (if None, uses default duration)

§Returns

The renewed lock with updated expiration time

§Errors

Returns QueueError::SessionNotFound if no lock exists for the session. Returns QueueError::SessionLocked if the lock is owned by a different consumer.

Source

pub async fn release_lock( &self, session_id: &SessionId, owner: &str, ) -> Result<(), QueueError>

Release a session lock.

Removes the lock, allowing other consumers to acquire it.

§Arguments

session_id - The session whose lock to release
owner - Identifier of the consumer releasing the lock

§Returns

Ok(()) if the lock was successfully released

§Errors

Returns QueueError::SessionNotFound if no lock exists for the session. Returns QueueError::SessionLocked if the lock is owned by a different consumer.

Source