YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-04-02 02:35:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

aquatic_udp: glommio: start work on periodic cleaning

Browse source
This commit is contained in:
Joakim Frostegård 2021-10-19 11:54:00 +02:00
parent b5aa07c21f
commit cad3618fad
1 changed files with 22 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

33
aquatic_udp/src/lib/glommio/handlers.rs
View file

@ -1,10 +1,12 @@
use std::cell::RefCell; use std::cell::RefCell;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::net::{IpAddr, SocketAddr};
use std::rc::Rc; use std::rc::Rc;
use std::time::Duration;
use futures_lite::{Stream, StreamExt}; use futures_lite::{Stream, StreamExt};
use glommio::channels::channel_mesh::{MeshBuilder, Partial, Role, Senders}; use glommio::channels::channel_mesh::{MeshBuilder, Partial, Role, Senders};
use glommio::prelude::*; use glommio::{enclose, prelude::*};
use glommio::timer::TimerActionRepeat;
use rand::prelude::SmallRng; use rand::prelude::SmallRng;
use rand::SeedableRng; use rand::SeedableRng;
@ -22,15 +24,25 @@ pub async fn run_request_worker(
let response_senders = Rc::new(response_senders); let response_senders = Rc::new(response_senders);
// Need to be cleaned periodically: use timer? let torrents= Rc::new(RefCell::new(TorrentMaps::default()));
let torrents_ipv4 = Rc::new(RefCell::new(TorrentMap::<Ipv4Addr>::default()));
let torrents_ipv6 = Rc::new(RefCell::new(TorrentMap::<Ipv6Addr>::default())); async fn clean(
config: Config,
torrents: Rc<RefCell<TorrentMaps>>,
) -> Option<Duration> {
torrents.borrow_mut(); // .clean(config, access_list);
Some(Duration::from_secs(config.cleaning.interval))
}
TimerActionRepeat::repeat(enclose!((config, torrents) move || {
clean(config.clone(), torrents.clone())
}));
for (_, receiver) in request_receivers.streams() { for (_, receiver) in request_receivers.streams() {
handle_request_stream( handle_request_stream(
&config, &config,
torrents_ipv4.clone(), torrents.clone(),
torrents_ipv6.clone(),
response_senders.clone(), response_senders.clone(),
receiver, receiver,
) )
@ -40,8 +52,7 @@ pub async fn run_request_worker(
async fn handle_request_stream<S>( async fn handle_request_stream<S>(
config: &Config, config: &Config,
torrents_ipv4: Rc<RefCell<TorrentMap<Ipv4Addr>>>, torrents: Rc<RefCell<TorrentMaps>>,
torrents_ipv6: Rc<RefCell<TorrentMap<Ipv6Addr>>>,
response_senders: Rc<Senders<(AnnounceResponse, SocketAddr)>>, response_senders: Rc<Senders<(AnnounceResponse, SocketAddr)>>,
mut stream: S, mut stream: S,
) where ) where
@ -57,7 +68,7 @@ async fn handle_request_stream<S>(
IpAddr::V4(ip) => handle_announce_request( IpAddr::V4(ip) => handle_announce_request(
&config, &config,
&mut rng, &mut rng,
&mut torrents_ipv4.borrow_mut(), &mut torrents.borrow_mut().ipv4,
request, request,
ip, ip,
peer_valid_until, peer_valid_until,
@ -65,7 +76,7 @@ async fn handle_request_stream<S>(
IpAddr::V6(ip) => handle_announce_request( IpAddr::V6(ip) => handle_announce_request(
&config, &config,
&mut rng, &mut rng,
&mut torrents_ipv6.borrow_mut(), &mut torrents.borrow_mut().ipv6,
request, request,
ip, ip,
peer_valid_until, peer_valid_until,