magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

use loop condition to account for loole channel close

Browse source 
Signed-off-by: Jason Volk <jason@zemos.net>
This commit is contained in:
Jason Volk 2024-10-06 22:08:55 +00:00 • committed by strawberry
parent 814b9e28b6
commit 56dd0f5139
2 changed files with 5 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

7
src/service/presence/mod.rs
View file

@ -55,14 +55,13 @@ impl crate::Service for Service {
async fn worker(self: Arc<Self>) -> Result<()> { async fn worker(self: Arc<Self>) -> Result<()> {
let mut presence_timers = FuturesUnordered::new(); let mut presence_timers = FuturesUnordered::new();
let receiver = self.timer_receiver.lock().await; let receiver = self.timer_receiver.lock().await;
loop { while !receiver.is_closed() {
debug_assert!(!receiver.is_closed(), "channel error");
tokio::select! { tokio::select! {
Some(user_id) = presence_timers.next() => { Some(user_id) = presence_timers.next() => {
self.process_presence_timer(&user_id).await.log_err().ok(); self.process_presence_timer(&user_id).await.log_err().ok();
}, },
event = receiver.recv_async() => match event { event = receiver.recv_async() => match event {
Err(_e) => return Ok(()), Err(_) => break,
Ok((user_id, timeout)) => { Ok((user_id, timeout)) => {
debug!("Adding timer {}: {user_id} timeout:{timeout:?}", presence_timers.len()); debug!("Adding timer {}: {user_id} timeout:{timeout:?}", presence_timers.len());
presence_timers.push(presence_timer(user_id, timeout)); presence_timers.push(presence_timer(user_id, timeout));
@ -70,6 +69,8 @@ impl crate::Service for Service {
}, },
} }
} }
Ok(())
} }
fn interrupt(&self) { fn interrupt(&self) {

3
src/service/sending/sender.rs
View file

@ -57,8 +57,7 @@ impl Service {
let receiver = self.receiver.lock().await; let receiver = self.receiver.lock().await;
self.initial_requests(&mut futures, &mut statuses).await; self.initial_requests(&mut futures, &mut statuses).await;
loop { while !receiver.is_closed() {
debug_assert!(!receiver.is_closed(), "channel error");
tokio::select! { tokio::select! {
request = receiver.recv_async() => match request { request = receiver.recv_async() => match request {
Ok(request) => self.handle_request(request, &mut futures, &mut statuses).await, Ok(request) => self.handle_request(request, &mut futures, &mut statuses).await,