subprocess: use promisify() and generators

Author: amezin

ddterm/shell/extension.js

@@ -283,19 +283,14 @@ class EnabledExtension {
         });
 
         this.service.connect('error', (service, ex) => {
-            const log_collector = service.subprocess?.log_collector;
-
-            if (!log_collector) {
-                this.notifications.show_error(ex);
-                return;
-            }
-
-            log_collector.collect().then(output => {
-                this.notifications.show_error(ex, output);
-            }).catch(ex2 => {
-                logError(ex2, 'Failed to collect logs');
-                this.notifications.show_error(ex);
-            });
+            (service.subprocess?.get_logs() ?? Promise.resolve()).then(
+                output => {
+                    this.notifications.show_error(ex, output);
+                }, ex2 => {
+                    logError(ex2, 'Failed to collect logs');
+                    this.notifications.show_error(ex);
+                }
+            );
         });
 
         this.window_geometry = new WindowGeometry();

ddterm/shell/subprocess.js

@@ -11,6 +11,7 @@ import Meta from 'gi://Meta';
 import Gi from 'gi';
 
 import { sd_journal_stream_fd } from './sd_journal.js';
+import { promisify } from '../util/promise.js';
 
 function try_require(namespace, version = undefined) {
     try {
@@ -60,115 +61,102 @@ function shell_join(argv) {
     return argv.map(arg => GLib.shell_quote(arg)).join(' ');
 }
 
-class JournalctlLogCollector {
-    constructor(journalctl, since, pid) {
-        this._argv = [
-            journalctl,
-            '--user',
-            '-b',
-            `--since=${since.format('%C%y-%m-%d %H:%M:%S UTC')}`,
-            '-ocat',
-            `-n${KEEP_LOG_LINES}`,
-            `_PID=${pid}`,
-        ];
-    }
+async function collect_journald_logs(journalctl, since, pid) {
+    const argv = [
+        journalctl,
+        '--user',
+        '-b',
+        `--since=${since.format('%C%y-%m-%d %H:%M:%S UTC')}`,
+        '-ocat',
+        `-n${KEEP_LOG_LINES}`,
+    ];
+
+    if (pid)
+        argv.push(`_PID=${pid}`);
+
+    const proc = Gio.Subprocess.new(
+        argv,
+        Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_MERGE
+    );
 
-    _begin(resolve, reject) {
-        const proc = Gio.Subprocess.new(
-            this._argv,
-            Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_MERGE
-        );
+    const communicate = promisify(proc.communicate_async, proc.communicate_finish);
+    const [, stdout_buf] = await communicate.call(proc, null, null);
 
-        proc.communicate_utf8_async(null, null, this._finish.bind(this, resolve, reject));
-    }
+    return new TextDecoder().decode(stdout_buf);
+}
 
-    _finish(resolve, reject, source, result) {
-        try {
-            const [, stdout_buf] = source.communicate_utf8_finish(result);
-            resolve(stdout_buf);
-        } catch (ex) {
-            reject(ex);
-        }
-    }
+async function *read_chunks(input_stream) {
+    const read_bytes =
+        promisify(input_stream.read_bytes_async, input_stream.read_bytes_finish);
 
-    collect() {
-        return new Promise(this._begin.bind(this));
-    }
-}
+    try {
+        for (;;) {
+            // eslint-disable-next-line no-await-in-loop
+            const chunk = await read_bytes.call(input_stream, 4096, GLib.PRIORITY_DEFAULT, null);
 
-class TeeLogCollector {
-    constructor(stream) {
-        this._input = stream;
-        this._output = new UnixOutputStream({ fd: STDERR_FD, close_fd: false });
-        this._collected = [];
-        this._collected_lines = 0;
-        this._promise = new Promise((resolve, reject) => {
-            this._resolve = resolve;
-            this._reject = reject;
-        });
-
-        this._read_more();
-    }
+            if (chunk.get_size() === 0)
+                return;
 
-    _read_more() {
-        this._input.read_bytes_async(4096, GLib.PRIORITY_DEFAULT, null, this._read_done.bind(this));
+            yield chunk.toArray();
+        }
+    } finally {
+        input_stream.close(null);
     }
+}
 
-    _read_done(source, result) {
-        try {
-            const chunk = source.read_bytes_finish(result).toArray();
+function *split_array_keep_delimiter(bytes, delimiter) {
+    let start = 0;
 
-            if (chunk.length === 0) {
-                this._input.close(null);
-                this._output.close(null);
-                this._resolve();
-                return;
-            }
+    for (;;) {
+        let end = bytes.indexOf(delimiter, start);
+
+        if (end === -1)
+            break;
 
-            const delimiter = '\n'.charCodeAt(0);
-            let start = 0;
+        yield bytes.subarray(start, end + 1);
 
-            for (;;) {
-                let end = chunk.indexOf(delimiter, start);
+        start = end + 1;
+    }
 
-                if (end === -1) {
-                    if (start < chunk.length)
-                        this._collected.push(chunk.subarray(start));
+    yield bytes.subarray(start);
+}
 
-                    break;
-                }
+async function collect_stdio_logs(input_stream) {
+    const delimiter = '\n'.charCodeAt(0);
+    const collected = [];
+    let lines = 0;
+    const stderr = new UnixOutputStream({ fd: STDERR_FD, close_fd: false });
 
-                this._collected.push(chunk.subarray(start, end + 1));
-                this._collected_lines += 1;
+    for await (const chunk of read_chunks(input_stream)) {
+        // I hope sync/blocking writes to stderr are fine.
+        // After all, this is the same thing that printerr() does.
+        stderr.write_all(chunk, null);
 
-                start = end + 1;
-            }
+        for (const sub_chunk of split_array_keep_delimiter(chunk, delimiter)) {
+            collected.push(sub_chunk);
 
-            let remove = 0;
+            if (sub_chunk.at(-1) === delimiter)
+                lines += 1;
+        }
 
-            while (this._collected_lines > KEEP_LOG_LINES) {
-                const remove_chunk = this._collected[remove];
+        let remove = 0;
 
-                remove += 1;
+        while (lines > KEEP_LOG_LINES) {
+            const remove_chunk = collected[remove];
 
-                if (remove_chunk[remove_chunk.length - 1] === delimiter)
-                    this._collected_lines -= 1;
-            }
+            remove += 1;
 
-            this._collected.splice(0, remove);
-            this._output.write(chunk, null);
-            this._read_more();
-        } catch (ex) {
-            this._reject(ex);
+            if (remove_chunk.at(-1) === delimiter)
+                lines -= 1;
         }
+
+        if (remove > 0)
+            collected.splice(0, remove);
     }
 
-    async collect() {
-        await this._promise;
+    const decoder = new TextDecoder();
 
-        const decoder = new TextDecoder();
-        return this._collected.map(line => decoder.decode(line)).join('\n');
-    }
+    return collected.map(v => decoder.decode(v)).join('');
 }
 
 export const Subprocess = GObject.registerClass({
@@ -213,10 +201,6 @@ export const Subprocess = GObject.registerClass({
             ? make_subprocess_launcher_journald(this.journal_identifier)
             : make_subprocess_launcher_fallback();
 
-        const launch_context = global.create_app_launch_context(0, -1);
-
-        subprocess_launcher.set_environ(launch_context.get_environment());
-
         for (const extra_env of this.environ) {
             const split_pos = extra_env.indexOf('=');
             const name = extra_env.slice(0, split_pos);
@@ -231,13 +215,18 @@ export const Subprocess = GObject.registerClass({
             subprocess_launcher.close();
         }
 
-        this.log_collector = logging_to_journald
-            ? new JournalctlLogCollector(journalctl, start_date, this._subprocess.get_identifier())
-            : new TeeLogCollector(this._subprocess.get_stdout_pipe());
+        const pid = this._subprocess.get_identifier();
+
+        this._get_logs = logging_to_journald
+            ? collect_journald_logs.bind(globalThis, journalctl, start_date, pid)
+            : collect_stdio_logs(this._subprocess.get_stdout_pipe()).catch(logError);
+
+        if (!pid)
+            return;
 
         GnomeDesktop.start_systemd_scope(
             this.journal_identifier,
-            parseInt(this._subprocess.get_identifier(), 10),
+            parseInt(pid, 10),
             null,
             null,
             null,
@@ -261,33 +250,28 @@ export const Subprocess = GObject.registerClass({
     }
 
     wait(cancellable = null) {
-        return new Promise((resolve, reject) => {
-            this.g_subprocess.wait_async(cancellable, (source, result) => {
-                try {
-                    resolve(source.wait_finish(result));
-                } catch (ex) {
-                    reject(ex);
-                }
-            });
-        });
+        const { wait_async, wait_finish } = this.g_subprocess;
+
+        return promisify(wait_async, wait_finish).call(this.g_subprocess, cancellable);
     }
 
     wait_check(cancellable = null) {
-        return new Promise((resolve, reject) => {
-            this.g_subprocess.wait_check_async(cancellable, (source, result) => {
-                try {
-                    resolve(source.wait_check_finish(result));
-                } catch (ex) {
-                    reject(ex);
-                }
-            });
-        });
+        const { wait_check_async, wait_check_finish } = this.g_subprocess;
+
+        return promisify(wait_check_async, wait_check_finish).call(this.g_subprocess, cancellable);
     }
 
     terminate() {
         this.g_subprocess.send_signal(SIGTERM);
     }
 
+    get_logs() {
+        if (this._get_logs instanceof Function)
+            return this._get_logs();
+
+        return this._get_logs;
+    }
+
     _spawn(subprocess_launcher) {
         log(`Starting subprocess: ${shell_join(this.argv)}`);
         return subprocess_launcher.spawnv(this.argv);