Struct SessionAffinityTracker 

pub struct SessionAffinityTracker { /* private fields */ }

Tracks session-to-consumer affinity mappings for ordered processing.

The affinity tracker ensures that all messages for a given session are routed to the same consumer, maintaining message ordering and processing consistency within sessions.

Thread Safety

This type uses Arc<RwLock<>> internally and can be safely shared across threads and tasks.

Examples

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

let tracker = SessionAffinityTracker::new(Duration::from_secs(600));
let session_id = SessionId::new("session-123".to_string()).unwrap();

// Assign session to consumer
let affinity = tracker.assign_session(session_id.clone(), "worker-1".to_string()).await.unwrap();
assert_eq!(affinity.consumer_id(), "worker-1");

// Query affinity
let consumer = tracker.get_consumer(&session_id).await;
assert_eq!(consumer, Some("worker-1".to_string()));

Implementations

impl SessionAffinityTracker

pub fn new(default_affinity_duration: Duration) -> Self

Create a new session affinity tracker.

Arguments

• default_affinity_duration - Default duration for affinity mappings

Returns

A new SessionAffinityTracker instance

pub async fn assign_session( &self, session_id: SessionId, consumer_id: String, ) -> Result<SessionAffinity, QueueError>

Assign a session to a consumer.

If the session is already assigned and not expired, returns an error. If the session affinity has expired, reassigns to the new consumer.

Arguments

• session_id - The session to assign
• consumer_id - The consumer to assign the session to

Returns

The created affinity mapping on success, or an error if the session is already assigned to a different consumer.

Errors

Returns QueueError::SessionLocked if the session is already assigned to a different consumer and the affinity has not expired.

pub async fn get_consumer(&self, session_id: &SessionId) -> Option<String>

Get the consumer assigned to a session.

Arguments

• session_id - The session to query

Returns

The consumer ID if the session has an active affinity, None otherwise

pub async fn get_affinity( &self, session_id: &SessionId, ) -> Option<SessionAffinity>

Get the full affinity information for a session.

Arguments

• session_id - The session to query

Returns

The affinity information if the session has an active affinity

pub async fn has_affinity(&self, session_id: &SessionId) -> bool

Check if a session has an active affinity.

Arguments

• session_id - The session to check

Returns

true if the session has an active affinity

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

Update the last activity time for a session.

This should be called when a message is processed for the session.

Arguments

• session_id - The session to update

Returns

Ok(()) if successful, error if session not found or expired

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

Release a session affinity.

Arguments

• session_id - The session to release
• consumer_id - The consumer releasing the session (for validation)

Returns

Ok(()) if successful

Errors

Returns error if the consumer doesn’t own the session

pub async fn extend_affinity( &self, session_id: &SessionId, consumer_id: &str, additional_duration: Duration, ) -> Result<SessionAffinity, QueueError>

Extend the affinity duration for a session.

Arguments

• session_id - The session to extend
• consumer_id - The consumer requesting the extension (for validation)
• additional_duration - Additional time to add

Returns

The updated affinity on success

Errors

Returns error if consumer doesn’t own the session or session not found

pub async fn get_consumer_sessions(&self, consumer_id: &str) -> Vec<SessionId>

Get all sessions assigned to a consumer.

Arguments

• consumer_id - The consumer to query

Returns

List of session IDs assigned to the consumer

pub async fn cleanup_expired(&self) -> usize

Clean up expired affinities.

Returns

Number of affinities removed

pub async fn affinity_count(&self) -> usize

Get the total number of affinity mappings (including expired).

pub async fn active_affinity_count(&self) -> usize

Get the number of active (non-expired) affinities.

Trait Implementations

impl Clone for SessionAffinityTracker

fn clone(&self) -> SessionAffinityTracker

Returns a duplicate of the value.

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

Performs copy-assignment from source.