device_change.rs

// NOTICE:
// Avoid running TestDeviceSwitcher with TestDevicePlugger or active full-duplex streams
// sequentially!
//
// The TestDeviceSwitcher cannot work with any test that will create an aggregate device that is
// soon being destroyed. The TestDeviceSwitcher will cache the available devices, upon it's
// created, as the candidates for the default device. Therefore, those created aggregate devices
// may be cached in TestDeviceSwitcher. However, those aggregate devices may be destroyed when
// TestDeviceSwitcher is using them or they are in the cached list of TestDeviceSwitcher.
//
// Running those tests by setting `test-threads=1` doesn't really help (e.g.,
// `cargo test test_register_device_changed_callback -- --ignored --nocapture --test-threads=1`).
// The aggregate device won't be destroyed immediately when `kAudioPlugInDestroyAggregateDevice`
// is set. As a result, the following tests requiring changing the devices will be run separately
// in the run_tests.sh script and marked by `ignore` by default.

use super::utils::{
    get_devices_info_in_scope, test_create_device_change_listener, test_device_in_scope,
    test_get_default_device, test_get_devices_in_scope,
    test_get_stream_with_default_data_callback_by_type, test_ops_stream_operation,
    test_set_default_device, Scope, StreamType, TestDevicePlugger, TestDeviceSwitcher,
};
use super::*;
use std::sync::{LockResult, WaitTimeoutResult};
use std::thread;
use std::time;

// Switch default devices used by the active streams, to test stream reinitialization
// ================================================================================================
#[ignore]
#[test]
fn test_switch_device() {
    test_switch_device_in_scope(Scope::Input);
    test_switch_device_in_scope(Scope::Output);
}

fn test_switch_device_in_scope(scope: Scope) {
    println!(
        "Switch default device for {:?} while the stream is working.",
        scope
    );

    // Do nothing if there is no 2 available devices at least.
    let devices = test_get_devices_in_scope(scope.clone());
    if devices.len() < 2 {
        println!("Need 2 devices for {:?} at least. Skip.", scope);
        return;
    }

    let mut device_switcher = TestDeviceSwitcher::new(scope.clone());

    let notifier = Arc::new(Notifier::new(0));
    let also_notifier = notifier.clone();
    let listener = run_serially(|| {
        test_create_device_change_listener(scope.clone(), move |_addresses| {
            let mut cnt = notifier.lock().unwrap();
            *cnt += 1;
            notifier.notify(cnt);
            NO_ERR
        })
    });
    run_serially(|| listener.start());

    let changed_watcher = Watcher::new(&also_notifier);
    test_get_started_stream_in_scope(scope.clone(), move |_stream| loop {
        let start_cnt = changed_watcher.lock().unwrap().clone();
        device_switcher.next();
        let mut guard = changed_watcher.lock().unwrap();
        guard = changed_watcher
            .wait_while(guard, |cnt| *cnt == start_cnt)
            .unwrap();
        if *guard >= devices.len() {
            break;
        }
    });
}

#[ignore]
#[test]
fn test_switch_while_paused() {
    test_switch_device_in_scope_while_paused(Scope::Input);
    test_switch_device_in_scope_while_paused(Scope::Output);
}

