char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somport http extension to typescript495c9a5

main
branchesmainall refs →
8.2 KiB199 linesraw
1mod file;
2
3use std::io;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::{Arc, Mutex};
6
7use avril_caps::{Capabilities, PathTarget};
8use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
9use avril_core::{Extension, Op};
10use avril_http::{Address, Authorize, Engine};
11use bytes::Bytes;
12
13use avril_ext_websocket::Sockets;
14
15type RespondArgs = (u64, u16, Vec<(String, String)>, Option<JsBuffer>, bool);
16
17pub fn init(
18    runtime: tokio::runtime::Handle,
19    caps: Arc<Capabilities>,
20    sockets: Arc<Sockets>,
21    files: Arc<avril_ext_file::Files>,
22    cli: Option<Address>,
23) -> Extension {
24    let engine = Arc::new(Engine::new(runtime.clone()));
25    let file_ops = file::ops(runtime.clone(), caps.clone(), files, engine.clone());
26    let cli = Mutex::new(cli);
27    let listen = engine.clone();
28    let host = engine.clone();
29    let next = engine.clone();
30    let finished = engine.clone();
31    let stop = engine.clone();
32    let read = engine.clone();
33    let respond = engine.clone();
34    let write = engine.clone();
35    let wait = engine.clone();
36    let upgrade = engine.clone();
37    let mut extension = Extension {
38        name: "ext:http",
39        dependencies: &[
40            "ext:http_types",
41            "ext:file",
42            "ext:web",
43            "ext:websocket",
44            "ext:webstreams",
45        ],
46        ops: vec![
47            Op::new("op_http_listen", move |mut address: Address| {
48                let caps = caps.clone();
49                let authorize: Authorize = match &mut address {
50                    Address::Tcp { hostname, .. } => {
51                        let hostname = hostname.clone();
52                        Arc::new(move |address| {
53                            let Address::Tcp { port, .. } = address else {
54                                return Err("Invalid listener transport".into());
55                            };
56                            caps.check_net(&hostname, *port).map_err(Into::into)
57                        })
58                    }
59                    Address::Unix { path } => {
60                        let label = path
61                            .as_ref()
62                            .ok_or_else(|| io::Error::other("Missing Unix socket path"))?;
63                        if label.is_empty() {
64                            return Err(io::Error::other("Empty Unix socket path"));
65                        }
66                        let target = PathTarget::for_entry(caps.cwd().join(label))?;
67                        *path = Some(
68                            target
69                                .as_path()
70                                .to_str()
71                                .ok_or_else(|| io::Error::other("Unix socket paths must be UTF-8"))?
72                                .to_owned(),
73                        );
74                        Arc::new(move |_| caps.check_write(&target).map_err(Into::into))
75                    }
76                };
77                listen.listen(address, authorize).map_err(io::Error::other)
78            }),
79            Op::with_scope("op_http_cli_listen", move |scope, _: ()| {
80                let address = cli
81                    .lock()
82                    .unwrap()
83                    .take()
84                    .ok_or_else(|| io::Error::other("No CLI listener configured"))?;
85                let weak = Arc::downgrade(&host);
86                let interrupted = Arc::new(AtomicBool::new(false));
87                let stopping = interrupted.clone();
88                let isolate = scope.thread_safe_handle();
89                let shutdown = Arc::new(move || {
90                    if let Some(engine) = weak.upgrade() {
91                        let force = stopping.swap(true, Ordering::SeqCst);
92                        engine.stop_all(force);
93                        if force {
94                            avril_ext_process::Tree::global().force_live_processes();
95                            isolate.terminate_execution();
96                        }
97                    }
98                });
99                let interrupt = shutdown.clone();
100                #[cfg(unix)]
101                // Serve owns SIGINT fate from here on through the shared
102                // chaining registry: top-level listeners observe the
103                // shutdown instead of being clobbered by it.
104                avril_ext_process::Tree::global().claim_interrupt(&runtime, interrupt)?;
105                #[cfg(not(unix))]
106                ctrlc::set_handler(move || interrupt()).map_err(io::Error::other)?;
107                let bound = host
108                    .listen(address, Arc::new(|_| Ok(())))
109                    .map_err(io::Error::other)?;
110                #[cfg(unix)]
111                {
112                    // Serve owns SIGTERM fate from here on: the subprocess
113                    // supervisor hands first deliveries to this shutdown
114                    // instead of stopping children itself.
115                    avril_ext_process::Tree::global().claim_term(&runtime, shutdown)?;
116                }
117                if interrupted.load(Ordering::SeqCst) {
118                    host.stop(bound.0, true);
119                    return Err(io::Error::other("Server shutdown requested during binding"));
120                }
121                match &bound.1 {
122                    Address::Tcp { hostname, port } => eprintln!(
123                        "Listening on http://{}:{port}",
124                        if hostname.contains(':') {
125                            format!("[{hostname}]")
126                        } else {
127                            hostname.clone()
128                        }
129                    ),
130                    Address::Unix { path } => {
131                        eprintln!("Listening on unix:{}", path.as_deref().unwrap_or_default())
132                    }
133                }
134                Ok::<_, io::Error>(bound)
135            }),
136            Op::asynchronous("op_http_next", move |id: u64| next.next(id)),
137            Op::asynchronous("op_http_finished", move |id: u64| finished.finished(id)),
138            Op::new("op_http_stop", move |(id, force): (u64, bool)| {
139                stop.stop(id, force);
140                Ok::<_, io::Error>(())
141            }),
142            Op::asynchronous("op_http_read", move |id: u64| {
143                let pending = read.read(id);
144                async move {
145                    // JS needs writable memory independent of Hyper's shared chunks.
146                    Ok::<_, io::Error>(pending.await?.map(|bytes| ToJsBuffer::from(bytes.to_vec())))
147                }
148            }),
149            Op::new(
150                "op_http_respond",
151                move |(id, status, headers, bytes, streaming): RespondArgs| {
152                    respond
153                        .respond((
154                            id,
155                            status,
156                            headers,
157                            bytes.map(|bytes| Bytes::copy_from_slice(&bytes)),
158                            streaming,
159                        ))
160                        .map_err(io::Error::other)
161                },
162            ),
163            Op::asynchronous(
164                "op_http_write",
165                move |(id, bytes): (u64, Option<JsBuffer>)| {
166                    write.write(id, bytes.map(|bytes| Bytes::copy_from_slice(&bytes)))
167                },
168            ),
169            Op::asynchronous("op_http_wait", move |id: u64| wait.wait(id)),
170            Op::new(
171                "op_http_upgrade",
172                move |(id, protocol): (u64, Option<String>)| {
173                    let (pending, headers) = upgrade
174                        .upgrade(id, protocol.as_deref())
175                        .map_err(io::Error::other)?;
176                    let socket = sockets.accept(pending, protocol.unwrap_or_default())?;
177                    Ok::<_, io::Error>((socket, headers))
178                },
179            ),
180            Op::new("op_http_cancel", move |id: u64| {
181                engine.cancel(id);
182                Ok::<_, io::Error>(())
183            }),
184        ],
185        modules: &[
186            (
187                "ext:http/main.js",
188                include_str!(concat!(env!("OUT_DIR"), "/main.js")),
189            ),
190            (
191                "ext:http/file.js",
192                include_str!(concat!(env!("OUT_DIR"), "/file.js")),
193            ),
194        ],
195        entry_point: None,
196    };
197    extension.ops.extend(file_ops);
198    extension
199}