Struct RedisClientImpl 

pub struct RedisClientImpl {
    pub client: Client,
    pub conn: OnceCell<MultiplexedConnection>,
    /* private fields */
}

Wrapper for the Redis connection

Fields

client: Client

Database connector string

conn: OnceCell<MultiplexedConnection>

Implementations

impl RedisClientImpl

pub fn new(metrics: Arc<StatsdClient>, settings: &DbSettings) -> DbResult<Self>

Trait Implementations

impl Clone for RedisClientImpl

fn clone(&self) -> RedisClientImpl

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl DbClient for RedisClientImpl

fn add_user<'life0, 'life1, 'async_trait>( &'life0 self, user: &'life1 User, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

add user to the database

fn update_user<'life0, 'life1, 'async_trait>( &'life0 self, user: &'life1 mut User, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

To update the TTL of the Redis entry we just have to SET again, with the new expiry

NOTE: This function is called by mobile during the daily [autoendpoint::routes::update_token_route] handling, and by desktop [autoconnect-ws-sm::get_or_create_user]which is called during theHELLO` handler. This should be enough to ensure that the ROUTER records are properly refreshed for “lively” clients.

NOTE: There is some, very small, potential risk that a desktop client that can somehow remain connected the duration of MAX_ROUTER_TTL, may be dropped as not being “lively”.

fn add_channels<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, channels: HashSet<Uuid>, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Add channels in bulk (used mostly during migration)

fn remove_channel<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, channel_id: &'life2 Uuid, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Delete the channel. Does not delete its associated pending messages.

fn remove_node_id<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, _node_id: &'life2 str, _connected_at: u64, _version: &'life3 Option<Uuid>, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Remove the node_id

fn save_message<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, message: Notification, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Write the notification to storage.

If the message contains a topic, we remove the old message

fn save_messages<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, messages: Vec<Notification>, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Save a batch of messages to the database.

Currently just iterating through the list and saving one at a time. There’s a bulk way to save messages, but there are other considerations (e.g. mutation limits)

fn increment_storage<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, timestamp: u64, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete expired messages

fn remove_message<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, chidmessageid: &'life2 str, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Delete the notification from storage.

fn fetch_topic_messages<'life0, 'life1, 'async_trait>( &'life0 self, _uaid: &'life1 Uuid, _limit: usize, ) -> Pin<Box<dyn Future<Output = DbResult<FetchMessageResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Topic messages are handled as other messages with redis, we return nothing.

fn fetch_timestamp_messages<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, timestamp: Option<u64>, limit: usize, ) -> Pin<Box<dyn Future<Output = DbResult<FetchMessageResponse>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Return [limit] messages pending for a [uaid] that have a record timestamp after [timestamp] (secs).

If [limit] = 0, we fetch all messages after [timestamp].

This can return expired messages, following bigtables behavior

fn router_table_exists<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns true, because there’s no table in Redis

fn message_table_exists<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns true, because there’s no table in Redis

fn get_user<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, ) -> Pin<Box<dyn Future<Output = DbResult<Option<User>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read a user from the database

fn remove_user<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a user from the router table

fn add_channel<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, channel_id: &'life2 Uuid, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Add a channel to a user

fn get_channels<'life0, 'life1, 'async_trait>( &'life0 self, uaid: &'life1 Uuid, ) -> Pin<Box<dyn Future<Output = DbResult<HashSet<Uuid>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get the set of channel IDs for a user

fn log_report<'life0, 'life1, 'async_trait>( &'life0 self, reliability_id: &'life1 str, state: ReliabilityState, ) -> Pin<Box<dyn Future<Output = DbResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Record the Reliability Report to long term storage.

fn health_check<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = DbResult<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Perform the health check on this data store

fn box_clone(&self) -> Box<dyn DbClient>

fn name(&self) -> String

Provide the module name. This was added for simple dual mode testing (legacy), but may be useful in other situations.

fn pool_status(&self) -> Option<Status>

Return the current deadpool Status (if using deadpool)