aether-lspd 0.3.4

use crate::protocol::{DaemonCall, DaemonRequest, DaemonResponse, LspErrorResponse, frame_reader, frame_writer};
use crate::workspace_registry::{WorkspaceBinding, WorkspaceRegistry};
use futures::{SinkExt, StreamExt};
use tokio::io::{ReadHalf, WriteHalf, split};
use tokio::net::UnixStream;
use tokio::spawn;
use tokio::sync::mpsc;

#[tracing::instrument(skip(stream, registry), fields(%client_id))]
pub(crate) async fn handle_client(stream: UnixStream, registry: WorkspaceRegistry, client_id: uuid::Uuid) {
    let (reader, writer) = split(stream);
    let (response_tx, response_rx) = mpsc::channel::<DaemonResponse>(100);
    let writer_task = spawn(run_writer(writer, response_rx));
    run_reader(reader, registry, client_id, response_tx).await;
    let _ = writer_task.await;
}

async fn run_writer(writer: WriteHalf<UnixStream>, mut response_rx: mpsc::Receiver<DaemonResponse>) {
    let mut writer = frame_writer::<_, DaemonResponse>(writer);
    while let Some(response) = response_rx.recv().await {
        if let Err(err) = writer.send(response).await {
            tracing::debug!(%err, "Error writing daemon response");
            break;
        }
    }
}

async fn run_reader(
    reader: ReadHalf<UnixStream>,
    registry: WorkspaceRegistry,
    client_id: uuid::Uuid,
    response_tx: mpsc::Sender<DaemonResponse>,
) {
    tracing::debug!("Client connected: {}", client_id);
    let mut binding: Option<WorkspaceBinding> = None;
    let mut reader = frame_reader::<_, DaemonRequest>(reader);

    while let Some(msg) = reader.next().await {
        let request = match msg {
            Ok(request) => request,
            Err(err) => {
                tracing::debug!(%err, "Error reading client request");
                break;
            }
        };

        match request {
            DaemonRequest::Disconnect => break,
            DaemonRequest::Initialize(init) => {
                let response = match registry.bind(&init.workspace_root, init.language).await {
                    Ok(bound) => {
                        binding = Some(bound);
                        DaemonResponse::Initialized
                    }
                    Err(err) => DaemonResponse::InitializeFailed(err.to_string()),
                };
                let _ = response_tx.send(response).await;
            }
            DaemonRequest::Call { id, call } => {
                let ordered = matches!(call, DaemonCall::QueueRefresh { .. });
                let reply = reply(registry.clone(), binding.clone(), id, call, response_tx.clone());
                if ordered {
                    reply.await;
                } else {
                    spawn(reply);
                }
            }
        }
    }
}

async fn reply(
    registry: WorkspaceRegistry,
    binding: Option<WorkspaceBinding>,
    id: i64,
    call: DaemonCall,
    response_tx: mpsc::Sender<DaemonResponse>,
) {
    let result = match binding {
        Some(binding) => registry.call(&binding, call).await,
        None => Err(LspErrorResponse::daemon("Not initialized")),
    };
    let _ = response_tx.send(DaemonResponse::Reply { id, result }).await;
}