Add sh/lines

It starts a command and retrieves each line from the command's standard output asynchronously.

spork/sh.janet

@@ -181,3 +181,101 @@
   "Output a string with all arguments correctly quoted"
   [& args]
   (string/join (map shell-quote args) " "))
+
+(defn lines
+  ```
+  It executes a list of command arguments described by args, starts an asynchronous task that retrieves each line from
+  the command's standard output as a buffer, and returns a struct with the following keys.
+
+  * `:chan` - The channel that gives each line from the task.
+  * `:fiber` - A fiber that yields each line from the channel.
+  * `:cancel` - An object-oriented function that accepts the struct and an optional argument which is signal passed to
+    `os/proc-kill`. The default signal is `:term`. It kills the process and ends the task. You can call this function
+    multiple times without errors.
+  * `:exit-code` - An object-oriented function that accepts the struct and returns the process exit code after waiting
+    for the task to end. After this function returns the exit code, it returns `nil`.
+
+  Each line is terminated by a newline character, `\n`. After end of process, the channel gives the last line from the
+  process. After the channel gives the last line, the process exit code is retrieved, and the process is closed before
+  the task ends. Make sure to either take all lines from the channel or call `:cancel`. If you don't, the process
+  remains frozen in the background or becomes a zombie process.
+
+  If `stderr` is false or nil, /dev/null is opened and fed the command's standard error and closed when the task
+  finishes. If `stderr` is a `core/file` value or a `core/stream` value, `stderr` is fed the command's standard error,
+  and it is not closed by this function. If `stderr` is any other value, the returned struct gains a key, `:stderr`.
+  `:stderr` is an object-oriented function that accepts the struct and returns standard error of the command as a
+  buffer after waiting for the task to finish. After it is called for the first time, it returns `nil`.
+  ```
+  [args &named stderr]
+  (def /dev/null (unless stderr
+                   (devnull)))
+  (def proc (os/spawn args :p {:out :pipe :err (cond
+                                                 # If `stderr` is false or nil
+                                                 (not (nil? /dev/null))
+                                                 /dev/null
+                                                 # If `stderr` is core/file or core/stream
+                                                 (let [stderr-type (type stderr)]
+                                                   (or (= stderr-type :core/file)
+                                                       (= stderr-type :core/stream)))
+                                                 stderr
+                                                 # If `stderr` is any other value
+                                                 :pipe)}))
+  (def {:out out :err err} proc)
+  (def exit-code (ev/chan 1))
+  (def stderr-ch (when err
+                   (ev/chan 1)))
+  (when stderr-ch
+    (ev/spawn
+      (ev/give stderr-ch (ev/read err :all))))
+  # lines channel must have capacity of 0. Otherwise, it can be closed before `ev/take` can take the last line.
+  (def lines (ev/chan))
+  (defn fetch-lines
+    [chunk]
+    # if-let breaks tail call optimization
+    (def idx (string/find "\n" chunk))
+    (if idx
+      (do
+        # Give the first line
+        (ev/give lines (buffer/slice chunk 0 idx))
+        # Eliminate the first line from chunk without creating a new buffer
+        (def idx+1 (inc idx))
+        (buffer/blit chunk chunk 0 idx+1)
+        (fetch-lines (buffer/popn chunk idx+1)))
+      (if (ev/read out 1024 chunk)
+        (fetch-lines chunk)
+        (when (not (empty? chunk))
+          (ev/give lines chunk)))))
+  (ev/spawn
+    (defer (do
+             (:close lines)
+             (when /dev/null
+               (:close /dev/null))
+             (ev/give exit-code (:close proc)))
+      (try
+        (when-let [chunk (ev/read out 1024)]
+          (fetch-lines chunk))
+        # :cancel causes an error.
+        ([_]))))
+  {:chan lines
+   :fiber (fiber/new (fn recurse []
+                       # if-let breaks tail call optimization
+                       (def line (ev/take lines))
+                       (when line
+                         (yield line)
+                         (recurse)))
+                     :yi)
+   :cancel (fn [self &opt signal]
+             (default signal :term)
+             (try
+               (os/proc-kill proc false signal)
+               # after the process exit code is retrieved, `os/proc-kill` throws an error.
+               ([_]))
+             (:close lines)
+             nil)
+   :exit-code (fn [self] (when-let [ec (ev/take exit-code)]
+                           (ev/chan-close exit-code)
+                           ec))
+   :stderr (when stderr-ch
+             (fn [self] (when-let [buf (ev/take stderr-ch)]
+                          (ev/chan-close stderr-ch)
+                          buf)))})