anchor.diff:    (user "hdfs"))
   
   (define-condition server-error (error)
  -  ((type :initarg :type :reader server-error-type)
  -   (code :initarg :code :reader server-error-code)
  -   (message :initarg :message :reader server-error-message)))
  +  ((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"
  -            (server-error-type object)
  -            (server-error-code object)
  -            (server-error-message object))))
  +            (error-type object)
  +            (error-code object)
  +            (error-message object))))
   
   (defun server-error (type code message)
     (error 'server-error
  
positive.code:                       (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)))
  
  (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))))
      (iter (for status in statuses)
            (collect
                (let ((mtime (alist-get :modification-time status))
                      (atime (alist-get :access-time status))
                      (type  (alist-get :type status)))
                  (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)))
  
  (defun mkdirs (context path &key permission)
    (let ((res (operation-request context :put "MKDIRS" path
                                  :parameters `(("permission" . ,permission)))))
      (alist-get :boolean res)))
  
  (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)))
  
negative.code: (defpackage :nellie.hdfs
    (:nicknames #:hdfs)
    (:use :cl :alists :simple-date-time :iterate :alexandria)
    (:import-from :metabang-bind :bind)
    (:import-from :flexi-streams
                  :make-flexi-stream :octets-to-string
                  :with-output-to-sequence :octet)
    (:export #:+webfs-port+
             #:+webhdfs-port+
             #:*default-port*
  