ffmpeg playback with rust

Cargo.toml

[package]
name = "player"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
cpal = "0.13.5"
ffmpeg-next = "5.0.3"
ringbuf = "0.2.8"

main.rs

extern crate ffmpeg_next as ffmpeg;

use cpal::{Sample, SampleFormat};
use cpal::traits::{HostTrait, DeviceTrait, StreamTrait};
use ffmpeg::{codec, decoder, frame, format, media};
use ffmpeg::software::resampling::context::Context as SwrContext;
use ringbuf::RingBuffer;

fn init_device_defalut() -> (cpal::Device, cpal::SupportedStreamConfig) {
    let device = cpal::default_host()
        .default_output_device()
        .expect("无法打开音频设备");

    let supported_config_range = device.supported_output_configs()
        .expect("获取音频设备输出格式时发生错误")
        .next()
        .expect("音频设备没有支持的输出格式");

    // 获取最高采样率的格式
    (device, supported_config_range.with_max_sample_rate())
}

fn ffmpeg_frame_to_slice<T: frame::audio::Sample>(frame: &frame::Audio) -> &[T] {
    if !frame.is_packed() {
        panic!("音频帧数据不是交错格式");
    }

    if !T::is_valid(frame.format(), frame.channels()) {
        panic!("无效的音频帧数据");
    }

    unsafe {
        std::slice::from_raw_parts((*frame.as_ptr()).data[0] as *const T, frame.samples() * frame.channels() as usize)
    }
}

fn blocking_write_buffer<T: Copy>(slice: &[T], producer: &mut ringbuf::Producer<T>) {
    // 先分块，避免缓冲区容量比帧小，产生死锁
    let chunks = slice.chunks(30752);
    for chunk in chunks {
        while producer.remaining() < chunk.len() {
            // 缓冲区已满，循环等待
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        // 向缓冲区写入数据
        producer.push_slice(chunk);
    }
}

fn decode_to_buffer (
    decoder: &mut decoder::Audio,
    producer_f32: &mut ringbuf::Producer<f32>,
    resampler: &mut SwrContext,
) -> Result<(), ffmpeg::Error> {
    let mut decoded = frame::Audio::empty();
    while decoder.receive_frame(&mut decoded).is_ok() {
        let mut resampled = frame::Audio::empty();
        let mut delay = resampler.run(&decoded, &mut resampled)?;
        loop {
            // 将重采样后的将音频数据写入对应的缓冲区中
            blocking_write_buffer(ffmpeg_frame_to_slice(&resampled), producer_f32);
            // 输出的大小装不下的部分会在重采样器里缓存，需要循环读取到缓存为空
            if delay == None {
                break;
            }
            delay = resampler.flush(&mut resampled)?;
        }
    }
    Ok(())
}

fn main() {
    // ffmpeg 日志级别
    ffmpeg::util::log::set_level(ffmpeg::util::log::Level::Debug);

    // 输入部分

    // 打开音频文件
    let file = &std::env::args().nth(1).expect("没有提供文件路径");
    let (device, device_config) = init_device_defalut();

    // 打开文件，获取音频流
    let mut input_ctx = format::input(&file).expect("打开文件时发生错误");
    let stream = input_ctx.streams().best(media::Type::Audio).expect("不是音频文件");
    let stream_index = stream.index();

    // 创建解码器
    let context = codec::context::Context::from_parameters(stream.parameters()).expect("无法解析音频编码");
    let mut decoder = context.decoder().audio().expect("找不到对应的音频解码器");
    // 将音频流相关信息拷贝到 AVCodecContext 中
    decoder.set_parameters(stream.parameters()).expect("创建解码器失败");

    if decoder.channel_layout().is_empty() {
        decoder.set_channel_layout(ffmpeg::ChannelLayout::default(decoder.channels().into()))
    }

    // 创建重采样器，转换音频数据为音频设备支持的格式
    let mut resampler =  SwrContext::get(
        // 输入格式
        decoder.format(),
        decoder.channel_layout(),
        decoder.rate(),
        // 输出格式 (一律使用32位浮点)
        format::Sample::F32(format::sample::Type::Packed),
        ffmpeg::ChannelLayout::default(device_config.channels().into()),
        device_config.sample_rate().0
    ).expect("创建重采样器失败");

    // 输出部分

    // 创建音频缓冲区
    let buffer_f32 = RingBuffer::<f32>::new(61504);
    let (mut producer_f32, mut consumer_f32) = buffer_f32.split();

    // 创建音频设备输出流，从缓冲区读取数据
    let error_callback = |err| {
        eprintln!("输出音频时发生错误: {}", err)
    };

    let device_output_stream = match device_config.sample_format() {
        SampleFormat::I16 => device.build_output_stream(&device_config.into(), move |data: &mut[i16], _| {
            for sample in data.iter_mut() {
                // 如果缓冲区里有数据则复制，否则生成静音数据
                *sample = consumer_f32.pop().unwrap_or(Sample::from(&0f32)).to_i16();
            }
        }, error_callback),
        SampleFormat::U16 => device.build_output_stream(&device_config.into(), move |data: &mut[u16], _| {
            for sample in data.iter_mut() {
                *sample = consumer_f32.pop().unwrap_or(Sample::from(&0f32)).to_u16();
            }
        }, error_callback),
        SampleFormat::F32 => device.build_output_stream(&device_config.into(), move |data: &mut[f32], _| {
            for sample in data.iter_mut() {
                *sample = consumer_f32.pop().unwrap_or(Sample::from(&0f32));
            }
        }, error_callback),
    }.unwrap();

    // 播放缓冲区中的音频
    device_output_stream.play().unwrap();

    // 音频解码部分
    for (stream, packet) in input_ctx.packets() {
        if stream.index() == stream_index {
            decoder.send_packet(&packet).unwrap();
            decode_to_buffer(&mut decoder, &mut producer_f32, &mut resampler).unwrap();
        }
    }

    decoder.send_eof().unwrap();
    decode_to_buffer(&mut decoder, &mut producer_f32, &mut resampler).unwrap();
}