settings/audio/

audio_controller.rs

// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
use super::AudioInfoLoader;
use crate::audio::types::{
    AudioInfo, AudioStream, AudioStreamType, SetAudioStream, AUDIO_STREAM_TYPE_COUNT,
};
use crate::audio::{create_default_modified_counters, ModifiedCounters, StreamVolumeControl};
use crate::base::SettingType;
use crate::handler::base::Request;
use crate::handler::setting_handler::persist::{controller as data_controller, ClientProxy};
use crate::handler::setting_handler::{
    controller, ControllerError, ControllerStateResult, Event, SettingHandlerResult, State,
};
use crate::{trace, trace_guard};
use async_trait::async_trait;
use futures::lock::Mutex;
use settings_storage::device_storage::{DeviceStorage, DeviceStorageCompatible};
use settings_storage::storage_factory::{DefaultLoader, StorageAccess, StorageFactory};
use settings_storage::UpdateState;
use std::collections::HashMap;
use std::marker::PhantomData;
use std::rc::Rc;
use {fuchsia_async as fasync, fuchsia_trace as ftrace};

type VolumeControllerHandle = Rc<Mutex<VolumeController>>;

pub(crate) struct VolumeController {
    client: ClientProxy,
    store: Rc<DeviceStorage>,
    audio_service_connected: bool,
    stream_volume_controls: HashMap<AudioStreamType, StreamVolumeControl>,
    modified_counters: ModifiedCounters,
    audio_info_loader: AudioInfoLoader,
}

enum UpdateFrom {
    AudioInfo(AudioInfo),
    NewStreams(Vec<SetAudioStream>),
}

impl VolumeController {
    async fn create_with(
        client: ClientProxy,
        audio_info_loader: AudioInfoLoader,
        storage_factory: Rc<impl StorageFactory<Storage = DeviceStorage>>,
    ) -> VolumeControllerHandle {
        let store = storage_factory.get_store().await;
        Rc::new(Mutex::new(Self {
            client,
            store,
            stream_volume_controls: HashMap::new(),
            audio_service_connected: false,
            modified_counters: create_default_modified_counters(),
            audio_info_loader,
        }))
    }

    /// Restores the necessary dependencies' state on boot.
    async fn restore(&mut self, id: ftrace::Id) -> ControllerStateResult {
        self.restore_volume_state(true, id).await
    }

    /// Extracts the audio state from persistent storage and restores it on
    /// the local state. Also pushes the changes to the audio core if
    /// `push_to_audio_core` is true.
    async fn restore_volume_state(
        &mut self,
        push_to_audio_core: bool,
        id: ftrace::Id,
    ) -> ControllerStateResult {
        let audio_info = self.store.get::<AudioInfo>().await;
        // Ignore the notification triggered result.
        let _ = self
            .update_volume_streams(UpdateFrom::AudioInfo(audio_info), push_to_audio_core, id)
            .await?;
        Ok(())
    }

    async fn get_info(&self) -> Result<AudioInfo, ControllerError> {
        let mut audio_info = self.store.get::<AudioInfo>().await;

        audio_info.modified_counters = Some(self.modified_counters.clone());
        Ok(audio_info)
    }

    async fn set_volume(
        &mut self,
        volume: Vec<SetAudioStream>,
        id: ftrace::Id,
    ) -> SettingHandlerResult {
        let guard = trace_guard!(id, c"set volume updating counters");
        // Update counters for changed streams.
        for stream in &volume {
            // We don't care what the value of the counter is, just that it is different from the
            // previous value. We use wrapping_add to avoid eventual overflow of the counter.
            let _ = self.modified_counters.insert(
                stream.stream_type,
                self.modified_counters
                    .get(&stream.stream_type)
                    .map_or(0, |flag| flag.wrapping_add(1)),
            );
        }
        drop(guard);

        if !(self.update_volume_streams(UpdateFrom::NewStreams(volume), true, id).await?) {
            trace!(id, c"set volume notifying");
            let info = self.get_info().await?.into();
            self.client.notify(Event::Changed(info)).await;
        }

        Ok(None)
    }