fn test_switch_device_in_scope_while_paused(scope: Scope) {
    println!(
        "Switch default device for {:?} while the stream is paused.",
        scope
    );

    // Do nothing if there is no 2 available devices at least.
    let devices = test_get_devices_in_scope(scope.clone());
    if devices.len() < 2 {
        println!("Need 2 devices for {:?} at least. Skip.", scope);
        return;
    }

    let mut device_switcher = TestDeviceSwitcher::new(scope.clone());

    let notifier = Arc::new(Notifier::new(0));
    let also_notifier = notifier.clone();
    let listener = run_serially(|| {
        test_create_device_change_listener(scope.clone(), move |_addresses| {
            let mut cnt = notifier.lock().unwrap();
            *cnt += 1;
            notifier.notify(cnt);
            NO_ERR
        })
    });
    run_serially(|| listener.start());

    let changed_watcher = Watcher::new(&also_notifier);
    test_get_started_stream_in_scope(scope.clone(), move |stream| loop {
        // pause the stream, change device, start the stream
        // peek under the hood to the find the device in use currently
        let stm = unsafe { &mut *(stream as *mut AudioUnitStream) };
        let before = if scope == Scope::Output {
            stm.core_stream_data.output_unit
        } else {
            stm.core_stream_data.input_unit
        };

        let check_devices = |current| {
            stm.queue.run_sync(|| {
                let (bus, unit, id) = if scope == Scope::Output {
                    (
                        AU_OUT_BUS,
                        stm.core_stream_data.output_unit,
                        stm.core_stream_data.output_device.id,
                    )
                } else {
                    (
                        AU_IN_BUS,
                        stm.core_stream_data.input_unit,
                        stm.core_stream_data.input_device.id,
                    )
                };
                let mut device_id: AudioDeviceID = 0;
                let mut size = std::mem::size_of::<AudioDeviceID>();
                let status = audio_unit_get_property(
                    unit,
                    kAudioOutputUnitProperty_CurrentDevice,
                    kAudioUnitScope_Global,
                    bus,
                    &mut device_id,
                    &mut size,
                );
                if status != NO_ERR {
                    panic!("Could not get device ID from audiounit");
                }
                assert_eq!(id, current);
                assert_eq!(device_id, current);
            });
        };

        check_devices(device_switcher.current());

        // Pause the stream, and change the default device
        assert_eq!(unsafe { OPS.stream_stop.unwrap()(stream) }, ffi::CUBEB_OK);

        let start_cnt = changed_watcher.lock().unwrap().clone();
        device_switcher.next();
        let mut guard = changed_watcher.lock().unwrap();
        guard = changed_watcher
            .wait_while(guard, |cnt| *cnt == start_cnt)
            .unwrap();
        if *guard >= devices.len() {
            break;
        }

        // Wait until the switch is effective and the stream has been marked for delayed
        // reinitialization. delayed_reinit can only be access from the queue thread. Depending on
        // the setup this can take some time so we sleep a bit to not hammer the main thread.
        let mut switched = false;
        while !switched {
            switched = stm.queue.run_sync(|| stm.delayed_reinit).unwrap();
            if !switched {
                let ten_millis = time::Duration::from_millis(10);
                thread::sleep(ten_millis);
            }
        }

        // Start the stream, and check that the device in use isn't the same as before pausing
        assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);

        check_devices(device_switcher.current());

        let after = if scope == Scope::Output {
            stm.core_stream_data.output_unit
        } else {
            stm.core_stream_data.input_unit
        };
        assert_ne!(before, after);
    });
}

