corsa_client 0.1.0

Typed stdio API client bindings for typescript-go
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

//! Process-spawning helpers for the stdio API transports.
//!
//! The public API exposes higher-level configuration types such as
//! [`crate::ApiSpawnConfig`]. This module is responsible for translating those
//! settings into the exact command-line arguments and transport objects used by
//! the JSON-RPC and msgpack clients.

use super::{
    callbacks::{callback_flag, jsonrpc_handlers},
    driver::ClientDriver,
};
use crate::{
    Result, TsgoError,
    jsonrpc::{JsonRpcConnection, JsonRpcConnectionOptions},
    process::{AsyncChildGuard, TsgoCommand},
};
use corsa_core::fast::{CompactString, SmallVec};
use std::{
    io::{BufReader, BufWriter},
    sync::Arc,
};

pub(super) async fn spawn_jsonrpc_stdio(
    command: &TsgoCommand,
    filesystem: Option<Arc<dyn super::ApiFileSystem>>,
    request_timeout: Option<std::time::Duration>,
    shutdown_timeout: std::time::Duration,
    outbound_capacity: usize,
    observer: Option<corsa_core::SharedObserver>,
) -> Result<ClientDriver> {
    // JSON-RPC mode is used for callback-capable, async request/response
    // flows. The worker process is wrapped in `AsyncChildGuard` so shutdown
    // always reaps the child.
    let args = stdio_args(command, filesystem.as_deref(), true);
    let mut child = command.spawn_async(args.iter().map(CompactString::as_str))?;
    let stdin = child.stdin.take().ok_or(TsgoError::Closed("api stdin"))?;
    let stdout = child.stdout.take().ok_or(TsgoError::Closed("api stdout"))?;
    let handlers = filesystem.map(jsonrpc_handlers).unwrap_or_default();
    let rpc = JsonRpcConnection::spawn_with_options(
        BufReader::new(stdout),
        BufWriter::new(stdin),
        handlers,
        JsonRpcConnectionOptions::new()
            .with_request_timeout(request_timeout)
            .with_outbound_capacity(outbound_capacity)
            .with_observer_if_some(observer),
    );
    Ok(ClientDriver::JsonRpc {
        rpc,
        process: Some(Arc::new(AsyncChildGuard::new(child))),
        shutdown_timeout,
    })
}

pub(super) fn spawn_msgpack_stdio(
    command: &TsgoCommand,
    filesystem: Option<Arc<dyn super::ApiFileSystem>>,
    request_timeout: Option<std::time::Duration>,
    outbound_capacity: usize,
    observer: Option<corsa_core::SharedObserver>,
) -> Result<ClientDriver> {
    // Msgpack mode keeps a dedicated worker thread around the blocking stdio
    // pipes. This avoids async framing overhead on the hot path.
    let args = stdio_args(command, filesystem.as_deref(), false);
    let child = command.spawn_blocking(args.iter().map(CompactString::as_str))?;
    let worker = super::msgpack_worker::MsgpackWorker::spawn(
        child,
        filesystem,
        request_timeout,
        outbound_capacity,
        observer,
    )?;
    Ok(ClientDriver::Msgpack {
        worker: Arc::new(worker),
    })
}

fn stdio_args(
    command: &TsgoCommand,
    filesystem: Option<&dyn super::ApiFileSystem>,
    async_mode: bool,
) -> SmallVec<[CompactString; 6]> {
    let mut args = SmallVec::<[CompactString; 6]>::new();
    args.push(CompactString::from("--api"));
    if async_mode {
        args.push(CompactString::from("--async"));
    }
    // Pass the resolved working directory explicitly so downstream tools and
    // diagnostics see the same root the Rust side expects.
    args.push(CompactString::from("--cwd"));
    args.push(CompactString::from(command.cwd().display().to_string()));
    if let Some(filesystem) = filesystem.and_then(callback_flag) {
        args.push(filesystem);
    }
    args
}