refactor: doerg는 publisher와 결합

2026-04-03 13:31:16 -06:00
parent 5ca59fdb5e
commit dcaac98252
88 changed files with 158 additions and 622 deletions
@@ -0,0 +1,125 @@
+(ns net.deertopia.doerg.common
+  (:require [babashka.process :as p]
+            [clojure.string :as str]
+            [clojure.tools.logging :as l]
+            [clojure.java.io :as io])
+  (:import (java.io FilterInputStream StringWriter InputStream
+                    OutputStream PrintStream ByteArrayOutputStream
+                    ByteArrayInputStream FilterOutputStream)
+           (java.nio.charset StandardCharsets)))
+
+(defn deref-with-timeout [process ms]
+  (let [p (promise)
+        process-future (future (deliver p @process))
+        timeout-future (future (Thread/sleep ms)
+                               (future-cancel process-future)
+                               (p/destroy-tree process)
+                               (deliver p ::timed-out))]
+    (if (= @p ::timed-out)
+      (throw (ex-info (format "external command `%s' timed out after %.2fs."
+                              (str/join " " (:cmd process))
+                              (/ (double ms) 1000))
+                      {:process process
+                       :timed-out-after-milliseconds ms}))
+      @p)))
+
+(defn tee-input-stream
+  "Return a wrapped `InputStream` that writes all bytes read from
+  input-stream to sink, à la the UNIX command tee(1)."
+  [input-stream sink]
+  (proxy [FilterInputStream] [input-stream]
+    (read
+      ([]
+       (let [c (proxy-super read)]
+         (when (not= c -1)
+           (.write sink c))
+         c))
+      ([^bytes bs]
+       (let [n (proxy-super read bs)]
+         (when (not= n -1)
+           (.write sink bs 0 n))
+         n))
+      ([^bytes bs off len]
+       (let [n (proxy-super read bs off len)]
+         (when (not= n -1)
+           (.write sink bs off n))
+         n)))
+    (close []
+      (try (proxy-super close)
+           (finally (.close sink))))))
+
+(defn tee-output-stream
+  "Return a wrapped `OutputStream` that writes all bytes written to
+  output-stream to sink, à la the UNIX command tee(1)."
+  [output-stream sink]
+  (proxy [FilterOutputStream] [output-stream]
+    (write
+      ([bs-or-b]
+       (proxy-super write bs-or-b)
+       (.write sink bs-or-b))
+      ([^bytes bs off len]
+       (proxy-super write bs off len)
+       (.write sink bs off len)))
+    (close []
+      (try (proxy-super close)
+           (finally (.close sink))))))
+
+#_
+(defn hook-input-stream [input-stream hook]
+  (proxy [FilterInputStream] [input-stream]
+    (read
+      ([]
+       (let [c (proxy-super read)]
+         (when (not= c -1)
+           (hook (byte-array [c])))
+         c))
+      ([^bytes bs]
+       (let [n (proxy-super read bs)]
+         (when (not= n -1)
+           (let [bs* (byte-array n 0)]
+             (System/arraycopy bs 0 bs* 0 n)
+             (hook bs*)))
+         n))
+      ([^bytes bs off len]
+       (let [n (proxy-super read bs off len)]
+         (when (not= n -1)
+           (.write sink bs off n))
+         n)))
+    (close []
+      (try (proxy-super close)
+           (finally (.close sink))))))
+
+(comment
+  (with-open [sink (ByteArrayOutputStream.)
+              out (ByteArrayOutputStream.)
+              in (ByteArrayInputStream. (.getBytes "hello worms"))]
+    (io/copy (tee-input-stream in sink) out)
+    (def the-out out)
+    (def the-sink sink)
+    {:out out
+     :sink sink})
+  (with-open [sink (l/log-stream :info "blah")
+              out (ByteArrayOutputStream.)
+              in (ByteArrayInputStream. (.getBytes "hello worms"))]
+    (io/copy (tee-input-stream in sink) out)
+    (def the-out out)
+    (def the-sink sink)
+    {:out out
+     :sink sink}))
+
+(comment
+  (let [out (ByteArrayOutputStream.)]
+    (p/shell {:out (tee-output-stream
+                    out (l/log-stream :info "blah"))}
+             "echo" "hello\n" "worms")
+    (.toString out)))
+
+(defn invoke [opts & cmd]
+  (l/info (str/join " " (cons "$" cmd)))
+  (let [r (apply p/shell
+                 (merge {:continue true
+                         :in nil :out :string :err :string}
+                        opts)
+                 cmd)
+        bin (first cmd)]
+    r))