fn test_get_started_stream_in_scope<F>(scope: Scope, operation: F)
where
    F: FnOnce(*mut ffi::cubeb_stream),
{
    use std::f32::consts::PI;
    const SAMPLE_FREQUENCY: u32 = 48_000;

    // Make sure the parameters meet the requirements of AudioUnitContext::stream_init
    // (in the comments).
    let mut stream_params = ffi::cubeb_stream_params::default();
    stream_params.format = ffi::CUBEB_SAMPLE_S16NE;
    stream_params.rate = SAMPLE_FREQUENCY;
    stream_params.prefs = ffi::CUBEB_STREAM_PREF_NONE;
    stream_params.channels = 1;
    stream_params.layout = ffi::CUBEB_LAYOUT_MONO;

    let (input_params, output_params) = match scope {
        Scope::Input => (
            &mut stream_params as *mut ffi::cubeb_stream_params,
            ptr::null_mut(),
        ),
        Scope::Output => (
            ptr::null_mut(),
            &mut stream_params as *mut ffi::cubeb_stream_params,
        ),
    };

    extern "C" fn state_callback(
        stream: *mut ffi::cubeb_stream,
        user_ptr: *mut c_void,
        state: ffi::cubeb_state,
    ) {
        assert!(!stream.is_null());
        assert!(!user_ptr.is_null());
        assert_ne!(state, ffi::CUBEB_STATE_ERROR);
    }

    extern "C" fn input_data_callback(
        stream: *mut ffi::cubeb_stream,
        user_ptr: *mut c_void,
        input_buffer: *const c_void,
        output_buffer: *mut c_void,
        nframes: i64,
    ) -> i64 {
        assert!(!stream.is_null());
        assert!(!user_ptr.is_null());
        assert!(!input_buffer.is_null());
        assert!(output_buffer.is_null());
        nframes
    }

    let mut position: i64 = 0; // TODO: Use Atomic instead.

    fn f32_to_i16_sample(x: f32) -> i16 {
        (x * f32::from(i16::max_value())) as i16
    }

    extern "C" fn output_data_callback(
        stream: *mut ffi::cubeb_stream,
        user_ptr: *mut c_void,
        input_buffer: *const c_void,
        output_buffer: *mut c_void,
        nframes: i64,
    ) -> i64 {
        assert!(!stream.is_null());
        assert!(!user_ptr.is_null());
        assert!(input_buffer.is_null());
        assert!(!output_buffer.is_null());

        let buffer = unsafe {
            let ptr = output_buffer as *mut i16;
            let len = nframes as usize;
            slice::from_raw_parts_mut(ptr, len)
        };

        let position = unsafe { &mut *(user_ptr as *mut i64) };

        // Generate tone on the fly.
        for data in buffer.iter_mut() {
            let t1 = (2.0 * PI * 350.0 * (*position) as f32 / SAMPLE_FREQUENCY as f32).sin();
            let t2 = (2.0 * PI * 440.0 * (*position) as f32 / SAMPLE_FREQUENCY as f32).sin();
            *data = f32_to_i16_sample(0.5 * (t1 + t2));
            *position += 1;
        }

        nframes
    }

    test_ops_stream_operation(
        "stream",
        ptr::null_mut(), // Use default input device.
        input_params,
        ptr::null_mut(), // Use default output device.
        output_params,
        4096, // TODO: Get latency by get_min_latency instead ?
        match scope {
            Scope::Input => Some(input_data_callback),
            Scope::Output => Some(output_data_callback),
        },
        Some(state_callback),
        &mut position as *mut i64 as *mut c_void,
        |stream| {
            assert_eq!(unsafe { OPS.stream_start.unwrap()(stream) }, ffi::CUBEB_OK);
            operation(stream);
            assert_eq!(unsafe { OPS.stream_stop.unwrap()(stream) }, ffi::CUBEB_OK);
        },
    );
}

// Plug and unplug devices, to test device collection changed callback
// ================================================================================================
#[ignore]
#[test]
fn test_plug_and_unplug_device() {
    test_plug_and_unplug_device_in_scope(Scope::Input);
    test_plug_and_unplug_device_in_scope(Scope::Output);
}

