ltk/event_loop/
sources.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
// SPDX-License-Identifier: LGPL-2.1-only
// Copyright (C) 2026 Liberux Labs, S. L. <info@liberux.net>

//! The timers and channels the runtime puts on the backend's run loop, held by `AppData::sources`: calloop's natively; in a browser and on Windows, a queue the backend runs from its own loop.

use std::time::Duration;

#[ cfg( not( queued_sources ) ) ]
use calloop::channel::Event;
#[ cfg( not( queued_sources ) ) ]
use calloop::timer::{ TimeoutAction, Timer };
#[ cfg( not( queued_sources ) ) ]
use calloop::{ LoopHandle, RegistrationToken };

use crate::app::ChannelSender;

#[ cfg( not( queued_sources ) ) ]
pub( crate ) struct Sources<D: 'static>
{
	handle: LoopHandle<'static, D>,
}

/// Names an armed timer to [`Sources::remove`].
#[ cfg( not( queued_sources ) ) ]
pub( crate ) struct TimerToken( RegistrationToken );

#[ cfg( not( queued_sources ) ) ]
impl<D: 'static> Sources<D>
{
	pub( crate ) fn new( handle: LoopHandle<'static, D> ) -> Self
	{
		Self { handle }
	}

	/// Run `tick` after `delay`, and again after every duration it returns, until it returns `None`.
	pub( crate ) fn insert_timer( &self, delay: Duration, mut tick: impl FnMut( &mut D ) -> Option<Duration> + 'static ) -> Result<TimerToken, String>
	{
		self.handle.insert_source( Timer::from_duration( delay ), move |_, _, data|
		{
			match tick( data )
			{
				Some( next ) => TimeoutAction::ToDuration( next ),
				None         => TimeoutAction::Drop,
			}
		} )
			.map( TimerToken )
			.map_err( |e| format!( "{e:?}" ) )
	}

	pub( crate ) fn remove( &self, token: TimerToken )
	{
		self.handle.remove( token.0 );
	}

	/// A sender whose messages reach `on_msg` on the run loop.
	pub( crate ) fn channel<T: 'static>( &self, mut on_msg: impl FnMut( T, &mut D ) + 'static ) -> Result<ChannelSender<T>, String>
	{
		let ( sender, messages ) = calloop::channel::channel::<T>();
		self.handle.insert_source( messages, move |event, _, data|
		{
			if let Event::Msg( msg ) = event { on_msg( msg, data ); }
		} )
			.map_err( |e| format!( "{e:?}" ) )?;
		Ok( ChannelSender::from_calloop( sender ) )
	}
}

#[ cfg( queued_sources ) ]
pub( crate ) type Tick<D> = Box<dyn FnMut( &mut D ) -> Option<Duration>>;

#[ cfg( queued_sources ) ]
pub( crate ) type Drain<D> = Box<dyn FnMut( &mut D )>;

#[ cfg( queued_sources ) ]
struct Armed<D>
{
	id:   u64,
	due:  crate::time::Instant,
	tick: Tick<D>,
}

/// The backend takes the due timers out with [`Self::take_due`], runs them on its `AppData` and hands them back to [`Self::rearm`]; likewise the channels with [`Self::take_channels`] and [`Self::restore_channels`].
#[ cfg( queued_sources ) ]
pub( crate ) struct Sources<D: 'static>
{
	wake:      fn(),
	next_id:   std::cell::Cell<u64>,
	armed:     std::cell::RefCell<Vec<Armed<D>>>,
	/// Taken out by `take_due` and not handed back yet.
	running:   std::cell::RefCell<Vec<u64>>,
	/// Removed while running, so `rearm` drops them.
	cancelled: std::cell::RefCell<Vec<u64>>,
	channels:  std::cell::RefCell<Vec<Drain<D>>>,
}

#[ cfg( queued_sources ) ]
pub( crate ) struct TimerToken( u64 );

#[ cfg( queued_sources ) ]
impl<D: 'static> Sources<D>
{
	/// `wake` asks the backend for an iteration once a message is sent.
	pub( crate ) fn new( wake: fn() ) -> Self
	{
		Self
		{
			wake,
			next_id:   std::cell::Cell::new( 0 ),
			armed:     std::cell::RefCell::new( Vec::new() ),
			running:   std::cell::RefCell::new( Vec::new() ),
			cancelled: std::cell::RefCell::new( Vec::new() ),
			channels:  std::cell::RefCell::new( Vec::new() ),
		}
	}

