Add streamBody for requests

Author: owickstrom

examples/NodeStreamRequest.purs

@@ -0,0 +1,67 @@
+-- This example shows how you can stream request body data. It
+-- logs the size of each chunk it receives from the POST body.
+--
+-- Test it out by first running the server,
+--
+--     $ pulp run -I examples -m Examples.NodeStreamRequest
+--
+-- and then, POST a large file with something like this command:
+--
+--     $ curl -X POST --data-binary @/your/large/file localhost:3000
+--
+module Examples.NodeStreamRequest where
+
+import Prelude
+import Node.Buffer as Buffer
+import Node.Stream as Stream
+import Control.IxMonad (ibind, (:>>=))
+import Control.Monad.Eff (Eff)
+import Control.Monad.Eff.Class (class MonadEff, liftEff)
+import Control.Monad.Eff.Console (CONSOLE, log)
+import Control.Monad.Eff.Exception (EXCEPTION, catchException, message)
+import Data.Either (Either(..), either)
+import Data.HTTP.Method (Method(..))
+import Hyper.Node.Server (defaultOptionsWithLogging, runServer)
+import Hyper.Request (getRequestData, streamBody)
+import Hyper.Response (closeHeaders, respond, writeStatus)
+import Hyper.Status (statusMethodNotAllowed, statusOK)
+import Node.Buffer (BUFFER)
+import Node.HTTP (HTTP)
+
+type ExampleEffects e = (http :: HTTP, console :: CONSOLE, buffer :: BUFFER | e)
+
+logRequestBodyChunks
+  :: forall m e
+   . MonadEff (ExampleEffects e) m
+  => Stream.Readable () (ExampleEffects (exception :: EXCEPTION | e))
+  -> m Unit
+logRequestBodyChunks body =
+  Stream.onData body (Buffer.size >=> (log <<< ("Got chunk of size: " <> _) <<< show))
+  # catchException (log <<< ("Error: " <> _) <<< message)
+  # liftEff
+
+main :: forall e. Eff (ExampleEffects e) Unit
+main =
+  let
+    app =
+      getRequestData :>>=
+      case _ of
+
+        -- Only handle POST requests:
+        { method: Left POST } -> do
+            body <- streamBody
+            logRequestBodyChunks body
+            writeStatus statusOK
+            closeHeaders
+            respond "OK"
+
+        -- Non-POST requests are not allowed:
+        { method } -> do
+          writeStatus statusMethodNotAllowed
+          closeHeaders
+          respond ("Method not allowed: " <> either show show method)
+
+        where
+            bind = ibind
+            discard = ibind
+  in runServer defaultOptionsWithLogging {} app

src/Hyper/Node/Server.purs

@@ -24,7 +24,7 @@ import Control.Monad.Aff.Class (class MonadAff, liftAff)
 import Control.Monad.Eff (Eff)
 import Control.Monad.Eff.Class (class MonadEff, liftEff)
 import Control.Monad.Eff.Console (CONSOLE, log)
-import Control.Monad.Eff.Exception (Error, catchException, error)
+import Control.Monad.Eff.Exception (EXCEPTION, Error, catchException, error)
 import Control.Monad.Error.Class (throwError)
 import Data.Either (Either(..), either)
 import Data.Maybe (Maybe(..))
@@ -34,13 +34,13 @@ import Hyper.Conn (Conn)
 import Hyper.Middleware (Middleware, evalMiddleware, lift')
 import Hyper.Middleware.Class (getConn, modifyConn)
 import Hyper.Port (Port(..))
-import Hyper.Request (class ReadableBody, class Request, RequestData, readBody)
+import Hyper.Request (class ReadableBody, class Request, class StreamableBody, RequestData, readBody)
 import Hyper.Response (class ResponseWritable, class Response, ResponseEnded, StatusLineOpen)
 import Hyper.Status (Status(..))
 import Node.Buffer (BUFFER, Buffer)
 import Node.Encoding (Encoding(..))
 import Node.HTTP (HTTP)
-import Node.Stream (Writable)
+import Node.Stream (Stream, Writable)
 
 
 data HttpRequest
@@ -128,6 +128,16 @@ instance readableBodyHttpRequestBuffer :: (Monad m, MonadAff (http :: HTTP, avar
     case _ of
       r -> liftAff (readBodyAsBuffer r)
 
+instance streamableBodyHttpRequestReadable :: MonadAff (http :: HTTP | e) m
+                                           => StreamableBody
+                                              HttpRequest
+                                              m
+                                              (Stream (read :: Stream.Read) (http :: HTTP, exception :: EXCEPTION | e)) where
+  streamBody =
+    _.request <$> getConn :>>=
+    case _ of
+      HttpRequest request _ -> ipure (HTTP.requestAsStream request)
+
 -- TODO: Make a newtype
 data HttpResponse state = HttpResponse HTTP.Response
 

src/Hyper/Request.purs

@@ -5,6 +5,8 @@ module Hyper.Request
   , class BaseRequest
   , class ReadableBody
   , readBody
+  , class StreamableBody
+  , streamBody
   ) where
 
 import Data.Either (Either)
@@ -32,8 +34,9 @@ class Request req m where
 
 class Request req m <= BaseRequest req m
 
--- | A ReadableBody instance reads the request body for a specific body
--- | type.
+-- | A `ReadableBody` instance reads the complete request body as a
+-- | value of type `b`. For streaming the request body, see the
+-- | [StreamableBody](#streamablebody) class.
 class ReadableBody req m b where
   readBody
     :: forall res c
@@ -42,3 +45,15 @@ class ReadableBody req m b where
        (Conn req res c)
        (Conn req res c)
        b
+
+-- | A `StreamableBody` instance returns a stream of the request body,
+-- | of type `stream`. To read the whole body as a value, without
+-- | streaming, see the [ReadableBody](#readablebody) class.
+class StreamableBody req m stream | req -> stream where
+  streamBody
+    :: forall res c
+     . Middleware
+       m
+       (Conn req res c)
+       (Conn req res c)
+       stream