fn test_plug_and_unplug_device_in_scope(scope: Scope) {
    let default_device = test_get_default_device(scope.clone());
    if default_device.is_none() {
        println!("No device for {:?} to test", scope);
        return;
    }

    println!("Run test for {:?}", scope);
    println!("NOTICE: The test will hang if the default input or output is an aggregate device.\nWe will fix this later.");

    let default_device = default_device.unwrap();
    let is_input = test_device_in_scope(default_device, Scope::Input);
    let is_output = test_device_in_scope(default_device, Scope::Output);

    let mut context = AudioUnitContext::new();

    // Register the devices-changed callbacks.
    #[derive(Clone, PartialEq)]
    struct Counts {
        input: u32,
        output: u32,
    }
    impl Counts {
        fn new() -> Self {
            Self {
                input: 0,
                output: 0,
            }
        }
    }
    let counts = Arc::new(Notifier::new(Counts::new()));
    let counts_notifier_ptr = counts.as_ref() as *const Notifier<Counts>;

    assert!(context
        .register_device_collection_changed(
            DeviceType::INPUT,
            Some(input_changed_callback),
            counts_notifier_ptr as *mut c_void,
        )
        .is_ok());

    assert!(context
        .register_device_collection_changed(
            DeviceType::OUTPUT,
            Some(output_changed_callback),
            counts_notifier_ptr as *mut c_void,
        )
        .is_ok());

    let counts_watcher = Watcher::new(&counts);

    let mut device_plugger = TestDevicePlugger::new(scope).unwrap();

    {
        // Simulate adding devices and monitor the devices-changed callbacks.
        let mut counts_guard = counts.lock().unwrap();
        let counts_start = counts_guard.clone();

        assert!(device_plugger.plug().is_ok());

        counts_guard = counts_watcher
            .wait_while(counts_guard, |counts| {
                (is_input && counts.input == counts_start.input)
                    || (is_output && counts.output == counts_start.output)
            })
            .unwrap();

        // Check changed count.
        assert_eq!(counts_guard.input, if is_input { 1 } else { 0 });
        assert_eq!(counts_guard.output, if is_output { 1 } else { 0 });
    }

    {
        // Simulate removing devices and monitor the devices-changed callbacks.
        let mut counts_guard = counts.lock().unwrap();
        let counts_start = counts_guard.clone();

        assert!(device_plugger.unplug().is_ok());

        counts_guard = counts_watcher
            .wait_while(counts_guard, |counts| {
                (is_input && counts.input == counts_start.input)
                    || (is_output && counts.output == counts_start.output)
            })
            .unwrap();

        // Check changed count.
        assert_eq!(counts_guard.input, if is_input { 2 } else { 0 });
        assert_eq!(counts_guard.output, if is_output { 2 } else { 0 });
    }

    extern "C" fn input_changed_callback(context: *mut ffi::cubeb, data: *mut c_void) {
        println!(
            "Input device collection @ {:p} is changed. Data @ {:p}",
            context, data
        );
        let notifier = unsafe { &*(data as *const Notifier<Counts>) };
        {
            let mut counts = notifier.lock().unwrap();
            counts.input += 1;
            notifier.notify(counts);
        }
    }

    extern "C" fn output_changed_callback(context: *mut ffi::cubeb, data: *mut c_void) {
        println!(
            "output device collection @ {:p} is changed. Data @ {:p}",
            context, data
        );
        let notifier = unsafe { &*(data as *const Notifier<Counts>) };
        {
            let mut counts = notifier.lock().unwrap();
            counts.output += 1;
            notifier.notify(counts);
        }
    }

    context.register_device_collection_changed(DeviceType::OUTPUT, None, ptr::null_mut());
    context.register_device_collection_changed(DeviceType::INPUT, None, ptr::null_mut());
}

// Switch default devices used by the active streams, to test device changed callback
// ================================================================================================
#[ignore]
#[test]
fn test_register_device_changed_callback_to_check_default_device_changed_input() {
    test_register_device_changed_callback_to_check_default_device_changed(StreamType::INPUT);
}

#[ignore]
#[test]
fn test_register_device_changed_callback_to_check_default_device_changed_output() {
    test_register_device_changed_callback_to_check_default_device_changed(StreamType::OUTPUT);
}

#[ignore]
#[test]
fn test_register_device_changed_callback_to_check_default_device_changed_duplex() {
    test_register_device_changed_callback_to_check_default_device_changed(StreamType::DUPLEX);
}

