alamin/websurfx
alamin/websurfx
0
0
Fork 0
mirror of https://github.com/neon-mmd/websurfx.git synced 2024-11-23 06:28:23 -05:00
Code Issues Releases Activity

⚡ perf: initialize new async connections parallely using tokio spawn tasks (#486)

Browse Source
This commit is contained in:
neon_arch 2024-02-10 21:38:40 +03:00
parent a56a2f1345
commit d78fd75979
1 changed files with 14 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/cache/redis_cacher.rs vendored
View File

@ -1,11 +1,10 @@
//! This module provides the functionality to cache the aggregated results fetched and aggregated //! This module provides the functionality to cache the aggregated results fetched and aggregated
//! from the upstream search engines in a json format. //! from the upstream search engines in a json format.
use error_stack::Report;
use futures::future::try_join_all;
use redis::{aio::ConnectionManager, AsyncCommands, Client, RedisError};
use super::error::CacheError; use super::error::CacheError;
use error_stack::Report;
use futures::stream::FuturesUnordered;
use redis::{aio::ConnectionManager, AsyncCommands, Client, RedisError};
/// A named struct which stores the redis Connection url address to which the client will /// A named struct which stores the redis Connection url address to which the client will
/// connect to. /// connect to.
@ -41,14 +40,22 @@ impl RedisCache {
cache_ttl: u16, cache_ttl: u16,
) -> Result<Self, Box<dyn std::error::Error>> { ) -> Result<Self, Box<dyn std::error::Error>> {
let client = Client::open(redis_connection_url)?; let client = Client::open(redis_connection_url)?;
let mut tasks: Vec<_> = Vec::new(); let tasks: FuturesUnordered<_> = FuturesUnordered::new();
for _ in 0..pool_size { for _ in 0..pool_size {
tasks.push(client.get_connection_manager()); let client_partially_cloned = client.clone();
tasks.push(tokio::spawn(async move {
client_partially_cloned.get_tokio_connection_manager().await
}));
}
let mut outputs = Vec::new();
for task in tasks {
outputs.push(task.await??);
} }
let redis_cache = RedisCache { let redis_cache = RedisCache {
connection_pool: try_join_all(tasks).await?, connection_pool: outputs,
pool_size, pool_size,
current_connection: Default::default(), current_connection: Default::default(),
cache_ttl, cache_ttl,