    async fn get_streams_array_from_map(
        &self,
        stream_map: &HashMap<AudioStreamType, StreamVolumeControl>,
    ) -> [AudioStream; AUDIO_STREAM_TYPE_COUNT] {
        let mut streams: [AudioStream; AUDIO_STREAM_TYPE_COUNT] =
            self.audio_info_loader.default_value().streams;
        for stream in &mut streams {
            if let Some(volume_control) = stream_map.get(&stream.stream_type) {
                *stream = volume_control.stored_stream;
            }
        }

        streams
    }

    /// Updates the state with the given streams' volume levels.
    ///
    /// If `push_to_audio_core` is true, pushes the changes to the audio core.
    /// If not, just sets it on the local stored state. Should be called with
    /// true on first restore and on volume changes, and false otherwise.
    /// Returns whether the change triggered a notification.
    async fn update_volume_streams(
        &mut self,
        update_from: UpdateFrom,
        push_to_audio_core: bool,
        id: ftrace::Id,
    ) -> Result<bool, ControllerError> {
        let mut new_vec = vec![];
        trace!(id, c"update volume streams");
        let calculating_guard = trace_guard!(id, c"check and bind");
        let (stored_value, new_streams) = match &update_from {
            UpdateFrom::AudioInfo(audio_info) => (None, audio_info.streams.iter()),
            UpdateFrom::NewStreams(streams) => {
                trace!(id, c"reading setting");
                let stored_value = self.store.get::<AudioInfo>().await;
                for set_stream in streams.iter() {
                    let stored_stream = stored_value
                        .streams
                        .iter()
                        .find(|stream| stream.stream_type == set_stream.stream_type)
                        .ok_or_else(|| {
                            ControllerError::InvalidArgument(
                                SettingType::Audio,
                                "stream".into(),
                                format!("{set_stream:?}").into(),
                            )
                        })?;
                    new_vec.push(AudioStream {
                        stream_type: stored_stream.stream_type,
                        source: set_stream.source,
                        user_volume_level: set_stream
                            .user_volume_level
                            .unwrap_or(stored_stream.user_volume_level),
                        user_volume_muted: set_stream
                            .user_volume_muted
                            .unwrap_or(stored_stream.user_volume_muted),
                    });
                }
                (Some(stored_value), new_vec.iter())
            }
        };

        if push_to_audio_core {
            let guard = trace_guard!(id, c"push to core");
            self.check_and_bind_volume_controls(
                id,
                self.audio_info_loader.default_value().streams.iter(),
            )
            .await?;
            drop(guard);

            trace!(id, c"setting core");
            for stream in new_streams {
                if let Some(volume_control) =
                    self.stream_volume_controls.get_mut(&stream.stream_type)
                {
                    volume_control.set_volume(id, *stream).await?;
                }
            }
        } else {
            trace!(id, c"without push to core");
            self.check_and_bind_volume_controls(id, new_streams).await?;
        }
        drop(calculating_guard);

        if let Some(mut stored_value) = stored_value {
            let guard = trace_guard!(id, c"updating streams and counters");
            stored_value.streams =
                self.get_streams_array_from_map(&self.stream_volume_controls).await;
            stored_value.modified_counters = Some(self.modified_counters.clone());
            drop(guard);

            let guard = trace_guard!(id, c"writing setting");
            let write_result = self.client.storage_write(&self.store, stored_value, id).await;
            drop(guard);
            Ok(write_result
                .as_ref()
                .map_or(false, |update_state| UpdateState::Updated == *update_state))
        } else {
            Ok(false)
        }
    }