fn test_register_device_changed_callback_to_check_default_device_changed(stm_type: StreamType) {
    println!("NOTICE: The test will hang if the default input or output is an aggregate device.\nWe will fix this later.");

    let inputs = if stm_type.contains(StreamType::INPUT) {
        let devices = test_get_devices_in_scope(Scope::Input).len();
        if devices >= 2 {
            Some(devices)
        } else {
            None
        }
    } else {
        None
    };

    let outputs = if stm_type.contains(StreamType::OUTPUT) {
        let devices = test_get_devices_in_scope(Scope::Output).len();
        if devices >= 2 {
            Some(devices)
        } else {
            None
        }
    } else {
        None
    };

    if inputs.is_none() && outputs.is_none() {
        println!("No enough devices to run the test!");
        return;
    }

    let changed_count = Arc::new(Notifier::new(0u32));
    let notifier_ptr = changed_count.as_ref() as *const Notifier<u32>;

    test_get_stream_with_device_changed_callback(
        "stream: test callback for default device changed",
        stm_type,
        None, // Use default input device.
        None, // Use default output device.
        notifier_ptr as *mut c_void,
        state_callback,
        device_changed_callback,
        |stream| {
            // If the duplex stream uses different input and output device,
            // an aggregate device will be created and it will work for this duplex stream.
            // This aggregate device will be added into the device list, but it won't
            // be assigned to the default device, since the device list for setting
            // default device is cached upon {input, output}_device_switcher is initialized.

            let changed_watcher = Watcher::new(&changed_count);

            if let Some(devices) = inputs {
                let mut device_switcher = TestDeviceSwitcher::new(Scope::Input);
                for _ in 0..devices {
                    // While the stream is re-initializing for the default device switch,
                    // switching for the default device again will be ignored.
                    while stream.switching_device.load(atomic::Ordering::SeqCst) {
                        std::hint::spin_loop()
                    }
                    let guard = changed_watcher.lock().unwrap();
                    let start_cnt = guard.clone();
                    device_switcher.next();
                    changed_watcher
                        .wait_while(guard, |cnt| *cnt == start_cnt)
                        .unwrap();
                }
            }

            if let Some(devices) = outputs {
                let mut device_switcher = TestDeviceSwitcher::new(Scope::Output);
                for _ in 0..devices {
                    // While the stream is re-initializing for the default device switch,
                    // switching for the default device again will be ignored.
                    while stream.switching_device.load(atomic::Ordering::SeqCst) {
                        std::hint::spin_loop()
                    }
                    let guard = changed_watcher.lock().unwrap();
                    let start_cnt = guard.clone();
                    device_switcher.next();
                    changed_watcher
                        .wait_while(guard, |cnt| *cnt == start_cnt)
                        .unwrap();
                }
            }
        },
    );

    extern "C" fn state_callback(
        stream: *mut ffi::cubeb_stream,
        _user_ptr: *mut c_void,
        state: ffi::cubeb_state,
    ) {
        assert!(!stream.is_null());
        assert_ne!(state, ffi::CUBEB_STATE_ERROR);
    }

    extern "C" fn device_changed_callback(data: *mut c_void) {
        println!("Device change callback. data @ {:p}", data);
        let notifier = unsafe { &*(data as *const Notifier<u32>) };
        let mut count_guard = notifier.lock().unwrap();
        *count_guard += 1;
        notifier.notify(count_guard);
    }
}

// Unplug the devices used by the active streams, to test
// 1) device changed callback, or state callback
// 2) stream reinitialization that may race with stream destroying
// ================================================================================================

// Input-only stream
// -----------------

// Unplug the non-default input device for an input stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_input_stream_after_unplugging_a_nondefault_input_device() {
    // The stream can be destroyed before running device-changed event handler
    test_unplug_a_device_on_an_active_stream(StreamType::INPUT, Scope::Input, false, 0);
}

#[ignore]
#[test]
fn test_suspend_input_stream_by_unplugging_a_nondefault_input_device() {
    // Expect to get an error state callback by device-changed event handler
    test_unplug_a_device_on_an_active_stream(StreamType::INPUT, Scope::Input, false, 2000);
}

// Unplug the default input device for an input stream
// ------------------------------------------------------------------------------------------------
#[ignore]
#[test]
fn test_destroy_input_stream_after_unplugging_a_default_input_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes, at the same when
    // the stream is being destroyed
    test_unplug_a_device_on_an_active_stream(StreamType::INPUT, Scope::Input, true, 0);
}

#[ignore]
#[test]
fn test_reinit_input_stream_by_unplugging_a_default_input_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes
    test_unplug_a_device_on_an_active_stream(StreamType::INPUT, Scope::Input, true, 2000);
}

