anchor.diff:                        (context-host context)
                         (context-port context)
                         path))
  -         (parameters (concatenate 'list
  -                                  `(("user.name" . ,user)
  -                                    ("op" . ,operation)
  -                                    ("namenoderpcaddress" . ,*namenode-rpc-address*))
  -                                  parameters)))
  -    (send-request method url :parameters parameters)))
  +         (op-parameters `(("user.name" . ,user)
  +                          ("op" . ,operation)
  +                          ("namenoderpcaddress" . ,*namenode-rpc-address*)))
  +         (parameters (alist-merge (list parameters
  +                                        op-parameters))))
  +    (send-request method url
  +                  :parameters parameters
  +                  :response-handler response-handler
  +                  :redirect-handler redirect-handler)))
  +
  +(defun open-file (context path output-stream
  +                  &key offset length buffersize progress-callback)
  +  (labels ((stream-response (response-stream headers)
  +             (declare (ignore headers))
  +             (copy-stream output-stream
  +                          response-stream
  +                          :progress-callback progress-callback)
  +             t)
  +           (location-redirect (location)
  +             (send-request :get location
  +                           :response-handler #'stream-response)))
  +    (send-operation-request context :get "OPEN" path
  +                            :parameters `(("offset" . ,offset)
  +                                          ("length" . ,length)
  +                                          ("buffersize" . ,buffersize))
  +                            :response-handler #'stream-response
  +                            :redirect-handler #'location-redirect)))
  +
  +(defun send-data-request (context method operation path
  +                          content-stream content-length
  +                          &key parameters progress-callback)
  +  (unless (subtypep (stream-element-type content-stream) 'octet)
  +    (error "Expected a binary content stream"))
  +  (let ((length (or content-length
  +                    (when (typep content-stream 'file-stream)
  +                      (file-length content-stream))
  +                    (error "Can't determine the content length")))
  +        (wrapper (lambda (request-stream)
  +                   (copy-stream request-stream
  +                                content-stream
  +                                :progress-callback progress-callback))))
  +    (flet ((unexpected-response (response-stream headers)
  +             (declare (ignore response-stream headers))
  +             (error "Expected a redirect, got data."))
  +           (upload-data (location)
  +             (send-request method location
  +                           :content wrapper
  +                           :content-length length
  +                           :response-handler #'ignore-response)))
  +      (send-operation-request context method operation path
  +                              :parameters parameters
  +                              :response-handler #'unexpected-response
  +                              :redirect-handler #'upload-data)))
  +  t)
  +
  +(defun create (context path content-stream
  +               &key
  +                 overwrite blocksize replication permission buffersize
  +                 content-length progress-callback)
  +  (let ((parameters `(("overwrite" . ,(if overwrite "true" "false"))
  +                      ("blocksize" . ,blocksize)
  +                      ("replication" . ,replication)
  +                      ("permission" . ,permission)
  +                      ("buffersize" . ,buffersize))))
  +    (send-data-request context :put "CREATE" path content-stream content-length
  +                       :parameters parameters
  +                       :progress-callback progress-callback)))
  +
  +(defun append-to (context path content-stream
  +                  &key buffersize content-length progress-callback)
  +  (bind ((parameters `(("buffersize" . ,buffersize))))
  +    (send-data-request context :post "APPEND" path content-stream content-length
  +                       :parameters parameters
  +                       :progress-callback progress-callback)))
   
   (defun from-hadoop-time (hadoop-time)
     (let ((secs (floor (/ hadoop-time 1000.0))))
       (from-posix-time secs)))
   
  -(defun send-content (content callback stream)
  -  (unless (and (streamp content)
  -               (input-stream-p content)
  -               (open-stream-p content)
  -               (subtypep (stream-element-type content) 'flexi-streams:octet))
  -    (error "Expected a binary input stream"))
  -  (iter (with buf = (make-array 8192 :element-type 'flexi-streams:octet))
  -        (for num = (read-sequence buf content))
  -        (sum num into pos)
  -        (when callback (for dat first nil then (funcall callback pos dat)))
  -        (if (zerop num)
  -            (terminate)
  -            (write-sequence buf stream :end num))))
  -
  -(defun create (context path content length
  -               &key overwrite blocksize replication permission buffersize callback)
  -  (let* ((parameters `(("overwrite" . ,(if overwrite "true" "false"))
  -                       ("blocksize" . ,blocksize)
  -                       ("replication" . ,replication)
  -                       ("permission" . ,permission)
  -                       ("buffersize" . ,buffersize)))
  -         (headers1 (nth-value 2 (operation-request context :put "CREATE" path
  -                                                   :parameters parameters)))
  -         (location1 (alist-get :location headers1))
  -         (headers2 (nth-value 2 (send-request :put location1
  -                                              :content (curry #'send-content content callback)
  -                                              :content-length length)))
  -         (location2 (or (alist-get :location headers2) path)))
  -    location2))
  -
  -(defun append-to (context path content length &key buffersize callback)
  -  (let* ((parameters `(("buffersize" . ,buffersize)))
  -         (headers1 (nth-value 2 (operation-request context :post "APPEND" path
  -                                                   :parameters parameters)))
  -         (location1 (alist-get :location headers1))
  -         (headers2 (nth-value 2 (send-request :post location1
  -                                              :content (curry #'send-content content callback)
  -                                              :content-length length)))
  -         (location2 (or (alist-get :location headers2) path)))
  -    location2))
  -
   (defun list-status (context path)
  -  (let* ((res (operation-request context :get "LISTSTATUS" path))
  -         (statuses (alist-get :*file-status (alist-get :*file-statuses res))))
  +  (let* ((res (send-operation-request context :get "LISTSTATUS" path))
  +         (statuses (alist-get (alist-get res :*file-statuses) :*file-status)))
       (iter (for status in statuses)
             (collect
  -              (let ((mtime (alist-get :modification-time status))
  -                    (atime (alist-get :access-time status))
  -                    (type  (alist-get :type status)))
  +              (let ((mtime (alist-get status :modification-time))
  +                    (atime (alist-get status :access-time))
  +                    (type  (alist-get status :type)))
                   (alist-merge `(,status
                                  ((:modification-time . ,(from-hadoop-time mtime))
                                   (:access-time . ,(from-hadoop-time atime))
                                   (:type . ,(make-keyword type))))))))))
   
   (defun get-file-status (context path)
  -  (let ((res (operation-request context :get "GETFILESTATUS" path)))
  -    (alist-get :*file-status res)))
  +  (let ((response (send-operation-request context :get "GETFILESTATUS" path)))
  +    (alist-get response :*file-status)))
   
   (defun mkdirs (context path &key permission)
  -  (let ((res (operation-request context :put "MKDIRS" path
  -                                :parameters `(("permission" . ,permission)))))
  -    (alist-get :boolean res)))
  +  (bind ((parameters `(("permission" . ,permission)))
  +         (response (operation-request context :put "MKDIRS" path
  +                                      :parameters parameters)))
  +    (alist-get response :boolean)))
   
   (defun delete-file-or-dir (context path &key recursive)
  -  (let ((res (operation-request context :delete "DELETE" path
  -                                :parameters `(("recursive" . ,(if recursive "true" "false"))))))
  -    (alist-get :boolean res)))
  +  (let* ((parameters `(("recursive" . ,(if recursive "true" "false"))))
  +         (response (send-operation-request context :delete "DELETE" path
  +                                           :parameters parameters
  +                                           :response-handler #'ignore-response)))
  +    (alist-get response :boolean)))
  
positive.code: (defpackage :nellie.hdfs
    (:nicknames #:hdfs)
    (:use :cl :alists :simple-date-time :iterate :alexandria)
    (:export #:+webfs-port+
             #:+webhdfs-port+
             #:*default-port*
  
negative.code:   (user "hdfs"))
  
  (define-condition server-error (error)
    ((type :initarg :type :reader error-type)
     (code :initarg :code :reader error-code)
     (message :initarg :message :reader error-message)))
  
  (defmethod print-object ((object server-error) stream)
    (print-unreadable-object (object stream :type t :identity t)
      (format stream "[~D ~A]: ~A"
              (error-type object)
              (error-code object)
              (error-message object))))
  
  (defun server-error (type code message)
    (error 'server-error
  