servant-streaming-server (empty) → 0.2.0.0
raw patch · 7 files changed
+419/−0 lines, 7 filesdep +QuickCheckdep +basedep +bytestringsetup-changed
Dependencies added: QuickCheck, base, bytestring, directory, hspec, http-media, http-types, pipes, pipes-http, resourcet, servant, servant-server, servant-streaming, servant-streaming-server, streaming, streaming-bytestring, streaming-wai, wai, warp
Files
- LICENSE +31/−0
- Setup.hs +2/−0
- servant-streaming-server.cabal +78/−0
- src/Servant/Streaming/Server.hs +16/−0
- src/Servant/Streaming/Server/Internal.hs +93/−0
- test/Servant/Streaming/ServerSpec.hs +198/−0
- test/Spec.hs +1/−0
+ LICENSE view
@@ -0,0 +1,31 @@+Copyright Julian K. Arni (c) 2015++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Julian K. Arni nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ servant-streaming-server.cabal view
@@ -0,0 +1,78 @@+-- This file has been generated from package.yaml by hpack version 0.20.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 2c782b81db834803cac4a8b8aeffb9f7e765d9e4097f1acee1d86460c5a266fd++name: servant-streaming-server+version: 0.2.0.0+synopsis: Server instances for the 'servant-streaming' package.+description: This package defines instances that allow using the 'StreamBody' and 'StreamResponse' combinators in 'servant' servers.+homepage: http://github.com/plow-technologies/servant-streaming-server#readme+bug-reports: https://github.com/plow-technologies/servant-streaming-server/issues+author: Julian K. Arni+maintainer: jkarni@gmail.com+copyright: (c) Julian K. Arni+license: BSD3+license-file: LICENSE+tested-with: GHC == 8.2.2+build-type: Simple+cabal-version: >= 1.10++source-repository head+ type: git+ location: https://github.com/plow-technologies/servant-streaming-server++library+ hs-source-dirs:+ src+ default-extensions: AutoDeriveTypeable ConstraintKinds DataKinds DefaultSignatures DeriveFoldable DeriveFunctor DeriveGeneric DeriveTraversable FlexibleContexts FlexibleInstances FunctionalDependencies GADTs KindSignatures MultiParamTypeClasses OverloadedStrings RankNTypes ScopedTypeVariables TypeFamilies TypeOperators+ ghc-options: -Wall+ build-depends:+ base >=4.7 && <4.11+ , bytestring+ , http-media >=0.6 && <0.8+ , http-types >=0.9 && <0.13+ , resourcet >=1.1 && <1.3+ , servant+ , servant-server >=0.8 && <0.14+ , servant-streaming >=0.2 && <0.3+ , streaming >=0.1 && <0.3+ , streaming-wai >=0.1 && <0.2+ , wai >=3.0 && <3.3+ exposed-modules:+ Servant.Streaming.Server+ Servant.Streaming.Server.Internal+ default-language: Haskell2010++test-suite spec+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ hs-source-dirs:+ test+ default-extensions: AutoDeriveTypeable ConstraintKinds DataKinds DefaultSignatures DeriveFoldable DeriveFunctor DeriveGeneric DeriveTraversable FlexibleContexts FlexibleInstances FunctionalDependencies GADTs KindSignatures MultiParamTypeClasses OverloadedStrings RankNTypes ScopedTypeVariables TypeFamilies TypeOperators+ ghc-options: -Wall -Wall -with-rtsopts=-T -threaded+ build-depends:+ QuickCheck >=2.8 && <2.11+ , base >=4.7 && <4.11+ , bytestring+ , directory+ , hspec >2 && <3+ , http-media >=0.6 && <0.8+ , http-types >=0.9 && <0.13+ , pipes >=4 && <5+ , pipes-http >=1 && <2+ , resourcet >=1.1 && <1.3+ , servant+ , servant-server >=0.8 && <0.14+ , servant-streaming >=0.2 && <0.3+ , servant-streaming-server+ , streaming >=0.1 && <0.3+ , streaming-bytestring >=0.1 && <0.2+ , streaming-wai >=0.1 && <0.2+ , wai >=3.0 && <3.3+ , warp >=3.2.4 && <3.3+ other-modules:+ Servant.Streaming.ServerSpec+ Paths_servant_streaming_server+ default-language: Haskell2010
+ src/Servant/Streaming/Server.hs view
@@ -0,0 +1,16 @@+-- | This module provides an instance for HasServant for 'servant-streaming'+-- combinators, so that servers with streaming (requests or responses) may be+-- used.+--+-- As as convenience, it also re-exports the combinators themselves.+module Servant.Streaming.Server+ ( StreamResponse+ , StreamBody+ , StreamResponseGet+ , StreamResponsePost+ , StreamResponsePut+ , StreamResponsePatch+ ) where++import Servant.Streaming+import Servant.Streaming.Server.Internal ()
+ src/Servant/Streaming/Server/Internal.hs view
@@ -0,0 +1,93 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Servant.Streaming.Server.Internal where++import Control.Exception (bracket)+import Control.Monad.IO.Class+import Control.Monad.Trans.Resource (ResourceT,+ InternalState,+ createInternalState,+ closeInternalState,+ runInternalState)+import qualified Data.ByteString as BS+import Data.Maybe (fromMaybe)+import GHC.TypeLits (KnownNat, natVal)+import qualified Network.HTTP.Media as M+import Network.HTTP.Types (Method, Status,+ hAccept,+ hContentType)+import Network.Wai (Response,+ requestHeaders)+import Network.Wai.Streaming (streamingRequest,+ streamingResponse)+import Servant hiding (Stream)+import Servant.API.ContentTypes (AllMime (allMime))+import Servant.Server.Internal (ct_wildcard,+ methodCheck,+ acceptCheck)+import Servant.Server.Internal.Router (leafRouter)+import Servant.Server.Internal.RoutingApplication (DelayedIO,+ RouteResult (Route),+ addBodyCheck,+ addMethodCheck,+ addAcceptCheck,+ delayedFailFatal,+ runAction,+ withRequest)+import Servant.Streaming+import Streaming++instance ( AllMime contentTypes, HasServer subapi ctx, MonadIO n+ ) => HasServer (StreamBodyMonad contentTypes n :> subapi) ctx where+ type ServerT (StreamBodyMonad contentTypes n :> subapi) m+ = (M.MediaType, Stream (Of BS.ByteString) (ResourceT n) ())+ -> ServerT subapi m++ route _ ctxt subapi =+ route (Proxy :: Proxy subapi) ctxt+ $ addBodyCheck subapi getContentType makeBody+ where+ getContentType :: DelayedIO M.MediaType+ getContentType = withRequest $ \request -> do+ let contentTypeHdr+ = fromMaybe ("application" M.// "octet-stream")+ $ lookup hContentType (requestHeaders request) >>= M.parseAccept+ if contentTypeHdr `elem` contentTypeList+ then return contentTypeHdr else delayedFailFatal err415++ contentTypeList :: [M.MediaType]+ contentTypeList = allMime (Proxy :: Proxy contentTypes)++ makeBody :: MonadIO m => a -> DelayedIO (a, Stream (Of BS.ByteString) m ())+ makeBody a = withRequest $ \req -> return (a, streamingRequest req)++ hoistServerWithContext _ a b c+ = hoistServerWithContext (Proxy :: Proxy subapi) a b . c+++instance ( KnownNat status, AllMime contentTypes, ReflectMethod method+ ) => HasServer (StreamResponse method status contentTypes) ctx where+ type ServerT (StreamResponse method status contentTypes) m+ = m (Stream (Of BS.ByteString) (ResourceT IO) ())++ route _ _ctxt subapi = leafRouter $ \env request respond ->+ let action = subapi `addMethodCheck` methodCheck method request+ `addAcceptCheck` acceptCheck contentTypeProxy accept+ accept = fromMaybe ct_wildcard $ lookup hAccept $ requestHeaders request+ in bracket createInternalState+ closeInternalState+ (runAction action env request respond . streamResponse)+ where+ method :: Method+ method = reflectMethod (Proxy :: Proxy method)++ contentTypeProxy :: Proxy contentTypes+ contentTypeProxy = Proxy++ streamResponse :: InternalState -> Stream (Of BS.ByteString) (ResourceT IO) () -> RouteResult Response+ streamResponse st stream = Route $ streamingResponse (hoist (`runInternalState` st) stream) status []++ status :: Status+ status = toEnum $ fromInteger $ natVal (Proxy :: Proxy status)++ hoistServerWithContext _ _ b c+ = b c
+ test/Servant/Streaming/ServerSpec.hs view
@@ -0,0 +1,198 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_GHC -fno-warn-type-defaults #-}+module Servant.Streaming.ServerSpec (spec) where++import Control.Concurrent+import Control.Exception (bracket)+import Control.Monad.Trans.Resource (ResourceT, getInternalState,+ runInternalState, runResourceT)+import Data.ByteString (ByteString)+import qualified Data.ByteString.Streaming as BSS+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL++import Data.IORef+import Data.String (fromString, IsString)+import GHC.Stats+import qualified Network.HTTP.Media as M+import Network.HTTP.Types (status200, status405, status406,+ status415)+import Network.Wai.Handler.Warp+import Pipes.HTTP (Request, RequestBody (..),+ Response, brRead,+ defaultManagerSettings,+ defaultRequest, httpLbs, method,+ newManager, path, port,+ requestBody, requestHeaders,+ responseBody, responseStatus,+ responseTimeout,+ responseTimeoutNone,+ withResponse)+import Servant ((:<|>) ((:<|>)), (:>), JSON,+ MimeRender (..), PlainText, Post,+ Proxy (..), Server, serve)+import Servant.Streaming.Server+import Streaming+import qualified Streaming.Prelude as S+import System.Directory (removeFile)+import Test.Hspec+++spec :: Spec+spec = do+ streamBodySpec+ streamResponseSpec++streamBodySpec :: Spec+streamBodySpec = describe "StreamBody instance" $ around withServer $ do++ it "streams the request body" $ \port' -> do+ let req = streamReq port' "length" (S.each ["h","i"])+ responseBody <$> makeRequest req `shouldReturn` "2"++ it "does not keep the request in memory" $ \port' -> do+ let req = streamReq port' "length"+ $ S.replicate megabyte+ $ BS.replicate 1000 97 -- 1000 MB total+ responseBody <$> makeRequest req+ `shouldReturn` fromString (show (1000 * megabyte :: Int))+ bytes <- max_live_bytes <$> getRTSStats+ bytes < 200 * megabyte `shouldBe` True++ it "passes as argument the content-type" $ \port' -> do+ let req = streamReq port' "contentType" (S.each ["h","i"])+ responseBody <$> makeRequest req `shouldReturn` "application/json"+ let req' = req { requestHeaders = [("Content-Type", "text/plain;charset=utf-8")] }+ responseBody <$> makeRequest req' `shouldReturn` "text/plain;charset=utf-8"++ it "responds with '415 - Unsupported Media Type' on wrong content type" $ \port' -> do+ let req' = streamReq port' "length" (S.each ["h","i"])+ req = req' { requestHeaders = [("Content-Type", "bla/bla")] }+ responseStatus <$> makeRequest req `shouldReturn` status415++streamResponseSpec :: Spec+streamResponseSpec = describe "StreamResponse instance" $ around withServer $ do++ it "streams the response body" $ \port' -> do+ let req = streamReq port' "echo" (S.each ["h","i"])+ responseBody <$> makeRequest req `shouldReturn` "hi"++ it "does not keep the response in memory" $ \port' -> do+ let req = streamReq port' "echo"+ $ S.replicate 100+ $ BS.replicate megabyte 97 -- 100 MB total+ _ <- makeRequestStreamResponse req (runResourceT . S.length)+ bytes <- max_live_bytes <$> getRTSStats+ bytes < 200 * megabyte `shouldBe` True++ it "sets the specified status code" $ \port' -> do+ let req = streamReq port' "length" (S.each ["h","i"])+ responseStatus <$> makeRequest req `shouldReturn` status200++ it "responds with '405 - Method Not Allowed' on wrong method" $ \port' -> do+ let req' = streamReq port' "echo" (S.each ["h","i"])+ req = req' { method = "GET" }+ responseStatus <$> makeRequest req `shouldReturn` status405++ it "responds with '406 - Not Acceptable' on wrong content type" $ \port' -> do+ let req' = streamReq port' "echo" (S.each ["h","i"])+ req = req' { requestHeaders = ("Accept", "bla/bla"):requestHeaders req' }+ responseStatus <$> makeRequest req `shouldReturn` status406++ it "handles resource deallocation correctly" $ \port' -> do+ let req' = streamReq port' "getfile" (S.each []) -- TODO: simplify+ contents :: IsString a => a+ contents = "foobar"+ x <- bracket+ (BS.writeFile "hello.txt" contents)+ (\_ -> removeFile "hello.txt")+ (\_ -> responseBody <$> makeRequest req')+ x `shouldBe` fromString contents+++------------------------------------------------------------------------------+-- API++type API+ = "length" :> StreamBody '[JSON] :> Post '[PlainText] Int+ :<|> "contentType" :> StreamBody '[JSON, PlainText] :> Post '[PlainText] M.MediaType+ :<|> "echo" :> StreamBody '[JSON] :> StreamResponsePost '[JSON]+ :<|> "getfile" :> StreamResponsePost '[PlainText]++api :: Proxy API+api = Proxy++server :: Server API+server = lengthH :<|> contentTypeH :<|> echoH :<|> getfileH+ where+ lengthH (_contentType, stream')+ = liftIO . runResourceT $ S.sum_ $ S.subst (\x -> BS.length x :> ()) stream'+ contentTypeH (contentType, _stream')+ = return contentType+ echoH (_contentType, stream')+ = return stream'+ getfileH = return $ BSS.toChunks (BSS.readFile "hello.txt")++withServer :: (Port -> IO ()) -> IO ()+withServer = withApplicationSettings settings (return $ serve api server)+ where+ settings = setTimeout 60 defaultSettings++------------------------------------------------------------------------------+-- Utils++makeRequest+ :: Request -> IO (Response BSL.ByteString)+makeRequest req = do+ manager <- newManager defaultManagerSettings+ httpLbs req manager++makeRequestStreamResponse+ :: Request -> (Stream (Of ByteString) (ResourceT IO) () -> IO r) -> IO r+makeRequestStreamResponse req responseAction = do+ manager <- newManager defaultManagerSettings+ withResponse req manager (responseAction . go . responseBody)+ where+ go :: MonadIO m => IO BS.ByteString -> Stream (Of ByteString) m ()+ go respReader = do+ n <- liftIO $ brRead respReader+ if BS.null n+ then return ()+ else do+ S.yield n+ go respReader++streamReq+ :: Port -> ByteString -> Stream (Of ByteString) (ResourceT IO) () -> Request+streamReq appPort urlPath requestStream = defaultRequest+ { port = appPort+ , requestHeaders = [("Content-Type", "application/json")]+ , method = "POST"+ , path = urlPath+ , requestBody = streamReqBody requestStream+ , responseTimeout = responseTimeoutNone+ }+ where+ streamReqBody body+ = RequestBodyStreamChunked body'+ where+ body' :: (IO BS.ByteString -> IO ()) -> IO ()+ body' write = void . forkIO . runResourceT $ do+ ref <- liftIO $ newIORef body+ is <- getInternalState+ let popper :: IO BS.ByteString+ popper = do+ rsrc <- readIORef ref+ mres <- runInternalState (S.uncons rsrc) is+ case mres of+ Nothing -> return BS.empty+ Just (bs, str)+ | BS.null bs -> writeIORef ref str >> popper+ | otherwise -> writeIORef ref str >> return bs+ liftIO $ write popper++instance Show a => MimeRender PlainText a where+ mimeRender _ = fromString . show++megabyte :: Num a => a+megabyte = 1000 ^ 2
+ test/Spec.hs view
@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}