    /// Populates the local state with the given `streams` and binds it to the audio core service.
    async fn check_and_bind_volume_controls(
        &mut self,
        id: ftrace::Id,
        streams: impl Iterator<Item = &AudioStream>,
    ) -> ControllerStateResult {
        trace!(id, c"check and bind fn");
        if self.audio_service_connected {
            return Ok(());
        }

        let guard = trace_guard!(id, c"connecting to service");
        let service_result = self
            .client
            .get_service_context()
            .connect::<fidl_fuchsia_media::AudioCoreMarker>()
            .await;

        let audio_service = service_result.map_err(|e| {
            ControllerError::ExternalFailure(
                SettingType::Audio,
                "fuchsia.media.audio".into(),
                "connect for audio_core".into(),
                format!("{e:?}").into(),
            )
        })?;

        // The stream_volume_controls are generated in two steps instead of
        // one so that if one of the bindings fails during the first loop,
        // none of the streams are modified.
        drop(guard);
        let mut stream_tuples = Vec::new();
        for stream in streams {
            trace!(id, c"create stream volume control");
            let client = self.client.clone();

            // Generate a tuple with stream type and StreamVolumeControl.
            stream_tuples.push((
                stream.stream_type,
                StreamVolumeControl::create(
                    id,
                    &audio_service,
                    *stream,
                    Some(Rc::new(move || {
                        // When the StreamVolumeControl exits early, inform the
                        // proxy we have exited. The proxy will then cleanup this
                        // AudioController.
                        let client = client.clone();
                        fasync::Task::local(async move {
                            trace!(id, c"stream exit");
                            client
                                .notify(Event::Exited(Err(ControllerError::UnexpectedError(
                                    "stream_volume_control exit".into(),
                                ))))
                                .await;
                        })
                        .detach();
                    })),
                    None,
                )
                .await?,
            ));
        }

        stream_tuples.into_iter().for_each(|(stream_type, stream_volume_control)| {
            // Ignore the previous value, if any.
            let _ = self.stream_volume_controls.insert(stream_type, stream_volume_control);
        });
        self.audio_service_connected = true;

        Ok(())
    }
}

pub(crate) struct AudioController<F> {
    volume: VolumeControllerHandle,
    _phantom: PhantomData<F>,
}

impl<F> StorageAccess for AudioController<F> {
    type Storage = DeviceStorage;
    type Data = AudioInfo;
    const STORAGE_KEY: &'static str = AudioInfo::KEY;
}

#[async_trait(?Send)]
impl<F> data_controller::CreateWithAsync for AudioController<F>
where
    F: StorageFactory<Storage = DeviceStorage>,
{
    type Data = (Rc<F>, AudioInfoLoader);
    async fn create_with(client: ClientProxy, data: Self::Data) -> Result<Self, ControllerError> {
        Ok(AudioController {
            volume: VolumeController::create_with(client, data.1, data.0).await,
            _phantom: PhantomData,
        })
    }
}

#[async_trait(?Send)]
impl<F> controller::Handle for AudioController<F> {
    async fn handle(&self, request: Request) -> Option<SettingHandlerResult> {
        match request {
            Request::Restore => Some({
                let id = ftrace::Id::new();
                trace!(id, c"controller restore");
                self.volume.lock().await.restore(id).await.map(|_| None)
            }),
            Request::SetVolume(volume, id) => {
                trace!(id, c"controller set");
                // Validate volume contains valid volume level numbers.
                for audio_stream in &volume {
                    if !audio_stream.has_valid_volume_level() {
                        return Some(Err(ControllerError::InvalidArgument(
                            SettingType::Audio,
                            "stream".into(),
                            format!("{audio_stream:?}").into(),
                        )));
                    }
                }
                Some(self.volume.lock().await.set_volume(volume, id).await)
            }
            Request::Get => {
                Some(self.volume.lock().await.get_info().await.map(|info| Some(info.into())))
            }
            _ => None,
        }
    }

    async fn change_state(&mut self, state: State) -> Option<ControllerStateResult> {
        match state {
            State::Startup => {
                // Restore the volume state locally but do not push to the audio core.
                Some({
                    let id = ftrace::Id::new();
                    trace!(id, c"controller startup");
                    self.volume.lock().await.restore_volume_state(false, id).await
                })
            }
            _ => None,
        }
    }
}