// Copyright 2020 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 {
    crate::{
        descriptor::EventDescriptor,
        events::{EventSource, EventStream},
    },
    anyhow::Error,
    fuchsia_async as fasync,
    futures::{
        future::{AbortHandle, Abortable, TryFutureExt},
        lock::Mutex,
    },
    std::{convert::TryFrom, sync::Arc},
};

/// Records events from an EventStream, allowing them to be
/// flushed out into a vector at a later point in time.
pub struct EventLog {
    recorded_events: Arc<Mutex<Vec<EventDescriptor>>>,
    abort_handle: AbortHandle,
}

impl EventLog {
    /// Subscribe to the provided `event_names`, and log the `EventDescriptors`
    /// in a separate task.
    pub async fn record_events(
        event_source: &mut EventSource,
        event_names: Vec<impl AsRef<str>>,
    ) -> Result<EventLog, Error> {
        let event_stream = event_source.subscribe(event_names).await?;
        Ok(EventLog::new(event_stream))
    }

    fn new(mut event_stream: EventStream) -> Self {
        let (abort_handle, abort_registration) = AbortHandle::new_pair();
        let recorded_events = Arc::new(Mutex::new(vec![]));
        {
            // Start an async task that records events from the event_stream
            let recorded_events = recorded_events.clone();
            fasync::Task::spawn(
                Abortable::new(
                    async move {
                        loop {
                            // Get the next event from the event_stream
                            let event = event_stream
                                .next()
                                .await
                                .expect("Failed to get next event from EventStreamSync");

                            // Construct the EventDescriptor from the Event
                            let recorded_event = EventDescriptor::try_from(&event)
                                .expect("Failed to convert Event to EventDescriptor");

                            // Insert the event into the list
                            {
                                let mut recorded_events = recorded_events.lock().await;
                                recorded_events.push(recorded_event);
                            }
                        }
                    },
                    abort_registration,
                )
                .unwrap_or_else(|_| ()),
            )
            .detach();
        }
        Self { recorded_events, abort_handle }
    }

    pub async fn flush(&self) -> Vec<EventDescriptor> {
        // Lock and flush out all events from the vector
        let mut recorded_events = self.recorded_events.lock().await;
        recorded_events.drain(..).collect()
    }
}

impl Drop for EventLog {
    fn drop(&mut self) {
        self.abort_handle.abort();
    }
}