// Output-only stream
// ------------------

// Unplug the non-default output device for an output stream
// ------------------------------------------------------------------------------------------------
#[ignore]
#[test]
fn test_destroy_output_stream_after_unplugging_a_nondefault_output_device() {
    test_unplug_a_device_on_an_active_stream(StreamType::OUTPUT, Scope::Output, false, 0);
}

#[ignore]
#[test]
fn test_suspend_output_stream_by_unplugging_a_nondefault_output_device() {
    test_unplug_a_device_on_an_active_stream(StreamType::OUTPUT, Scope::Output, false, 2000);
}

// Unplug the default output device for an output stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_output_stream_after_unplugging_a_default_output_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes, at the same when
    // the stream is being destroyed
    test_unplug_a_device_on_an_active_stream(StreamType::OUTPUT, Scope::Output, true, 0);
}

#[ignore]
#[test]
fn test_reinit_output_stream_by_unplugging_a_default_output_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes
    test_unplug_a_device_on_an_active_stream(StreamType::OUTPUT, Scope::Output, true, 2000);
}

// Duplex stream
// -------------

// Unplug the non-default input device for a duplex stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_duplex_stream_after_unplugging_a_nondefault_input_device() {
    // The stream can be destroyed before running device-changed event handler
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Input, false, 0);
}

#[ignore]
#[test]
fn test_suspend_duplex_stream_by_unplugging_a_nondefault_input_device() {
    // Expect to get an error state callback by device-changed event handler
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Input, false, 2000);
}

// Unplug the non-default output device for a duplex stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_duplex_stream_after_unplugging_a_nondefault_output_device() {
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Output, false, 0);
}

#[ignore]
#[test]
fn test_suspend_duplex_stream_by_unplugging_a_nondefault_output_device() {
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Output, false, 2000);
}

// Unplug the non-default in-out device for a duplex stream
// ------------------------------------------------------------------------------------------------
// TODO: Implement an in-out TestDevicePlugger

// Unplug the default input device for a duplex stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_duplex_stream_after_unplugging_a_default_input_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes, at the same when
    // the stream is being destroyed
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Input, true, 0);
}

#[ignore]
#[test]
fn test_reinit_duplex_stream_by_unplugging_a_default_input_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Input, true, 2000);
}

// Unplug the default ouput device for a duplex stream
// ------------------------------------------------------------------------------------------------

#[ignore]
#[test]
fn test_destroy_duplex_stream_after_unplugging_a_default_output_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes, at the same when
    // the stream is being destroyed
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Output, true, 0);
}

#[ignore]
#[test]
fn test_reinit_duplex_stream_by_unplugging_a_default_output_device() {
    // Expect to get an device-changed callback by device-changed event handler,
    // which will reinitialize the stream behind the scenes
    test_unplug_a_device_on_an_active_stream(StreamType::DUPLEX, Scope::Output, true, 2000);
}

