nandi/jolt-native · mod.rs

⑂ 2ab40bf193b6a39bc8d1397cc36d17572bbf59cd

mod.rs · 479 lines · 16.2 KBRust    Blame HistoryRaw
Lift freeq's AV media plane out of sleek   1use futures::executor::block_on;
2use pulseaudio::protocol;
4mod stream;
6pub use stream::Stream;
8use crate::{
  traits::{DeviceTrait, HostTrait},
  BackendSpecificError, BuildStreamError, Data, DefaultStreamConfigError, DeviceDescription,
  DeviceDescriptionBuilder, DeviceDirection, DeviceId, DeviceIdError, DeviceNameError,
  DevicesError, FrameCount, HostId, HostUnavailable, InputCallbackInfo, OutputCallbackInfo,
  SampleFormat, StreamConfig, StreamError, SupportedBufferSize, SupportedStreamConfig,
  SupportedStreamConfigRange, SupportedStreamConfigsError,
17const PULSE_FORMATS: &[SampleFormat] = &[
  SampleFormat::U8,
  SampleFormat::I16,
  SampleFormat::I24,
  SampleFormat::I32,
  SampleFormat::F32,
25impl TryFrom<protocol::SampleFormat> for SampleFormat {
  type Error = ();
  fn try_from(spec: protocol::SampleFormat) -> Result<Self, Self::Error> {
      match spec {
          protocol::SampleFormat::U8 => Ok(SampleFormat::U8),
          protocol::SampleFormat::S16Le | protocol::SampleFormat::S16Be => Ok(SampleFormat::I16),
          protocol::SampleFormat::S24Le | protocol::SampleFormat::S24Be => Ok(SampleFormat::I24),
          protocol::SampleFormat::S32Le | protocol::SampleFormat::S32Be => Ok(SampleFormat::I32),
          protocol::SampleFormat::Float32Le | protocol::SampleFormat::Float32Be => {
              Ok(SampleFormat::F32)
          _ => Err(()),
42impl TryFrom<SampleFormat> for protocol::SampleFormat {
  type Error = ();
  fn try_from(format: SampleFormat) -> Result<Self, Self::Error> {
      match (format, cfg!(target_endian = "little")) {
          (SampleFormat::U8, _) => Ok(protocol::SampleFormat::U8),
          (SampleFormat::I16, true) => Ok(protocol::SampleFormat::S16Le),
          (SampleFormat::I16, false) => Ok(protocol::SampleFormat::S16Be),
          (SampleFormat::I24, true) => Ok(protocol::SampleFormat::S24Le),
          (SampleFormat::I24, false) => Ok(protocol::SampleFormat::S24Be),
          (SampleFormat::I32, true) => Ok(protocol::SampleFormat::S32Le),
          (SampleFormat::I32, false) => Ok(protocol::SampleFormat::S32Be),
          (SampleFormat::F32, true) => Ok(protocol::SampleFormat::Float32Le),
          (SampleFormat::F32, false) => Ok(protocol::SampleFormat::Float32Be),
          _ => Err(()),
61impl From<pulseaudio::ClientError> for BackendSpecificError {
  fn from(err: pulseaudio::ClientError) -> Self {
      BackendSpecificError {
          description: err.to_string(),
69/// A Host for connecting to the popular PulseAudio and PipeWire (via
70/// pipewire-pulse) audio servers on linux.
71pub struct Host {
  client: pulseaudio::Client,
75impl Host {
  pub fn new() -> Result<Self, HostUnavailable> {
      let client =
          pulseaudio::Client::from_env(c"cpal-pulseaudio").map_err(|_| HostUnavailable)?;
      Ok(Self { client })
84impl HostTrait for Host {
  type Devices = std::vec::IntoIter<Device>;
  type Device = Device;
  fn is_available() -> bool {
      pulseaudio::socket_path_from_env().is_some()
  fn devices(&self) -> Result<Self::Devices, DevicesError> {
      let sinks = block_on(self.client.list_sinks()).map_err(|err| BackendSpecificError {
          description: format!("Failed to list sinks: {err}"),
      })?;
      let sources = block_on(self.client.list_sources()).map_err(|err| BackendSpecificError {
          description: format!("Failed to list sources: {err}"),
      })?;
      Ok(sinks
          .into_iter()
          .map(|sink_info| Device::Sink {
              client: self.client.clone(),
              info: sink_info,
          .chain(sources.into_iter().map(|source_info| Device::Source {
              client: self.client.clone(),
              info: source_info,
          .collect::<Vec<_>>()
          .into_iter())
  fn default_input_device(&self) -> Option<Self::Device> {
      let source_info = block_on(
          self.client
              .source_info_by_name(protocol::DEFAULT_SOURCE.to_owned()),
      .ok()?;
      Some(Device::Source {
          client: self.client.clone(),
          info: source_info,
  fn default_output_device(&self) -> Option<Self::Device> {
      let sink_info = block_on(
          self.client
              .sink_info_by_name(protocol::DEFAULT_SINK.to_owned()),
      .ok()?;
      Some(Device::Sink {
          client: self.client.clone(),
          info: sink_info,
142/// A PulseAudio sink or source.
143#[derive(Debug, Clone)]
144pub enum Device {
145
      client: pulseaudio::Client,
      info: protocol::SinkInfo,
  Source {
      client: pulseaudio::Client,
      info: protocol::SourceInfo,
155fn supported_config_ranges() -> Vec<SupportedStreamConfigRange> {
  let mut ranges = vec![];
  for format in PULSE_FORMATS {
      for channel_count in 1..protocol::sample_spec::MAX_CHANNELS {
          let bytes_per_frame = channel_count as usize * format.sample_size();
          let max_frames = (protocol::MAX_MEMBLOCKQ_LENGTH / bytes_per_frame) as FrameCount;
          ranges.push(SupportedStreamConfigRange {
              channels: channel_count as _,
              min_sample_rate: 1,
              max_sample_rate: protocol::sample_spec::MAX_RATE,
              buffer_size: SupportedBufferSize::Range {
                  max: max_frames,
              sample_format: *format,
173
176fn default_config_from_spec(
  sample_spec: &protocol::SampleSpec,
  channel_map: &protocol::ChannelMap,
179) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
  let sample_format: SampleFormat = sample_spec
      .format
      .try_into()
      .map_err(|_| DefaultStreamConfigError::StreamTypeNotSupported)?;
  let bytes_per_frame = channel_map.num_channels() as usize * sample_format.sample_size();
  let max_frames = (protocol::MAX_MEMBLOCKQ_LENGTH / bytes_per_frame) as u32;
  Ok(SupportedStreamConfig {
      channels: channel_map.num_channels() as _,
      sample_rate: sample_spec.sample_rate,
      buffer_size: SupportedBufferSize::Range {
          min: 0,
          max: max_frames,
      sample_format,
197impl DeviceTrait for Device {
  type SupportedInputConfigs = std::vec::IntoIter<SupportedStreamConfigRange>;
  type SupportedOutputConfigs = std::vec::IntoIter<SupportedStreamConfigRange>;
  type Stream = Stream;
  fn name(&self) -> Result<String, DeviceNameError> {
      let name = match self {
          Device::Sink { info, .. } => &info.name,
          Device::Source { info, .. } => &info.name,
      Ok(String::from_utf8_lossy(name.as_bytes()).into_owned())
  fn supported_input_configs(
      &self,
  ) -> Result<Self::SupportedInputConfigs, SupportedStreamConfigsError> {
      let Device::Source { .. } = self else {
          return Ok(vec![].into_iter());
      Ok(supported_config_ranges().into_iter())
  fn supported_output_configs(
      &self,
  ) -> Result<Self::SupportedOutputConfigs, SupportedStreamConfigsError> {
      let Device::Sink { .. } = self else {
          return Ok(vec![].into_iter());
      Ok(supported_config_ranges().into_iter())
  fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
      let Device::Source { info, .. } = self else {
          return Err(DefaultStreamConfigError::StreamTypeNotSupported);
      default_config_from_spec(&info.sample_spec, &info.channel_map)
  fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
      let Device::Sink { info, .. } = self else {
          return Err(DefaultStreamConfigError::StreamTypeNotSupported);
      default_config_from_spec(&info.sample_spec, &info.channel_map)
  fn build_input_stream_raw<D, E>(
      &self,
      config: StreamConfig,
      sample_format: SampleFormat,
      data_callback: D,
      error_callback: E,
      _timeout: Option<std::time::Duration>,
  ) -> Result<Self::Stream, BuildStreamError>
251
      D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
      E: FnMut(StreamError) + Send + 'static,
      let Device::Source { client, info } = self else {
          return Err(BuildStreamError::StreamConfigNotSupported);
      let format: protocol::SampleFormat = sample_format
          .try_into()
          .map_err(|_| BuildStreamError::StreamConfigNotSupported)?;
      let sample_spec = make_sample_spec(config, format);
      let channel_map = make_channel_map(config);
      let buffer_attr = make_record_buffer_attr(config, format);
      let adjust_latency = matches!(config.buffer_size, crate::BufferSize::Fixed(_));
      let params = protocol::RecordStreamParams {
          sample_spec,
          channel_map,
          source_index: Some(info.index),
          buffer_attr,
          flags: protocol::stream::StreamFlags {
              // Start the stream suspended.
              start_corked: true,
              // When a fixed buffer size is requested, ask PA to configure
              // the source hardware to hit the requested latency end-to-end.
              adjust_latency,
              ..Default::default()
          ..Default::default()
      stream::Stream::new_record(client.clone(), params, data_callback, error_callback)
  fn build_output_stream_raw<D, E>(
      &self,
      config: StreamConfig,
      sample_format: SampleFormat,
      data_callback: D,
      error_callback: E,
      _timeout: Option<std::time::Duration>,
  ) -> Result<Self::Stream, BuildStreamError>
295
      D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
      E: FnMut(StreamError) + Send + 'static,
      let Device::Sink { client, info } = self else {
          return Err(BuildStreamError::StreamConfigNotSupported);
      let format: protocol::SampleFormat = sample_format
          .try_into()
          .map_err(|_| BuildStreamError::StreamConfigNotSupported)?;
      let sample_spec = make_sample_spec(config, format);
      let channel_map = make_channel_map(config);
      let buffer_attr = make_playback_buffer_attr(config, format);
      let adjust_latency = matches!(config.buffer_size, crate::BufferSize::Fixed(_));
      let params = protocol::PlaybackStreamParams {
          sink_index: Some(info.index),
          sample_spec,
          channel_map,
          buffer_attr,
          flags: protocol::stream::StreamFlags {
              // Start the stream suspended.
              start_corked: true,
              // When a fixed buffer size is requested, ask PA to configure
              // the sink hardware to hit the requested latency end-to-end.
              adjust_latency,
              ..Default::default()
          ..Default::default()
      stream::Stream::new_playback(client.clone(), params, data_callback, error_callback)
  fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
      let (name, description, direction) = match self {
          Device::Sink { info, .. } => (&info.name, &info.description, DeviceDirection::Output),
          Device::Source { info, .. } => (&info.name, &info.description, DeviceDirection::Input),
      let mut builder = DeviceDescriptionBuilder::new(String::from_utf8_lossy(name.as_bytes()))
          .direction(direction);
      if let Some(desc) = description {
          builder = builder.add_extended_line(String::from_utf8_lossy(desc.as_bytes()));
      Ok(builder.build())
  fn id(&self) -> Result<DeviceId, DeviceIdError> {
      let id = match self {
          Device::Sink { info, .. } => info.index,
          Device::Source { info, .. } => info.index,
      Ok(DeviceId(HostId::PulseAudio, id.to_string()))
356fn make_sample_spec(config: StreamConfig, format: protocol::SampleFormat) -> protocol::SampleSpec {
  protocol::SampleSpec {
      format,
      sample_rate: config.sample_rate,
      channels: config.channels as _,
364fn make_channel_map(config: StreamConfig) -> protocol::ChannelMap {
  use protocol::ChannelPosition::*;
  // Standard channel layouts following the PulseAudio default channel map
  // (PA_CHANNEL_MAP_DEFAULT) for 1-8 channels, and common Atmos height-
  // channel conventions for 10 and 12 channels. Counts without a widely
  // agreed layout (9, 11, >12) fall back to sequential Aux positions.
  let standard: &[protocol::ChannelPosition] = match config.channels {
      1 => &[Mono],
      2 => &[FrontLeft, FrontRight],
      3 => &[FrontLeft, FrontRight, FrontCenter],
      4 => &[FrontLeft, FrontRight, RearLeft, RearRight],
      5 => &[FrontLeft, FrontRight, FrontCenter, RearLeft, RearRight],
      6 => &[FrontLeft, FrontRight, FrontCenter, Lfe, RearLeft, RearRight],
      7 => &[
          FrontLeft,
          FrontRight,
          FrontCenter,
          Lfe,
          RearLeft,
          RearRight,
          RearCenter,
      8 => &[
          FrontLeft,
          FrontRight,
          FrontCenter,
          Lfe,
          RearLeft,
          RearRight,
          SideLeft,
          SideRight,
      // 7.1.2 (Dolby Atmos): 7.1 + top-front L/R
      10 => &[
          FrontLeft,
          FrontRight,
          FrontCenter,
          Lfe,
          RearLeft,
          RearRight,
          SideLeft,
          SideRight,
          TopFrontLeft,
          TopFrontRight,
      // 7.1.4 (Dolby Atmos): 7.1 + top-front L/R + top-rear L/R
      12 => &[
          FrontLeft,
          FrontRight,
          FrontCenter,
          Lfe,
          RearLeft,
          RearRight,
          SideLeft,
          SideRight,
          TopFrontLeft,
          TopFrontRight,
          TopRearLeft,
          TopRearRight,
      _ => &[],
  if !standard.is_empty() {
      return protocol::ChannelMap::new(standard.iter().copied());
  let aux = [
      Aux0, Aux1, Aux2, Aux3, Aux4, Aux5, Aux6, Aux7, Aux8, Aux9, Aux10, Aux11, Aux12, Aux13,
      Aux14, Aux15, Aux16, Aux17, Aux18, Aux19, Aux20, Aux21, Aux22, Aux23, Aux24, Aux25, Aux26,
      Aux27, Aux28, Aux29, Aux30, Aux31,
  protocol::ChannelMap::new(aux.iter().copied().take(config.channels as usize))
440fn make_playback_buffer_attr(
  config: StreamConfig,
  format: protocol::SampleFormat,
443) -> protocol::stream::BufferAttr {
  match config.buffer_size {
      crate::BufferSize::Default => Default::default(),
      crate::BufferSize::Fixed(frame_count) => {
          let len = frame_count * config.channels as u32 * format.bytes_per_sample() as u32;
          protocol::stream::BufferAttr {
              // Double-buffer: total buffer = 2 callback periods. With
              // adjust_latency this becomes the end-to-end latency target,
              // Minimum request = one callback period, ensuring the server
              // always asks for exactly frame_count frames per call.
              max_length: 2 * len,
              target_length: 2 * len,
              minimum_request_length: len,
              ..Default::default()
462fn make_record_buffer_attr(
  config: StreamConfig,
  format: protocol::SampleFormat,
465) -> protocol::stream::BufferAttr {
  match config.buffer_size {
      crate::BufferSize::Default => Default::default(),
      crate::BufferSize::Fixed(frame_count) => {
          let len = frame_count * config.channels as u32 * format.bytes_per_sample() as u32;
          protocol::stream::BufferAttr {
              // fragment_size controls the delivery chunk size for record
              // streams; target_length is playback-only and is ignored here.
              max_length: len,
              fragment_size: len,
              ..Default::default()

Lift freeq's AV media plane out of sleek 90f8b89 nandi 19d ago	1	use futures::executor::block_on;
	2	use pulseaudio::protocol;
	3
	4	mod stream;
	5
	6	pub use stream::Stream;
	7
	8	use crate::{
	9	traits::{DeviceTrait, HostTrait},
	10	BackendSpecificError, BuildStreamError, Data, DefaultStreamConfigError, DeviceDescription,
	11	DeviceDescriptionBuilder, DeviceDirection, DeviceId, DeviceIdError, DeviceNameError,
	12	DevicesError, FrameCount, HostId, HostUnavailable, InputCallbackInfo, OutputCallbackInfo,
	13	SampleFormat, StreamConfig, StreamError, SupportedBufferSize, SupportedStreamConfig,
	14	SupportedStreamConfigRange, SupportedStreamConfigsError,
	15	};
	16
	17	const PULSE_FORMATS: &[SampleFormat] = &[
	18	SampleFormat::U8,
	19	SampleFormat::I16,
	20	SampleFormat::I24,
	21	SampleFormat::I32,
	22	SampleFormat::F32,
	23	];
	24
	25	impl TryFrom<protocol::SampleFormat> for SampleFormat {
	26	type Error = ();
	27
	28	fn try_from(spec: protocol::SampleFormat) -> Result<Self, Self::Error> {
	29	match spec {
	30	protocol::SampleFormat::U8 => Ok(SampleFormat::U8),
	31	protocol::SampleFormat::S16Le \| protocol::SampleFormat::S16Be => Ok(SampleFormat::I16),
	32	protocol::SampleFormat::S24Le \| protocol::SampleFormat::S24Be => Ok(SampleFormat::I24),
	33	protocol::SampleFormat::S32Le \| protocol::SampleFormat::S32Be => Ok(SampleFormat::I32),
	34	protocol::SampleFormat::Float32Le \| protocol::SampleFormat::Float32Be => {
	35	Ok(SampleFormat::F32)
	36	}
	37	_ => Err(()),
	38	}
	39	}
	40	}
	41
	42	impl TryFrom<SampleFormat> for protocol::SampleFormat {
	43	type Error = ();
	44
	45	fn try_from(format: SampleFormat) -> Result<Self, Self::Error> {
	46	match (format, cfg!(target_endian = "little")) {
	47	(SampleFormat::U8, _) => Ok(protocol::SampleFormat::U8),
	48	(SampleFormat::I16, true) => Ok(protocol::SampleFormat::S16Le),
	49	(SampleFormat::I16, false) => Ok(protocol::SampleFormat::S16Be),
	50	(SampleFormat::I24, true) => Ok(protocol::SampleFormat::S24Le),
	51	(SampleFormat::I24, false) => Ok(protocol::SampleFormat::S24Be),
	52	(SampleFormat::I32, true) => Ok(protocol::SampleFormat::S32Le),
	53	(SampleFormat::I32, false) => Ok(protocol::SampleFormat::S32Be),
	54	(SampleFormat::F32, true) => Ok(protocol::SampleFormat::Float32Le),
	55	(SampleFormat::F32, false) => Ok(protocol::SampleFormat::Float32Be),
	56	_ => Err(()),
	57	}
	58	}
	59	}
	60
	61	impl From<pulseaudio::ClientError> for BackendSpecificError {
	62	fn from(err: pulseaudio::ClientError) -> Self {
	63	BackendSpecificError {
	64	description: err.to_string(),
65	}
66	}
67	}
68
69	/// A Host for connecting to the popular PulseAudio and PipeWire (via
70	/// pipewire-pulse) audio servers on linux.
71	pub struct Host {
72	client: pulseaudio::Client,
73	}
74
75	impl Host {
76	pub fn new() -> Result<Self, HostUnavailable> {
77	let client =
78	pulseaudio::Client::from_env(c"cpal-pulseaudio").map_err(\|_\| HostUnavailable)?;
79
80	Ok(Self { client })
81	}
82	}
83
84	impl HostTrait for Host {
85	type Devices = std::vec::IntoIter<Device>;
86	type Device = Device;
87
88	fn is_available() -> bool {
89	pulseaudio::socket_path_from_env().is_some()
90	}
91
92	fn devices(&self) -> Result<Self::Devices, DevicesError> {
93	let sinks = block_on(self.client.list_sinks()).map_err(\|err\| BackendSpecificError {
94	description: format!("Failed to list sinks: {err}"),
95	})?;
96
97	let sources = block_on(self.client.list_sources()).map_err(\|err\| BackendSpecificError {
98	description: format!("Failed to list sources: {err}"),
99	})?;
100
101	Ok(sinks
102	.into_iter()
103	.map(\|sink_info\| Device::Sink {
104	client: self.client.clone(),
105	info: sink_info,
106	})
107	.chain(sources.into_iter().map(\|source_info\| Device::Source {
108	client: self.client.clone(),
109	info: source_info,
110	}))
111	.collect::<Vec<_>>()
112	.into_iter())
113	}
114
115	fn default_input_device(&self) -> Option<Self::Device> {
116	let source_info = block_on(
117	self.client
118	.source_info_by_name(protocol::DEFAULT_SOURCE.to_owned()),
119	)
120	.ok()?;
121
122	Some(Device::Source {
123	client: self.client.clone(),
124	info: source_info,
125	})
126	}
127
128	fn default_output_device(&self) -> Option<Self::Device> {
129	let sink_info = block_on(
130	self.client
131	.sink_info_by_name(protocol::DEFAULT_SINK.to_owned()),
132	)
133	.ok()?;
134
135	Some(Device::Sink {
136	client: self.client.clone(),
137	info: sink_info,
138	})
139	}
140	}
141
142	/// A PulseAudio sink or source.
143	#[derive(Debug, Clone)]
144	pub enum Device {
145	Sink {
146	client: pulseaudio::Client,
147	info: protocol::SinkInfo,
148	},
149	Source {
150	client: pulseaudio::Client,
151	info: protocol::SourceInfo,
152	},
153	}
154
155	fn supported_config_ranges() -> Vec<SupportedStreamConfigRange> {
156	let mut ranges = vec![];
157	for format in PULSE_FORMATS {
158	for channel_count in 1..protocol::sample_spec::MAX_CHANNELS {
159	let bytes_per_frame = channel_count as usize * format.sample_size();
160	let max_frames = (protocol::MAX_MEMBLOCKQ_LENGTH / bytes_per_frame) as FrameCount;
161	ranges.push(SupportedStreamConfigRange {
162	channels: channel_count as _,
163	min_sample_rate: 1,
164	max_sample_rate: protocol::sample_spec::MAX_RATE,
165	buffer_size: SupportedBufferSize::Range {
166	min: 0,
167	max: max_frames,
168	},
169	sample_format: *format,
170	});
171	}
172	}
173	ranges
174	}
175
176	fn default_config_from_spec(
177	sample_spec: &protocol::SampleSpec,
178	channel_map: &protocol::ChannelMap,
179	) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
180	let sample_format: SampleFormat = sample_spec
181	.format
182	.try_into()
183	.map_err(\|_\| DefaultStreamConfigError::StreamTypeNotSupported)?;
184	let bytes_per_frame = channel_map.num_channels() as usize * sample_format.sample_size();
185	let max_frames = (protocol::MAX_MEMBLOCKQ_LENGTH / bytes_per_frame) as u32;
186	Ok(SupportedStreamConfig {
187	channels: channel_map.num_channels() as _,
188	sample_rate: sample_spec.sample_rate,
189	buffer_size: SupportedBufferSize::Range {
190	min: 0,
191	max: max_frames,
192	},
193	sample_format,
194	})
195	}
196
197	impl DeviceTrait for Device {
198	type SupportedInputConfigs = std::vec::IntoIter<SupportedStreamConfigRange>;
199	type SupportedOutputConfigs = std::vec::IntoIter<SupportedStreamConfigRange>;
200	type Stream = Stream;
201
202	fn name(&self) -> Result<String, DeviceNameError> {
203	let name = match self {
204	Device::Sink { info, .. } => &info.name,
205	Device::Source { info, .. } => &info.name,
206	};
207
208	Ok(String::from_utf8_lossy(name.as_bytes()).into_owned())
209	}
210
211	fn supported_input_configs(
212	&self,
213	) -> Result<Self::SupportedInputConfigs, SupportedStreamConfigsError> {
214	let Device::Source { .. } = self else {
215	return Ok(vec![].into_iter());
216	};
217	Ok(supported_config_ranges().into_iter())
218	}
219
220	fn supported_output_configs(
221	&self,
222	) -> Result<Self::SupportedOutputConfigs, SupportedStreamConfigsError> {
223	let Device::Sink { .. } = self else {
224	return Ok(vec![].into_iter());
225	};
226	Ok(supported_config_ranges().into_iter())
227	}
228
229	fn default_input_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
230	let Device::Source { info, .. } = self else {
231	return Err(DefaultStreamConfigError::StreamTypeNotSupported);
232	};
233	default_config_from_spec(&info.sample_spec, &info.channel_map)
234	}
235
236	fn default_output_config(&self) -> Result<SupportedStreamConfig, DefaultStreamConfigError> {
237	let Device::Sink { info, .. } = self else {
238	return Err(DefaultStreamConfigError::StreamTypeNotSupported);
239	};
240	default_config_from_spec(&info.sample_spec, &info.channel_map)
241	}
242
243	fn build_input_stream_raw<D, E>(
244	&self,
245	config: StreamConfig,
246	sample_format: SampleFormat,
247	data_callback: D,
248	error_callback: E,
249	_timeout: Option<std::time::Duration>,
250	) -> Result<Self::Stream, BuildStreamError>
251	where
252	D: FnMut(&Data, &InputCallbackInfo) + Send + 'static,
253	E: FnMut(StreamError) + Send + 'static,
254	{
255	let Device::Source { client, info } = self else {
256	return Err(BuildStreamError::StreamConfigNotSupported);
257	};
258
259	let format: protocol::SampleFormat = sample_format
260	.try_into()
261	.map_err(\|_\| BuildStreamError::StreamConfigNotSupported)?;
262
263	let sample_spec = make_sample_spec(config, format);
264	let channel_map = make_channel_map(config);
265	let buffer_attr = make_record_buffer_attr(config, format);
266	let adjust_latency = matches!(config.buffer_size, crate::BufferSize::Fixed(_));
267
268	let params = protocol::RecordStreamParams {
269	sample_spec,
270	channel_map,
271	source_index: Some(info.index),
272	buffer_attr,
273	flags: protocol::stream::StreamFlags {
274	// Start the stream suspended.
275	start_corked: true,
276	// When a fixed buffer size is requested, ask PA to configure
277	// the source hardware to hit the requested latency end-to-end.
278	adjust_latency,
279	..Default::default()
280	},
281	..Default::default()
282	};
283
284	stream::Stream::new_record(client.clone(), params, data_callback, error_callback)
285	}
286
287	fn build_output_stream_raw<D, E>(
288	&self,
289	config: StreamConfig,
290	sample_format: SampleFormat,
291	data_callback: D,
292	error_callback: E,
293	_timeout: Option<std::time::Duration>,
294	) -> Result<Self::Stream, BuildStreamError>
295	where
296	D: FnMut(&mut Data, &OutputCallbackInfo) + Send + 'static,
297	E: FnMut(StreamError) + Send + 'static,
298	{
299	let Device::Sink { client, info } = self else {
300	return Err(BuildStreamError::StreamConfigNotSupported);
301	};
302
303	let format: protocol::SampleFormat = sample_format
304	.try_into()
305	.map_err(\|_\| BuildStreamError::StreamConfigNotSupported)?;
306
307	let sample_spec = make_sample_spec(config, format);
308	let channel_map = make_channel_map(config);
309	let buffer_attr = make_playback_buffer_attr(config, format);
310	let adjust_latency = matches!(config.buffer_size, crate::BufferSize::Fixed(_));
311
312	let params = protocol::PlaybackStreamParams {
313	sink_index: Some(info.index),
314	sample_spec,
315	channel_map,
316	buffer_attr,
317	flags: protocol::stream::StreamFlags {
318	// Start the stream suspended.
319	start_corked: true,
320	// When a fixed buffer size is requested, ask PA to configure
321	// the sink hardware to hit the requested latency end-to-end.
322	adjust_latency,
323	..Default::default()
324	},
325	..Default::default()
326	};
327
328	stream::Stream::new_playback(client.clone(), params, data_callback, error_callback)
329	}
330
331	fn description(&self) -> Result<DeviceDescription, DeviceNameError> {
332	let (name, description, direction) = match self {
333	Device::Sink { info, .. } => (&info.name, &info.description, DeviceDirection::Output),
334	Device::Source { info, .. } => (&info.name, &info.description, DeviceDirection::Input),
335	};
336
337	let mut builder = DeviceDescriptionBuilder::new(String::from_utf8_lossy(name.as_bytes()))
338	.direction(direction);
339	if let Some(desc) = description {
340	builder = builder.add_extended_line(String::from_utf8_lossy(desc.as_bytes()));
341	}
342
343	Ok(builder.build())
344	}
345
346	fn id(&self) -> Result<DeviceId, DeviceIdError> {
347	let id = match self {
348	Device::Sink { info, .. } => info.index,
349	Device::Source { info, .. } => info.index,
350	};
351
352	Ok(DeviceId(HostId::PulseAudio, id.to_string()))
353	}
354	}
355
356	fn make_sample_spec(config: StreamConfig, format: protocol::SampleFormat) -> protocol::SampleSpec {
357	protocol::SampleSpec {
358	format,
359	sample_rate: config.sample_rate,
360	channels: config.channels as _,
361	}
362	}
363
364	fn make_channel_map(config: StreamConfig) -> protocol::ChannelMap {
365	use protocol::ChannelPosition::*;
366
367	// Standard channel layouts following the PulseAudio default channel map
368	// (PA_CHANNEL_MAP_DEFAULT) for 1-8 channels, and common Atmos height-
369	// channel conventions for 10 and 12 channels. Counts without a widely
370	// agreed layout (9, 11, >12) fall back to sequential Aux positions.
371	let standard: &[protocol::ChannelPosition] = match config.channels {
372	1 => &[Mono],
373	2 => &[FrontLeft, FrontRight],
374	3 => &[FrontLeft, FrontRight, FrontCenter],
375	4 => &[FrontLeft, FrontRight, RearLeft, RearRight],
376	5 => &[FrontLeft, FrontRight, FrontCenter, RearLeft, RearRight],
377	6 => &[FrontLeft, FrontRight, FrontCenter, Lfe, RearLeft, RearRight],
378	7 => &[
379	FrontLeft,
380	FrontRight,
381	FrontCenter,
382	Lfe,
383	RearLeft,
384	RearRight,
385	RearCenter,
386	],
387	8 => &[
388	FrontLeft,
389	FrontRight,
390	FrontCenter,
391	Lfe,
392	RearLeft,
393	RearRight,
394	SideLeft,
395	SideRight,
396	],
397	// 7.1.2 (Dolby Atmos): 7.1 + top-front L/R
398	10 => &[
399	FrontLeft,
400	FrontRight,
401	FrontCenter,
402	Lfe,
403	RearLeft,
404	RearRight,
405	SideLeft,
406	SideRight,
407	TopFrontLeft,
408	TopFrontRight,
409	],
410	// 7.1.4 (Dolby Atmos): 7.1 + top-front L/R + top-rear L/R
411	12 => &[
412	FrontLeft,
413	FrontRight,
414	FrontCenter,
415	Lfe,
416	RearLeft,
417	RearRight,
418	SideLeft,
419	SideRight,
420	TopFrontLeft,
421	TopFrontRight,
422	TopRearLeft,
423	TopRearRight,
424	],
425	_ => &[],
426	};
427
428	if !standard.is_empty() {
429	return protocol::ChannelMap::new(standard.iter().copied());
430	}
431
432	let aux = [
433	Aux0, Aux1, Aux2, Aux3, Aux4, Aux5, Aux6, Aux7, Aux8, Aux9, Aux10, Aux11, Aux12, Aux13,
434	Aux14, Aux15, Aux16, Aux17, Aux18, Aux19, Aux20, Aux21, Aux22, Aux23, Aux24, Aux25, Aux26,
435	Aux27, Aux28, Aux29, Aux30, Aux31,
436	];
437	protocol::ChannelMap::new(aux.iter().copied().take(config.channels as usize))
438	}
439
440	fn make_playback_buffer_attr(
441	config: StreamConfig,
442	format: protocol::SampleFormat,
443	) -> protocol::stream::BufferAttr {
444	match config.buffer_size {
445	crate::BufferSize::Default => Default::default(),
446	crate::BufferSize::Fixed(frame_count) => {
447	let len = frame_count * config.channels as u32 * format.bytes_per_sample() as u32;
448	protocol::stream::BufferAttr {
449	// Double-buffer: total buffer = 2 callback periods. With
450	// adjust_latency this becomes the end-to-end latency target,
451	// Minimum request = one callback period, ensuring the server
452	// always asks for exactly frame_count frames per call.
453	max_length: 2 * len,
454	target_length: 2 * len,
455	minimum_request_length: len,
456	..Default::default()
457	}
458	}
459	}
460	}
461
462	fn make_record_buffer_attr(
463	config: StreamConfig,
464	format: protocol::SampleFormat,
465	) -> protocol::stream::BufferAttr {
466	match config.buffer_size {
467	crate::BufferSize::Default => Default::default(),
468	crate::BufferSize::Fixed(frame_count) => {
469	let len = frame_count * config.channels as u32 * format.bytes_per_sample() as u32;
470	protocol::stream::BufferAttr {
471	// fragment_size controls the delivery chunk size for record
472	// streams; target_length is playback-only and is ignored here.
473	max_length: len,
474	fragment_size: len,
475	..Default::default()
476	}
477	}
478	}
479	}