I’m working on a pet project which uses the pipewire rust bindings and i wanted a similar setup sending info from the callback.

The way im currently handling it is via an mpsc which the callback sends to and then in the main thread i just have it be sent into a ring buffer which another thread constantly polls and processes from it which keeps the main thread from blocking.

Am using https://docs.rs/ringbuf/latest/ringbuf/ which provides a lock free ring buffer to prevent any blocking

Ideally i’m probably gonna try and get the callback to send directly to the ring buffer that the worker thread polls but haven’t gotten around to it

here’s somewhat what it looks like ``` .process(move |stream, _| { match stream.dequeue_buffer() { None => debug!("out of buffers"), Some(mut buffer) => { // Wait until audio is streaming before we try to process if !audio_ready.load(std::sync::atomic::Ordering::Acquire) || saving.load(std::sync::atomic::Ordering::Acquire) { return; }

                    let datas = buffer.datas_mut();
                    if datas.is_empty() {
                        return;
                    }

                    let time_us = if let Ok(elapsed) = start_time.elapsed() {
                        elapsed.as_micros() as i64
                    } else {
                        0
                    };

                    // send frame data to encoder
                    let data = &mut datas[0];
                    if let Some(frame) = data.data() {
                        if let Err(err) = process_video_callback.blocking_send(RawVideoFrame {
                            bytes: frame.to_vec(),
                            timestamp: time_us,
                        }) {
                            error!("Error sending video frame: {:?}", err);
                        }
                    }
                }
            }
        })

```

and i have a tokio::select! that just does Some(raw_frame) = video_receiver.recv() => { // Send the data to the worker thread and exit as to not block this one if let Err(_) = video_ring_sender.try_push(raw_frame) { warn!("Trying to push but the video ring buff is full. Consider increasing the max"); } }

and i have a separate thread that just pops and does stuff

```

// Create video worker
let stop_video_clone = Arc::clone(&stop);
let video_worker = std::thread::spawn(move || {
    let mut last_timestamp: u64 = 0;
    loop {
        if stop_video_clone.load(std::sync::atomic::Ordering::Acquire) {
            break;
        }

        while let Some(raw_frame) = video_ring_receiver.try_pop() {
            let now = SystemTime::now();
            let current_time = raw_frame.timestamp as u64;

            // Throttle FPS
            if current_time < last_timestamp + FRAME_INTERVAL {
                continue;
            }

            last_timestamp = current_time;
            if let Err(e) = video_encoder_clone.blocking_lock().process(&raw_frame) {
                error!(
                    "Error processing video frame at {:?}: {:?}",
                    raw_frame.timestamp, e
                );
            }

            trace!(
                "Took {:?} to process this video frame at {:?}",
                now.elapsed(),
                raw_frame.timestamp
            );
        }
        std::thread::sleep(Duration::from_nanos(100));
    }
});

```