fn test_unplug_a_device_on_an_active_stream(
    stream_type: StreamType,
    device_scope: Scope,
    set_device_to_default: bool,
    wait_up_to_ms: u64,
) {
    let has_input = test_get_default_device(Scope::Input).is_some();
    let has_output = test_get_default_device(Scope::Output).is_some();

    if stream_type.contains(StreamType::INPUT) && !has_input {
        println!("No input device for input or duplex stream.");
        return;
    }

    if stream_type.contains(StreamType::OUTPUT) && !has_output {
        println!("No output device for output or duplex stream.");
        return;
    }

    let default_device_before_plugging = test_get_default_device(device_scope.clone()).unwrap();
    println!(
        "Before plugging, default {:?} device is {}",
        device_scope, default_device_before_plugging
    );

    let mut plugger = TestDevicePlugger::new(device_scope.clone()).unwrap();
    assert!(plugger.plug().is_ok());
    assert_ne!(plugger.get_device_id(), kAudioObjectUnknown);
    println!(
        "Create plugger device: {} for {:?}",
        plugger.get_device_id(),
        device_scope
    );

    let default_device_after_plugging = test_get_default_device(device_scope.clone()).unwrap();
    println!(
        "After plugging, default {:?} device is {}",
        device_scope, default_device_after_plugging
    );

    // The new device, plugger, is possible to be set to the default device.
    // Before running the test, we need to set the default device to the correct one.
    if set_device_to_default {
        // plugger should be the default device for the test.
        // If it's not, then set it to the default device.
        if default_device_after_plugging != plugger.get_device_id() {
            let prev_def_dev =
                test_set_default_device(plugger.get_device_id(), device_scope.clone()).unwrap();
            assert_eq!(prev_def_dev, default_device_after_plugging);
        }
    } else {
        // plugger should NOT be the default device for the test.
        // If it is, reset the default device to another one.
        if default_device_after_plugging == plugger.get_device_id() {
            let prev_def_dev =
                test_set_default_device(default_device_before_plugging, device_scope.clone())
                    .unwrap();
            assert_eq!(prev_def_dev, default_device_after_plugging);
        }
    }

    // Ignore the return devices' info since we only need to print them.
    let _ = get_devices_info_in_scope(device_scope.clone());
    println!(
        "Current default {:?} device is {}",
        device_scope,
        test_get_default_device(device_scope.clone()).unwrap()
    );

    let (input_device, output_device) = match device_scope {
        Scope::Input => (
            if set_device_to_default {
                None // default input device.
            } else {
                Some(plugger.get_device_id())
            },
            None,
        ),
        Scope::Output => (
            None,
            if set_device_to_default {
                None // default output device.
            } else {
                Some(plugger.get_device_id())
            },
        ),
    };

    #[derive(Clone, PartialEq)]
    struct Data {
        changed_count: u32,
        states: Vec<ffi::cubeb_state>,
    }

    impl Data {
        fn new() -> Self {
            Self {
                changed_count: 0,
                states: vec![],
            }
        }
    }

    let notifier = Arc::new(Notifier::new(Data::new()));
    let notifier_ptr = notifier.as_ref() as *const Notifier<Data>;

    test_get_stream_with_device_changed_callback(
        "stream: test stream reinit/destroy after unplugging a device",
        stream_type,
        input_device,
        output_device,
        notifier_ptr as *mut c_void,
        state_callback,
        device_changed_callback,
        |stream| {
            assert_eq!(stream.start(), Ok(()));

            println!(
                "Stream runs on the device {} for {:?}",
                plugger.get_device_id(),
                device_scope
            );

            let dev = plugger.get_device_id();
            let start_changed_count = {
                let guard = notifier.lock().unwrap();
                assert_eq!(guard.states.last().unwrap(), &ffi::CUBEB_STATE_STARTED);
                guard.changed_count.clone()
            };

            assert!(plugger.unplug().is_ok());

            let changed_watcher = Watcher::new(&notifier);

            if set_device_to_default {
                // The stream will be reinitialized if it follows the default input or output device.
                println!("Waiting for default device to change and reinit");
                let mut data_guard = notifier.lock().unwrap();
                data_guard = changed_watcher
                    .wait_while(data_guard, |data| {
                        data.changed_count == start_changed_count
                            || data.states.last().unwrap_or(&ffi::CUBEB_STATE_ERROR)
                                != &ffi::CUBEB_STATE_STARTED
                    })
                    .unwrap();
            } else if wait_up_to_ms > 0 {
                // stream can be dropped immediately before device-changed callback
                // so we only check the states if we wait for it explicitly.
                println!("Waiting for non-default device to enter error state");
                let mut data_guard = notifier.lock().unwrap();
                let (new_guard, timeout_res) = changed_watcher
                    .wait_timeout_while(data_guard, Duration::from_millis(wait_up_to_ms), |data| {
                        data.states.last().unwrap_or(&ffi::CUBEB_STATE_STARTED)
                            != &ffi::CUBEB_STATE_ERROR
                    })
                    .unwrap();
                assert!(!timeout_res.timed_out());
                data_guard = new_guard;
            }

            println!(
                "Device {} for {:?} has been unplugged. The default {:?} device now is {}",
                dev,
                device_scope,
                device_scope,
                test_get_default_device(device_scope.clone()).unwrap()
            );

            println!("The stream is going to be destroyed soon");
        },
    );

    extern "C" fn state_callback(
        stream: *mut ffi::cubeb_stream,
        user_ptr: *mut c_void,
        state: ffi::cubeb_state,
    ) {
        println!("Device change callback. user_ptr @ {:p}", user_ptr);
        assert!(!stream.is_null());
        println!(
            "state: {}",
            match state {
                ffi::CUBEB_STATE_STARTED => "started",
                ffi::CUBEB_STATE_STOPPED => "stopped",
                ffi::CUBEB_STATE_DRAINED => "drained",
                ffi::CUBEB_STATE_ERROR => "error",
                _ => "unknown",
            }
        );
        let notifier = unsafe { &mut *(user_ptr as *mut Notifier<Data>) };
        let mut data_guard = notifier.lock().unwrap();
        data_guard.states.push(state);
        notifier.notify(data_guard);
    }

    extern "C" fn device_changed_callback(user_ptr: *mut c_void) {
        println!("Device change callback. user_ptr @ {:p}", user_ptr);
        let notifier = unsafe { &mut *(user_ptr as *mut Notifier<Data>) };
        let mut data_guard = notifier.lock().unwrap();
        data_guard.changed_count += 1;
        notifier.notify(data_guard);
    }
}

