rickub
nandi/jolt-nativepublic⑂ Fork 0
⑂ cfd3e3677bed92e80cfe447469a249cfdc0b1401
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/jolt-native.git
git clone ssh://git@rickub.com/nandi/jolt-native.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

stream.rs · 468 lines · 17.6 KBRust Blame HistoryRaw
Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago1use crate::traits::StreamTrait;
2use crate::ChannelCount;
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::{Arc, Mutex};
5
6use crate::{
7 BackendSpecificError, Data, InputCallbackInfo, OutputCallbackInfo, PauseStreamError,
8 PlayStreamError, SampleRate, StreamError,
9};
10
11use super::JACK_SAMPLE_FORMAT;
12
13type ErrorCallbackPtr = Arc<Mutex<dyn FnMut(StreamError) + Send + 'static>>;
14
15pub struct Stream {
16 // TODO: It might be faster to send a message when playing/pausing than to check this every iteration
17 playing: Arc<AtomicBool>,
18 async_client: jack::AsyncClient<JackNotificationHandler, LocalProcessHandler>,
19 // Port names are stored in order to connect them to other ports in jack automatically
20 input_port_names: Vec<String>,
21 output_port_names: Vec<String>,
22}
23
24// Compile-time assertion that Stream is Send and Sync
25crate::assert_stream_send!(Stream);
26crate::assert_stream_sync!(Stream);
27
28impl Stream {
29 // TODO: Return error messages
30 pub fn new_input<D, E>(
31 client: jack::Client,
32 channels: ChannelCount,
33 data_callback: D,
34 mut error_callback: E,
35 ) -> Stream
36 where
37 D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
38 E: FnMut(StreamError) + Send + 'static,
39 {
40 let mut ports = vec![];
41 let mut port_names: Vec<String> = vec![];
42 // Create ports
43 for i in 0..channels {
44 let port_try = client.register_port(&format!("in_{}", i), jack::AudioIn::default());
45 match port_try {
46 Ok(port) => {
47 // Get the port name in order to later connect it automatically
48 if let Ok(port_name) = port.name() {
49 port_names.push(port_name);
50 }
51 // Store the port into a Vec to move to the ProcessHandler
52 ports.push(port);
53 }
54 Err(e) => {
55 // If port creation failed, send the error back via the error_callback
56 error_callback(
57 BackendSpecificError {
58 description: e.to_string(),
59 }
60 .into(),
61 );
62 }
63 }
64 }
65
66 let playing = Arc::new(AtomicBool::new(true));
67
68 let error_callback_ptr = Arc::new(Mutex::new(error_callback)) as ErrorCallbackPtr;
69
70 let input_process_handler = LocalProcessHandler::new(
71 vec![],
72 ports,
73 client.sample_rate(),
74 client.buffer_size() as usize,
75 Some(Box::new(data_callback)),
76 None,
77 playing.clone(),
78 Arc::clone(&error_callback_ptr),
79 );
80
81 let notification_handler = JackNotificationHandler::new(error_callback_ptr);
82
83 let async_client = client
84 .activate_async(notification_handler, input_process_handler)
85 .unwrap();
86
87 Stream {
88 playing,
89 async_client,
90 input_port_names: port_names,
91 output_port_names: vec![],
92 }
93 }
94
95 pub fn new_output<D, E>(
96 client: jack::Client,
97 channels: ChannelCount,
98 data_callback: D,
99 mut error_callback: E,
100 ) -> Stream
101 where
102 D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
103 E: FnMut(StreamError) + Send + 'static,
104 {
105 let mut ports = vec![];
106 let mut port_names: Vec<String> = vec![];
107 // Create ports
108 for i in 0..channels {
109 let port_try = client.register_port(&format!("out_{}", i), jack::AudioOut::default());
110 match port_try {
111 Ok(port) => {
112 // Get the port name in order to later connect it automatically
113 if let Ok(port_name) = port.name() {
114 port_names.push(port_name);
115 }
116 // Store the port into a Vec to move to the ProcessHandler
117 ports.push(port);
118 }
119 Err(e) => {
120 // If port creation failed, send the error back via the error_callback
121 error_callback(
122 BackendSpecificError {
123 description: e.to_string(),
124 }
125 .into(),
126 );
127 }
128 }
129 }
130
131 let playing = Arc::new(AtomicBool::new(true));
132
133 let error_callback_ptr = Arc::new(Mutex::new(error_callback)) as ErrorCallbackPtr;
134
135 let output_process_handler = LocalProcessHandler::new(
136 ports,
137 vec![],
138 client.sample_rate(),
139 client.buffer_size() as usize,
140 None,
141 Some(Box::new(data_callback)),
142 playing.clone(),
143 Arc::clone(&error_callback_ptr),
144 );
145
146 let notification_handler = JackNotificationHandler::new(error_callback_ptr);
147
148 let async_client = client
149 .activate_async(notification_handler, output_process_handler)
150 .unwrap();
151
152 Stream {
153 playing,
154 async_client,
155 input_port_names: vec![],
156 output_port_names: port_names,
157 }
158 }
159
160 /// Connect to the standard system outputs in jack, system:playback_1 and system:playback_2
161 /// This has to be done after the client is activated, doing it just after creating the ports doesn't work.
162 pub fn connect_to_system_outputs(&mut self) {
163 // Get the system ports
164 let system_ports = self.async_client.as_client().ports(
165 Some("system:playback_.*"),
166 None,
167 jack::PortFlags::empty(),
168 );
169
170 // Connect outputs from this client to the system playback inputs
171 for i in 0..self.output_port_names.len() {
172 if i >= system_ports.len() {
173 break;
174 }
175 match self
176 .async_client
177 .as_client()
178 .connect_ports_by_name(&self.output_port_names[i], &system_ports[i])
179 {
180 Ok(_) => (),
181 Err(e) => println!("Unable to connect to port with error {}", e),
182 }
183 }
184 }
185
186 /// Connect to the standard system outputs in jack, system:capture_1 and system:capture_2
187 /// This has to be done after the client is activated, doing it just after creating the ports doesn't work.
188 pub fn connect_to_system_inputs(&mut self) {
189 // Get the system ports
190 let system_ports = self.async_client.as_client().ports(
191 Some("system:capture_.*"),
192 None,
193 jack::PortFlags::empty(),
194 );
195
196 // Connect outputs from this client to the system playback inputs
197 for i in 0..self.input_port_names.len() {
198 if i >= system_ports.len() {
199 break;
200 }
201 match self
202 .async_client
203 .as_client()
204 .connect_ports_by_name(&system_ports[i], &self.input_port_names[i])
205 {
206 Ok(_) => (),
207 Err(e) => println!("Unable to connect to port with error {}", e),
208 }
209 }
210 }
211}
212
213impl StreamTrait for Stream {
214 fn play(&self) -> Result<(), PlayStreamError> {
215 self.playing.store(true, Ordering::SeqCst);
216 Ok(())
217 }
218
219 fn pause(&self) -> Result<(), PauseStreamError> {
220 self.playing.store(false, Ordering::SeqCst);
221 Ok(())
222 }
223
224 fn buffer_size(&self) -> Option<crate::FrameCount> {
225 Some(self.async_client.as_client().buffer_size() as crate::FrameCount)
226 }
227}
228
229type InputDataCallback = Box<dyn FnMut(&Data, &InputCallbackInfo) + Send + 'static>;
230type OutputDataCallback = Box<dyn FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static>;
231
232struct LocalProcessHandler {
233 /// No new ports are allowed to be created after the creation of the LocalProcessHandler as that would invalidate the buffer sizes
234 out_ports: Vec<jack::Port<jack::AudioOut>>,
235 in_ports: Vec<jack::Port<jack::AudioIn>>,
236
237 sample_rate: SampleRate,
238 buffer_size: usize,
239 input_data_callback: Option<InputDataCallback>,
240 output_data_callback: Option<OutputDataCallback>,
241
242 // JACK audio samples are 32-bit float (unless you do some custom dark magic)
243 temp_input_buffer: Vec<f32>,
244 temp_output_buffer: Vec<f32>,
245 playing: Arc<AtomicBool>,
246 creation_timestamp: std::time::Instant,
247 /// This should not be called on `process`, only on `buffer_size` because it can block.
248 error_callback_ptr: ErrorCallbackPtr,
249}
250
251impl LocalProcessHandler {
252 #[allow(clippy::too_many_arguments)]
253 fn new(
254 out_ports: Vec<jack::Port<jack::AudioOut>>,
255 in_ports: Vec<jack::Port<jack::AudioIn>>,
256 sample_rate: SampleRate,
257 buffer_size: usize,
258 input_data_callback: Option<InputDataCallback>,
259 output_data_callback: Option<OutputDataCallback>,
260 playing: Arc<AtomicBool>,
261 error_callback_ptr: ErrorCallbackPtr,
262 ) -> Self {
263 // These may be reallocated in the `buffer_size` callback.
264 let temp_input_buffer = vec![0.0; in_ports.len() * buffer_size];
265 let temp_output_buffer = vec![0.0; out_ports.len() * buffer_size];
266
267 LocalProcessHandler {
268 out_ports,
269 in_ports,
270 sample_rate,
271 buffer_size,
272 input_data_callback,
273 output_data_callback,
274 temp_input_buffer,
275 temp_output_buffer,
276 playing,
277 creation_timestamp: std::time::Instant::now(),
278 error_callback_ptr,
279 }
280 }
281}
282
283fn temp_buffer_to_data(temp_input_buffer: &mut [f32], total_buffer_size: usize) -> Data {
284 let slice = &mut temp_input_buffer[0..total_buffer_size];
285 let data: *mut () = slice.as_mut_ptr().cast();
286 let len = total_buffer_size;
287 unsafe { Data::from_parts(data, len, JACK_SAMPLE_FORMAT) }
288}
289
290impl jack::ProcessHandler for LocalProcessHandler {
291 fn process(&mut self, _: &jack::Client, process_scope: &jack::ProcessScope) -> jack::Control {
292 if !self.playing.load(Ordering::SeqCst) {
293 return jack::Control::Continue;
294 }
295
296 // This should be equal to self.buffer_size, but the implementation will
297 // work even if it is less. Will panic in `temp_buffer_to_data` if greater.
298 let current_frame_count = process_scope.n_frames() as usize;
299
300 // Get timestamp data
301 let (current_start_usecs, next_usecs_opt) = match process_scope.cycle_times() {
302 Ok(times) => (times.current_usecs, Some(times.next_usecs)),
303 Err(_) => {
304 // jack was unable to get the current time information
305 // Fall back to using Instants
306 let now = std::time::Instant::now();
307 let duration = now.duration_since(self.creation_timestamp);
308 (duration.as_micros() as u64, None)
309 }
310 };
311 let start_cycle_instant = micros_to_stream_instant(current_start_usecs);
312 let start_callback_instant = start_cycle_instant
313 .add(frames_to_duration(
314 process_scope.frames_since_cycle_start() as usize,
315 self.sample_rate,
316 ))
317 .expect("`playback` occurs beyond representation supported by `StreamInstant`");
318
319 if let Some(input_callback) = &mut self.input_data_callback {
320 // Let's get the data from the input ports and run the callback
321
322 let num_in_channels = self.in_ports.len();
323
324 // Read the data from the input ports into the temporary buffer
325 // Go through every channel and store its data in the temporary input buffer
326 for ch_ix in 0..num_in_channels {
327 let input_channel = &self.in_ports[ch_ix].as_slice(process_scope);
328 for i in 0..current_frame_count {
329 self.temp_input_buffer[ch_ix + i * num_in_channels] = input_channel[i];
330 }
331 }
332 // Create a slice of exactly current_frame_count frames
333 let data = temp_buffer_to_data(
334 &mut self.temp_input_buffer,
335 current_frame_count * num_in_channels,
336 );
337 // Create timestamp
338 let callback = start_callback_instant;
339 // Input data was made available at the start of the cycle (current_usecs).
340 let capture = start_cycle_instant;
341 let timestamp = crate::InputStreamTimestamp { callback, capture };
342 let info = crate::InputCallbackInfo { timestamp };
343 input_callback(&data, &info);
344 }
345
346 if let Some(output_callback) = &mut self.output_data_callback {
347 let num_out_channels = self.out_ports.len();
348
349 // Create a slice of exactly current_frame_count frames
350 let mut data = temp_buffer_to_data(
351 &mut self.temp_output_buffer,
352 current_frame_count * num_out_channels,
353 );
354 // Create timestamp
355 let callback = start_callback_instant;
356 // Use next_usecs (the hardware deadline for this cycle) when available; it is the
357 // exact instant at which the last sample written here will be consumed by the device.
358 let playback = match next_usecs_opt {
359 Some(next_usecs) => micros_to_stream_instant(next_usecs),
360 None => start_cycle_instant
361 .add(frames_to_duration(current_frame_count, self.sample_rate))
362 .expect("`playback` occurs beyond representation supported by `StreamInstant`"),
363 };
364 let timestamp = crate::OutputStreamTimestamp { callback, playback };
365 let info = crate::OutputCallbackInfo { timestamp };
366 output_callback(&mut data, &info);
367
368 // Deinterlace
369 for ch_ix in 0..num_out_channels {
370 let output_channel = &mut self.out_ports[ch_ix].as_mut_slice(process_scope);
371 for i in 0..current_frame_count {
372 output_channel[i] = self.temp_output_buffer[ch_ix + i * num_out_channels];
373 }
374 }
375 }
376
377 // Continue as normal
378 jack::Control::Continue
379 }
380
381 fn buffer_size(&mut self, _: &jack::Client, size: jack::Frames) -> jack::Control {
382 // The `buffer_size` callback is actually called on the process thread, but
383 // it does not need to be suitable for real-time use. Thus we can simply allocate
384 // new buffers here. It is also fine to call the error callback.
385 // Details: https://github.com/RustAudio/rust-jack/issues/137
386 let new_size = size as usize;
387 if new_size != self.buffer_size {
388 self.buffer_size = new_size;
389 self.temp_input_buffer = vec![0.0; self.in_ports.len() * new_size];
390 self.temp_output_buffer = vec![0.0; self.out_ports.len() * new_size];
391 let description = format!("buffer size changed to: {}", new_size);
392 if let Ok(mut mutex_guard) = self.error_callback_ptr.lock() {
393 let err = &mut *mutex_guard;
394 err(BackendSpecificError { description }.into());
395 }
396 }
397
398 jack::Control::Continue
399 }
400}
401
402fn micros_to_stream_instant(micros: u64) -> crate::StreamInstant {
403 crate::StreamInstant::from_nanos_i128(micros as i128 * 1_000)
404 .expect("`micros` out of range of `StreamInstant` representation")
405}
406
407// Convert the given duration in frames at the given sample rate to a `std::time::Duration`.
408fn frames_to_duration(frames: usize, rate: crate::SampleRate) -> std::time::Duration {
409 let secsf = frames as f64 / rate as f64;
410 let secs = secsf as u64;
411 let nanos = ((secsf - secs as f64) * 1_000_000_000.0) as u32;
412 std::time::Duration::new(secs, nanos)
413}
414
415/// Receives notifications from the JACK server. It is unclear if this may be run concurrent with itself under JACK2 specs
416/// so it needs to be Sync.
417struct JackNotificationHandler {
418 error_callback_ptr: ErrorCallbackPtr,
419 init_sample_rate_flag: Arc<AtomicBool>,
420}
421
422impl JackNotificationHandler {
423 pub fn new(error_callback_ptr: ErrorCallbackPtr) -> Self {
424 JackNotificationHandler {
425 error_callback_ptr,
426 init_sample_rate_flag: Arc::new(AtomicBool::new(false)),
427 }
428 }
429
430 fn send_error(&mut self, description: String) {
431 // This thread isn't the audio thread, it's fine to block
432 if let Ok(mut mutex_guard) = self.error_callback_ptr.lock() {
433 let err = &mut *mutex_guard;
434 err(BackendSpecificError { description }.into());
435 }
436 }
437}
438
439impl jack::NotificationHandler for JackNotificationHandler {
440 unsafe fn shutdown(&mut self, _status: jack::ClientStatus, reason: &str) {
441 self.send_error(format!("JACK was shut down for reason: {}", reason));
442 }
443
444 fn sample_rate(&mut self, _: &jack::Client, _srate: jack::Frames) -> jack::Control {
445 match self.init_sample_rate_flag.load(Ordering::SeqCst) {
446 false => {
447 // One of these notifications is sent every time a client is started.
448 self.init_sample_rate_flag.store(true, Ordering::SeqCst);
449 jack::Control::Continue
450 }
451 true => {
452 // The JACK server has changed the sample rate, invalidating this stream.
453 // The stream configuration must be rebuilt with the new sample rate.
454 if let Ok(mut cb) = self.error_callback_ptr.lock() {
455 cb(StreamError::StreamInvalidated);
456 }
457 jack::Control::Quit
458 }
459 }
460 }
461
462 fn xrun(&mut self, _: &jack::Client) -> jack::Control {
463 if let Ok(mut cb) = self.error_callback_ptr.lock() {
464 cb(StreamError::BufferUnderrun);
465 }
466 jack::Control::Continue
467 }
468}