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: (defpackage :nellie.hdfs
    (:nicknames #:hdfs)
    (:use :cl :alists :simple-date-time :iterate :alexandria)
    (:export #:+webfs-port+
             #:+webhdfs-port+
             #:*default-port*
  
negative.code:                       (context-host context)
                        (context-port context)
                        path))
           (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 list-status (context path)
    (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 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 ((response (send-operation-request context :get "GETFILESTATUS" path)))
      (alist-get response :*file-status)))
  
  (defun mkdirs (context path &key permission)
    (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* ((parameters `(("recursive" . ,(if recursive "true" "false"))))
           (response (send-operation-request context :delete "DELETE" path
                                             :parameters parameters
                                             :response-handler #'ignore-response)))
      (alist-get response :boolean)))
  