	/// Run `tick` after `delay`, and again after every duration it returns, until it returns `None`.
	pub( crate ) fn insert_timer( &self, delay: Duration, tick: impl FnMut( &mut D ) -> Option<Duration> + 'static ) -> Result<TimerToken, String>
	{
		let id = self.next_id.get();
		self.next_id.set( id + 1 );
		self.armed.borrow_mut().push( Armed { id, due: crate::time::Instant::now() + delay, tick: Box::new( tick ) } );
		Ok( TimerToken( id ) )
	}

	pub( crate ) fn remove( &self, token: TimerToken )
	{
		let mut armed = self.armed.borrow_mut();
		if let Some( i ) = armed.iter().position( |t| t.id == token.0 )
		{
			armed.remove( i );
		} else if self.running.borrow().contains( &token.0 ) {
			self.cancelled.borrow_mut().push( token.0 );
		}
	}

	/// A sender whose messages reach `on_msg` when the backend drains the channels.
	pub( crate ) fn channel<T: 'static>( &self, mut on_msg: impl FnMut( T, &mut D ) + 'static ) -> Result<ChannelSender<T>, String>
	{
		let ( sender, messages ) = crate::channel::channel::<T>( self.wake );
		self.channels.borrow_mut().push( Box::new( move |data|
		{
			while let Ok( msg ) = messages.try_recv() { on_msg( msg, data ); }
		} ) );
		Ok( sender )
	}

	/// The timers due by `now`, with their ids.
	pub( crate ) fn take_due( &self, now: crate::time::Instant ) -> Vec<( u64, Tick<D> )>
	{
		let ( due, later ): ( Vec<_>, Vec<_> ) = std::mem::take( &mut *self.armed.borrow_mut() ).into_iter().partition( |t| t.due <= now );
		self.armed.borrow_mut().extend( later );
		self.running.borrow_mut().extend( due.iter().map( |t| t.id ) );
		due.into_iter().map( |t| ( t.id, t.tick ) ).collect()
	}

	/// Arm a timer [`Self::take_due`] gave out again after `next`, unless it returned `None` or was removed meanwhile.
	pub( crate ) fn rearm( &self, id: u64, tick: Tick<D>, next: Option<Duration> )
	{
		self.running.borrow_mut().retain( |r| *r != id );
		let mut cancelled = self.cancelled.borrow_mut();
		if let Some( i ) = cancelled.iter().position( |c| *c == id )
		{
			cancelled.remove( i );
			return;
		}
		if let Some( next ) = next
		{
			self.armed.borrow_mut().push( Armed { id, due: crate::time::Instant::now() + next, tick } );
		}
	}

	/// When the next timer is due.
	pub( crate ) fn next_due( &self ) -> Option<crate::time::Instant>
	{
		self.armed.borrow().iter().map( |t| t.due ).min()
	}

	pub( crate ) fn take_channels( &self ) -> Vec<Drain<D>>
	{
		std::mem::take( &mut *self.channels.borrow_mut() )
	}

	/// Hand back what [`Self::take_channels`] took, keeping any channel created meanwhile.
	pub( crate ) fn restore_channels( &self, mut drains: Vec<Drain<D>> )
	{
		let mut channels = self.channels.borrow_mut();
		drains.append( &mut channels );
		*channels = drains;
	}
}

#[ cfg( queued_sources ) ]
impl<A: crate::app::App, P: crate::platform::Platform> super::AppData<A, P>
{
	/// Run the timers due by now and the messages waiting on the channels; a backend whose loop holds the queue calls it on every iteration.
	pub( crate ) fn dispatch_sources( &mut self )
	{
		for ( id, mut tick ) in self.sources.take_due( crate::time::Instant::now() )
		{
			let next = tick( self );
			self.sources.rearm( id, tick, next );
		}
		let mut drains = self.sources.take_channels();
		for drain in &mut drains
		{
			drain( self );
		}
		self.sources.restore_channels( drains );
	}
}