use crate::logs::{error::LogsError, listener::Listener, repository::LogsRepository};
use fidl::endpoints::DiscoverableProtocolMarker;
use fidl_fuchsia_diagnostics::StreamMode;
use fidl_fuchsia_logger as flogger;
use fuchsia_async as fasync;
use fuchsia_sync::Mutex;
use fuchsia_trace as ftrace;
use futures::{channel::mpsc, StreamExt};
use std::sync::Arc;
use tracing::warn;
pub struct LogServer {
logs_repo: Arc<LogsRepository>,
task_sender: Arc<Mutex<mpsc::UnboundedSender<fasync::Task<()>>>>,
drain_listeners_task: Mutex<Option<fasync::Task<()>>>,
}
impl LogServer {
pub fn new(logs_repo: Arc<LogsRepository>) -> Self {
let (task_sender, rcv) = mpsc::unbounded();
Self {
logs_repo,
task_sender: Arc::new(Mutex::new(task_sender)),
drain_listeners_task: Mutex::new(Some(fasync::Task::spawn(async move {
rcv.for_each_concurrent(None, |rx| rx).await;
}))),
}
}
pub fn spawn(&self, stream: flogger::LogRequestStream) {
let logs_repo = Arc::clone(&self.logs_repo);
let sender = Arc::clone(&self.task_sender);
if let Err(e) = self.task_sender.lock().unbounded_send(fasync::Task::spawn(async move {
if let Err(e) = Self::handle_requests(logs_repo, stream, sender).await {
warn!("error handling Log requests: {}", e);
}
})) {
warn!("Couldn't queue listener task: {:?}", e);
}
}
pub fn stop(&self) {
self.task_sender.lock().disconnect();
}
pub async fn wait_for_servers_to_complete(&self) {
let task = self
.drain_listeners_task
.lock()
.take()
.expect("The accessor server task is only awaited for once");
task.await;
}
async fn handle_requests(
logs_repo: Arc<LogsRepository>,
mut stream: flogger::LogRequestStream,
sender: Arc<Mutex<mpsc::UnboundedSender<fasync::Task<()>>>>,
) -> Result<(), LogsError> {
let connection_id = logs_repo.new_interest_connection();
while let Some(request) = stream.next().await {
let request = request.map_err(|source| LogsError::HandlingRequests {
protocol: flogger::LogMarker::PROTOCOL_NAME,
source,
})?;
let (listener, options, dump_logs, selectors) = match request {
flogger::LogRequest::ListenSafe { log_listener, options, .. } => {
(log_listener, options, false, None)
}
flogger::LogRequest::DumpLogsSafe { log_listener, options, .. } => {
(log_listener, options, true, None)
}
flogger::LogRequest::ListenSafeWithSelectors {
log_listener,
options,
selectors,
..
} => (log_listener, options, false, Some(selectors)),
};
let listener = Listener::new(listener, options)?;
let mode =
if dump_logs { StreamMode::Snapshot } else { StreamMode::SnapshotThenSubscribe };
let logs = logs_repo.logs_cursor(mode, None, ftrace::Id::random());
if let Some(s) = selectors {
logs_repo.update_logs_interest(connection_id, s);
}
sender.lock().unbounded_send(listener.spawn(logs, dump_logs)).ok();
}
logs_repo.finish_interest_connection(connection_id);
Ok(())
}
}