Add an option to set the process filter

The process filter is an optional function to be used as the process
filter for the curl process.  It can be used to handle HTTP responses
in a streaming way.  The function must accept 2 arguments, the process
object running curl, and a string which is output received from the
process.  The default process filter inserts the output of the process
into the process buffer.  The provided function should at least insert
output up to the HTTP body into the process buffer.

README.org

@@ -142,6 +142,9 @@ Synchronously download a JPEG file, then create an Emacs image object from the d
 
    ~NOQUERY~ is passed to ~make-process~, which see.
 
+   ~FILTER~ is an optional function to be used as the process filter for the curl process.  It can be used to handle HTTP responses in a streaming way.  The function must accept 2 arguments, the process object running curl, and a string which is output received from the process.  The default process filter inserts the output of the process into the process buffer.  The provided ~FILTER~ function should at least insert output up to the HTTP body into the process buffer.
+
+
 ** Queueing
 
 ~plz~ provides a simple system for queueing HTTP requests.  First, make a ~plz-queue~ struct by calling ~make-plz-queue~.  Then call ~plz-queue~ with the struct as the first argument, and the rest of the arguments being the same as those passed to ~plz~.  Then call ~plz-run~ to run the queued requests.

plz.el

@@ -254,7 +254,7 @@ connection phase and waiting to receive the response (the
 
 ;;;;; Public
 
-(cl-defun plz (method url &rest rest &key headers body else finally noquery
+(cl-defun plz (method url &rest rest &key headers body else filter finally noquery
                       (as 'string) (then 'sync)
                       (body-type 'text) (decode t decode-s)
                       (connect-timeout plz-connect-timeout) (timeout plz-timeout))
@@ -330,6 +330,15 @@ from a host, respectively.
 
 NOQUERY is passed to `make-process', which see.
 
+FILTER is an optional function to be used as the process filter
+for the curl process.  It can be used to handle HTTP responses in
+a streaming way.  The function must accept 2 arguments, the
+process object running curl, and a string which is output
+received from the process.  The default process filter inserts
+the output of the process into the process buffer.  The provided
+FILTER function should at least insert output up to the HTTP body
+into the process buffer.
+
 \(To silence checkdoc, we mention the internal argument REST.)"
   ;; FIXME(v0.8): Remove the note about error changes from the docstring.
   ;; FIXME(v0.8): Update error signals in docstring.
@@ -404,6 +413,15 @@ NOQUERY is passed to `make-process', which see.
                                 :coding 'binary
                                 :command (append (list plz-curl-program) curl-command-line-args)
                                 :connection-type 'pipe
+                                :filter (when filter
+                                          (lambda (process output)
+                                            (unwind-protect
+                                                (progn
+                                                  (process-put process :plz-filter-mark
+                                                               (1+ (or (process-get process :plz-filter-mark) 0)))
+                                                  (funcall filter process output))
+                                              (process-put process :plz-filter-mark
+                                                           (1- (process-get process :plz-filter-mark))))))
                                 :sentinel #'plz--sentinel
                                 :stderr stderr-process
                                 :noquery noquery))
@@ -734,9 +752,20 @@ for asynchronous ones)."
          (pred numberp)
          (rx "exited abnormally with code " (group (1+ digit))))
      (let ((buffer (process-buffer process)))
-       (if (process-get process :plz-sync)
-           (plz--respond process buffer status)
-         (run-at-time 0 nil #'plz--respond process buffer status))))))
+       (cond
+        ;; Respond to synchrounous requests immediately.
+        ((process-get process :plz-sync)
+         (plz--respond process buffer status))
+        ;; Respond to asynchrounous requests with timer if no process
+        ;; filter is set.
+        ((null (process-get process :plz-filter-mark))
+         (run-at-time 0 nil #'plz--respond process buffer status))
+        ;; Respond to asynchrounous requests with timer if process
+        ;; filter is set and all filters have completed.
+        ((zerop (process-get process :plz-filter-mark))
+         (run-at-time 0 nil #'plz--respond process buffer status))
+        ;; Otherwise wait for process filters to complete by trying again.
+        (t (run-at-time 0 nil #'plz--sentinel process status)))))))
 
 (defun plz--respond (process buffer status)
   "Respond to HTTP response from PROCESS in BUFFER.

tests/test-plz.el

@@ -60,17 +60,23 @@ If running httpbin locally, set to \"http://localhost\".")
 
 ;;;; Macros
 
-(cl-defun plz-test-wait (process &optional (seconds 0.1) (times 100))
-  "Wait for SECONDS seconds TIMES times for PROCESS to finish."
-  (when process
+(cl-defun plz-test-wait (process-or-fn &optional (seconds 0.1) (times 100))
+  "Wait for SECONDS seconds TIMES times for PROCESS-OR-FN.
+
+If PROCESS-OR-FN is a process, wait for it to finish.  If it's a
+function wait until it returns non-nil."
+  (when process-or-fn
     ;; Sometimes it seems that the process is killed, the THEN
     ;; function called by its sentinel, and its buffer killed, all
     ;; before this function gets called with the process argument;
     ;; when that happens, tests that use this can fail.  Testing
     ;; whether PROCESS is non-nil seems to fix it, but it's possible
     ;; that something funny is going on...
     (cl-loop for i upto times ;; 10 seconds
-             while (equal 'run (process-status process))
+             while (cond ((processp process-or-fn)
+                          (equal 'run (process-status process-or-fn)))
+                         ((functionp process-or-fn)
+                          (not (funcall process-or-fn))))
              do (sleep-for seconds))))
 
 (cl-defmacro plz-deftest (name () &body docstring-keys-and-body)
@@ -559,6 +565,84 @@ and only called once."
 
 ;; TODO: Add test for canceling queue.
 
+;; Process filter
+
+(defun test-plz-process-filter (process output)
+  "Write OUTPUT to the PROCESS buffer."
+  (when (buffer-live-p (process-buffer process))
+    (with-current-buffer (process-buffer process)
+      (let ((moving (= (point) (process-mark process))))
+        (save-excursion
+          (goto-char (process-mark process))
+          (insert output)
+          (set-marker (process-mark process) (point)))
+        (if moving (goto-char (process-mark process)))))))
+
+(plz-deftest plz-get-json-process-filter-async ()
+  (let* ((test-json) (outputs)
+         (process (plz 'get (url "/get")
+                    :as #'json-read
+                    :then (lambda (json)
+                            (setf test-json json))
+                    :filter (lambda (process output)
+                              (test-plz-process-filter process output)
+                              (push output outputs)))))
+    (plz-test-wait process)
+    (let-alist test-json
+      (should (string-match "curl" .headers.User-Agent)))
+    (let ((output (string-join (reverse outputs))))
+      (should (string-match "HTTP.*\s+200" output))
+      (should (string-match "Server: gunicorn" output))
+      (should (string-match "\"args\":\s*{}" output)))))
+
+(plz-deftest plz-get-json-process-filter-sync ()
+  (let* ((outputs)
+         (response (plz 'get (url "/get")
+                    :as 'response
+                    :filter (lambda (process output)
+                              (test-plz-process-filter process output)
+                              (push output outputs)))))
+    (plz-test-get-response response)
+    (let ((output (string-join (reverse outputs))))
+      (should (string-match "HTTP.*\s+200" output))
+      (should (string-match "Server: gunicorn" output))
+      (should (string-match "\"args\":\s*{}" output)))))
+
+(plz-deftest plz-get-json-slow-process-filter-async ()
+  (let* ((test-json) (outputs)
+         (process (plz 'get (url "/get")
+                    :as #'json-read
+                    :then (lambda (json)
+                            (setf test-json json))
+                    :filter (lambda (process output)
+                              (test-plz-process-filter process output)
+                              (push output outputs)
+                              (sleep-for 1)))))
+    (plz-test-wait process)
+    ;; The process finished, but the filter is still sleeping. Wait
+    ;; for the :then callback to complete.
+    (plz-test-wait (lambda () test-json))
+    (let-alist test-json
+      (should (string-match "curl" .headers.User-Agent)))
+    (let ((output (string-join (reverse outputs))))
+      (should (string-match "HTTP.*\s+200" output))
+      (should (string-match "Server: gunicorn" output))
+      (should (string-match "\"args\":\s*{}" output)))))
+
+(plz-deftest plz-get-json-slow-process-filter-sync ()
+  (let* ((outputs)
+         (response (plz 'get (url "/get")
+                    :as 'response
+                    :filter (lambda (process output)
+                              (test-plz-process-filter process output)
+                              (push output outputs)
+                              (sleep-for 1)))))
+    (plz-test-get-response response)
+    (let ((output (string-join (reverse outputs))))
+      (should (string-match "HTTP.*\s+200" output))
+      (should (string-match "Server: gunicorn" output))
+      (should (string-match "\"args\":\s*{}" output)))))
+
 ;;;; Footer
 
 (provide 'test-plz)