struct Notifier<T> {
    value: Mutex<T>,
    cvar: Condvar,
}

impl<T> Notifier<T> {
    fn new(value: T) -> Self {
        Self {
            value: Mutex::new(value),
            cvar: Condvar::new(),
        }
    }

    fn lock(&self) -> LockResult<MutexGuard<'_, T>> {
        self.value.lock()
    }

    fn notify(&self, _guard: MutexGuard<'_, T>) {
        self.cvar.notify_all();
    }
}

struct Watcher<T: Clone + PartialEq> {
    notifier: Arc<Notifier<T>>,
}

impl<T: Clone + PartialEq> Watcher<T> {
    fn new(value: &Arc<Notifier<T>>) -> Self {
        Self {
            notifier: Arc::clone(value),
        }
    }

    fn lock(&self) -> LockResult<MutexGuard<'_, T>> {
        self.notifier.lock()
    }

    fn wait_while<'a, F>(
        &self,
        guard: MutexGuard<'a, T>,
        condition: F,
    ) -> LockResult<MutexGuard<'a, T>>
    where
        F: FnMut(&mut T) -> bool,
    {
        self.notifier.cvar.wait_while(guard, condition)
    }

    fn wait_timeout_while<'a, F>(
        &self,
        guard: MutexGuard<'a, T>,
        dur: Duration,
        condition: F,
    ) -> LockResult<(MutexGuard<'a, T>, WaitTimeoutResult)>
    where
        F: FnMut(&mut T) -> bool,
    {
        self.notifier.cvar.wait_timeout_while(guard, dur, condition)
    }
}

fn test_get_stream_with_device_changed_callback<F>(
    name: &'static str,
    stm_type: StreamType,
    input_device: Option<AudioObjectID>,
    output_device: Option<AudioObjectID>,
    data: *mut c_void,
    state_callback: extern "C" fn(*mut ffi::cubeb_stream, *mut c_void, ffi::cubeb_state),
    device_changed_callback: extern "C" fn(*mut c_void),
    operation: F,
) where
    F: FnOnce(&mut AudioUnitStream),
{
    test_get_stream_with_default_data_callback_by_type(
        name,
        stm_type,
        input_device,
        output_device,
        state_callback,
        data,
        |stream| {
            assert!(stream
                .register_device_changed_callback(Some(device_changed_callback))
                .is_ok());
            operation(stream);
            assert!(stream.register_device_changed_callback(None).is_ok());
        },
    );
}