path: root/stdlib/source/library/lux/world/net/http/client.lux

(.require
 [library
  [lux (.except)
   ["[0]" ffi]
   [abstract
    ["[0]" monad (.only do)]]
   [control
    ["[0]" pipe]
    ["[0]" io (.only IO)]
    ["[0]" maybe (.use "[1]#[0]" functor)]
    ["[0]" try (.only Try)]
    [concurrency
     ["[0]" async (.only Async)]]]
   [data
    ["[0]" binary (.only Binary)]
    ["[0]" text]
    [collection
     ["[0]" dictionary]]]
   [math
    [number
     ["n" nat]
     ["i" int]]]
   [meta
    ["@" target]
    ["[0]" code (.only)
     ["<[1]>" \\parser]]
    [macro
     [syntax (.only syntax)]
     ["[0]" template]]]]]
 ["[0]" // (.only)
  [response (.only Response)]
  ["[0]" header (.only Headers)]
  [// (.only URL)]])

(type .public (Client !)
  (Interface
   (is (-> //.Method URL Headers (Maybe Binary)
           (! (Try (Response !))))
       request)))

(def method_function
  (syntax (_ [[_ name] <code>.symbol])
    (in (list (code.local (text.replaced "#" "" (text.lower_cased name)))))))

(with_template [<method>]
  [(with_expansions [<name> (method_function <method>)]
     (def .public (<name> url headers data client)
       (All (_ !)
         (-> URL Headers (Maybe Binary) (Client !)
             (! (Try (Response !)))))
       (at client request {<method>} url headers data)))]

  [//.#Post]
  [//.#Get]
  [//.#Put]
  [//.#Patch]
  [//.#Delete]
  [//.#Head]
  [//.#Connect]
  [//.#Options]
  [//.#Trace]
  )

(def default_buffer_size
  (n.* 1,024 1,024))

(def empty_body
  [Nat Binary]
  [0 (binary.empty 0)])

(def (body_of data)
  (-> Binary [Nat Binary])
  [(binary.size data) data])

(with_expansions [<jvm> (these (ffi.import java/lang/String
                                 "[1]::[0]")

                               (ffi.import java/lang/AutoCloseable
                                 "[1]::[0]"
                                 (close [] "io" "try" void))

                               (ffi.import java/io/InputStream
                                 "[1]::[0]")

                               (ffi.import java/io/OutputStream
                                 "[1]::[0]"
                                 (flush [] "io" "try" void)
                                 (write [[byte]] "io" "try" void))

                               (ffi.import java/net/URLConnection
                                 "[1]::[0]"
                                 (setDoOutput [boolean] "io" "try" void)
                                 (setRequestProperty [java/lang/String java/lang/String] "io" "try" void)
                                 (getInputStream [] "io" "try" java/io/InputStream)
                                 (getOutputStream [] "io" "try" java/io/OutputStream)
                                 (getHeaderFieldKey [int] "io" "try" "?" java/lang/String)
                                 (getHeaderField [int] "io" "try" "?" java/lang/String))

                               (ffi.import java/net/HttpURLConnection
                                 "[1]::[0]"
                                 (setRequestMethod [java/lang/String] "io" "try" void)
                                 (getResponseCode [] "io" "try" int))

                               (ffi.import java/net/URL
                                 "[1]::[0]"
                                 (new [java/lang/String])
                                 (openConnection [] "io" "try" java/net/URLConnection))

                               (ffi.import java/io/BufferedInputStream
                                 "[1]::[0]"
                                 (new [java/io/InputStream])
                                 (read [[byte] int int] "io" "try" int))

                               (def jvm_method
                                 (-> //.Method Text)
                                 (|>> (pipe.when
                                        {//.#Post} "POST"
                                        {//.#Get} "GET"
                                        {//.#Put} "PUT"
                                        {//.#Patch} "PATCH"
                                        {//.#Delete} "DELETE"
                                        {//.#Head} "HEAD"
                                        {//.#Connect} "CONNECT"
                                        {//.#Options} "OPTIONS"
                                        {//.#Trace} "TRACE")))

                               (def (default_body input)
                                 (-> java/io/BufferedInputStream (//.Body IO))
                                 (|>> (maybe#each (|>> [true]))
                                      (maybe.else [false ..default_buffer_size])
                                      (pipe.when
                                        [_ 0]
                                        (do (try.with io.monad)
                                          [_ (java/lang/AutoCloseable::close input)]
                                          (in ..empty_body))

                                        [partial? buffer_size]
                                        (let [buffer (binary.empty buffer_size)]
                                          (if partial?
                                            (loop (again [so_far +0])
                                              (do [! (try.with io.monad)]
                                                [.let [remaining (i.- so_far (.int buffer_size))]
                                                 bytes_read (at ! each (|>> ffi.of_int)
                                                                (java/io/BufferedInputStream::read buffer (ffi.as_int so_far) (ffi.as_int remaining) input))]
                                                (when bytes_read
                                                  -1 (do !
                                                       [_ (java/lang/AutoCloseable::close input)]
                                                       (in [(.nat so_far) buffer]))
                                                  +0 (again so_far)
                                                  _ (if (i.= remaining bytes_read)
                                                      (in [buffer_size buffer])
                                                      (again (i.+ bytes_read so_far))))))
                                            (loop (again [so_far +0
                                                          output (at binary.monoid identity)])
                                              (do [! (try.with io.monad)]
                                                [.let [remaining (i.- so_far (.int buffer_size))]
                                                 bytes_read (at ! each (|>> ffi.of_int)
                                                                (java/io/BufferedInputStream::read buffer (ffi.as_int so_far) (ffi.as_int remaining) input))]
                                                (when bytes_read
                                                  -1 (do !
                                                       [_ (java/lang/AutoCloseable::close input)]
                                                       (when so_far
                                                         +0 (in (..body_of output))
                                                         _ (|> buffer
                                                               (binary.slice 0 (.nat so_far))
                                                               (at try.functor each
                                                                   (|>> (at binary.monoid composite output)
                                                                        ..body_of))
                                                               (at io.monad in))))
                                                  +0 (again so_far output)
                                                  _ (if (i.= remaining bytes_read)
                                                      (again +0
                                                             (at binary.monoid composite output buffer))
                                                      (again (i.+ bytes_read so_far)
                                                             output))))))))))

                               (def (default_headers connection)
                                 (-> java/net/HttpURLConnection (IO (Try Headers)))
                                 (loop (again [index +0
                                               headers header.empty])
                                   (do [! (try.with io.monad)]
                                     [?name (java/net/URLConnection::getHeaderFieldKey (ffi.as_int index) connection)]
                                     (when ?name
                                       {.#Some name}
                                       (do !
                                         [?value (java/net/URLConnection::getHeaderField (ffi.as_int index) connection)]
                                         (again (++ index)
                                                (dictionary.has (ffi.of_string name) (maybe.else "" (maybe#each (|>> ffi.of_string) ?value)) headers)))

                                       {.#None}
                                       (in headers)))))

                               (def .public default
                                 (Client IO)
                                 (implementation
                                  (def (request method url headers data)
                                    (is (IO (Try (Response IO)))
                                        (do [! (try.with io.monad)]
                                          [connection (|> url ffi.as_string java/net/URL::new java/net/URL::openConnection)
                                           .let [connection (as java/net/HttpURLConnection connection)]
                                           _ (java/net/HttpURLConnection::setRequestMethod (ffi.as_string (..jvm_method method)) connection)
                                           _ (monad.each ! (function (_ [name value])
                                                             (java/net/URLConnection::setRequestProperty (ffi.as_string name) (ffi.as_string value) connection))
                                                         (dictionary.entries headers))
                                           _ (when data
                                               {.#Some data}
                                               (do !
                                                 [_ (java/net/URLConnection::setDoOutput true connection)
                                                  stream (java/net/URLConnection::getOutputStream connection)
                                                  _ (java/io/OutputStream::write data stream)
                                                  _ (java/io/OutputStream::flush stream)
                                                  _ (java/lang/AutoCloseable::close stream)]
                                                 (in []))
                                               
                                               {.#None}
                                               (in []))
                                           status (java/net/HttpURLConnection::getResponseCode connection)
                                           headers (..default_headers connection)
                                           input (|> connection
                                                     java/net/URLConnection::getInputStream
                                                     (at ! each (|>> java/io/BufferedInputStream::new)))]
                                          (in [(.nat (ffi.of_int status))
                                               [//.#headers headers
                                                //.#body (..default_body input)]])))))))]
  (for @.old (these <jvm>)
       @.jvm (these <jvm>)
       (these)))

(def .public (async client)
  (-> (Client IO) (Client Async))
  (implementation
   (def (request method url headers data)
     (|> (at client request method url headers data)
         async.future
         (at async.monad each
             (|>> (pipe.when
                    {try.#Success [status message]}
                    {try.#Success [status (revised //.#body (is (-> (//.Body IO) (//.Body Async))
                                                                (function (_ body)
                                                                  (|>> body async.future)))
                                                   message)]}
                    
                    {try.#Failure error}
                    {try.#Failure error})))))))

(def .public headers
  (-> (List [Text Text]) Headers)
  (dictionary.of_list text.hash))