eeva/servant
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Refactor Stream stuff

Browse source 
- Introduce SourceT, which is simple variant of "correct `ListT`".
  There are another variants possible (like in `streaming`),
  but I'm not sure there's much real difference.

- Introduce `Codensity`. There's a flag if people don't want to depend
  on `kan-extensions`.

- `StreamGenerator` and `ResultStream` are both `SourceT`.
  `Stream` combinator in `servant-client` uses `Codensity` for CPS.

- Add servant-machines, servant-conduit, servant-pipes
- Add streaming cookbook: just code, no explanations.
- Add a script to run streaming 'benchmarks'
This commit is contained in:
Oleg Grenrus 2018-06-26 20:11:28 +03:00
parent 79bbcaf819
commit 45c1cbdfd5
53 changed files with 2183 additions and 381 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -29,3 +29,6 @@ doc/_build
doc/venv
doc/tutorial/static/api.js
doc/tutorial/static/jq.js
# local versions of things
servant-multipart

22
.travis.yml
View file

@ -69,11 +69,11 @@ install:
- rm -fv cabal.project cabal.project.local
- "if [ $HCNUMVER -ge 70800 ]; then sed -i.bak 's/-- ghc-options:.*/ghc-options: -j2/' ${HOME}/.cabal/config; fi"
- grep -Ev -- '^\s*--' ${HOME}/.cabal/config | grep -Ev '^\s*$'
- "printf 'packages: \"servant\" \"servant-client\" \"servant-client-core\" \"servant-docs\" \"servant-foreign\" \"servant-server\" \"doc/tutorial\" \"doc/cookbook/basic-auth\" \"doc/cookbook/curl-mock\" \"doc/cookbook/db-postgres-pool\" \"doc/cookbook/db-sqlite-simple\" \"doc/cookbook/file-upload\" \"doc/cookbook/generic\" \"doc/cookbook/https\" \"doc/cookbook/sentry\" \"doc/cookbook/testing\" \"doc/cookbook/structuring-apis\" \"doc/cookbook/using-custom-monad\" \"doc/cookbook/using-free-client\"\\n' > cabal.project"
- "printf 'packages: \"servant\" \"servant-client\" \"servant-client-core\" \"servant-docs\" \"servant-foreign\" \"servant-server\" \"doc/tutorial\" \"servant-machines\" \"servant-conduit\" \"servant-pipes\" \"doc/cookbook/basic-auth\" \"doc/cookbook/curl-mock\" \"doc/cookbook/basic-streaming\" \"doc/cookbook/db-postgres-pool\" \"doc/cookbook/db-sqlite-simple\" \"doc/cookbook/file-upload\" \"doc/cookbook/generic\" \"doc/cookbook/https\" \"doc/cookbook/sentry\" \"doc/cookbook/testing\" \"doc/cookbook/structuring-apis\" \"doc/cookbook/using-custom-monad\" \"doc/cookbook/using-free-client\"\\n' > cabal.project"
- "echo 'constraints: foundation >=0.0.14,memory <0.14.12 || >0.14.12' >> cabal.project"
- "echo 'allow-newer: servant-auth-server:http-types,servant-auth-server:servant-server,servant-js:base, servant-pagination:servant,servant-pagination:servant-server' >> cabal.project"
- touch cabal.project.local
- "if ! $NOINSTALLEDCONSTRAINTS; then for pkg in $($HCPKG list --simple-output); do echo $pkg | grep -vw -- servant | grep -vw -- servant-client | grep -vw -- servant-client-core | grep -vw -- servant-docs | grep -vw -- servant-foreign | grep -vw -- servant-server | grep -vw -- tutorial | grep -vw -- cookbook-basic-auth | grep -vw -- cookbook-curl-mock | grep -vw -- cookbook-db-postgres-pool | grep -vw -- cookbook-db-sqlite-simple | grep -vw -- cookbook-file-upload | grep -vw -- cookbook-generic | grep -vw -- cookbook-https | grep -vw -- cookbook-sentry | grep -vw -- cookbook-testing | grep -vw -- cookbook-structuring-apis | grep -vw -- cookbook-using-custom-monad | grep -vw -- cookbook-using-free-client | sed 's/^/constraints: /' | sed 's/-[^-]*$/ installed/' >> cabal.project.local; done; fi"
- "if ! $NOINSTALLEDCONSTRAINTS; then for pkg in $($HCPKG list --simple-output); do echo $pkg | grep -vw -- servant | grep -vw -- servant-client | grep -vw -- servant-client-core | grep -vw -- servant-docs | grep -vw -- servant-foreign | grep -vw -- servant-server | grep -vw -- tutorial | grep -vw -- servant-machines | grep -vw -- servant-conduit | grep -vw -- servant-pipes | grep -vw -- cookbook-basic-auth | grep -vw -- cookbook-curl-mock | grep -vw -- cookbook-basic-streaming | grep -vw -- cookbook-db-postgres-pool | grep -vw -- cookbook-db-sqlite-simple | grep -vw -- cookbook-file-upload | grep -vw -- cookbook-generic | grep -vw -- cookbook-https | grep -vw -- cookbook-sentry | grep -vw -- cookbook-testing | grep -vw -- cookbook-structuring-apis | grep -vw -- cookbook-using-custom-monad | grep -vw -- cookbook-using-free-client | sed 's/^/constraints: /' | sed 's/-[^-]*$/ installed/' >> cabal.project.local; done; fi"
- cat cabal.project || true
- cat cabal.project.local || true
- if [ -f "servant/configure.ac" ]; then
@ -97,12 +97,24 @@ install:
- if [ -f "doc/tutorial/configure.ac" ]; then
(cd "doc/tutorial" && autoreconf -i);
fi
- if [ -f "servant-machines/configure.ac" ]; then
(cd "servant-machines" && autoreconf -i);
fi
- if [ -f "servant-conduit/configure.ac" ]; then
(cd "servant-conduit" && autoreconf -i);
fi
- if [ -f "servant-pipes/configure.ac" ]; then
(cd "servant-pipes" && autoreconf -i);
fi
- if [ -f "doc/cookbook/basic-auth/configure.ac" ]; then
(cd "doc/cookbook/basic-auth" && autoreconf -i);
fi
- if [ -f "doc/cookbook/curl-mock/configure.ac" ]; then
(cd "doc/cookbook/curl-mock" && autoreconf -i);
fi
- if [ -f "doc/cookbook/basic-streaming/configure.ac" ]; then
(cd "doc/cookbook/basic-streaming" && autoreconf -i);
fi
- if [ -f "doc/cookbook/db-postgres-pool/configure.ac" ]; then
(cd "doc/cookbook/db-postgres-pool" && autoreconf -i);
fi
@ -134,7 +146,7 @@ install:
(cd "doc/cookbook/using-free-client" && autoreconf -i);
fi
- rm -f cabal.project.freeze
- rm -rf .ghc.environment.* "servant"/dist "servant-client"/dist "servant-client-core"/dist "servant-docs"/dist "servant-foreign"/dist "servant-server"/dist "doc/tutorial"/dist "doc/cookbook/basic-auth"/dist "doc/cookbook/curl-mock"/dist "doc/cookbook/db-postgres-pool"/dist "doc/cookbook/db-sqlite-simple"/dist "doc/cookbook/file-upload"/dist "doc/cookbook/generic"/dist "doc/cookbook/https"/dist "doc/cookbook/sentry"/dist "doc/cookbook/testing"/dist "doc/cookbook/structuring-apis"/dist "doc/cookbook/using-custom-monad"/dist "doc/cookbook/using-free-client"/dist
- rm -rf .ghc.environment.* "servant"/dist "servant-client"/dist "servant-client-core"/dist "servant-docs"/dist "servant-foreign"/dist "servant-server"/dist "doc/tutorial"/dist "servant-machines"/dist "servant-conduit"/dist "servant-pipes"/dist "doc/cookbook/basic-auth"/dist "doc/cookbook/curl-mock"/dist "doc/cookbook/basic-streaming"/dist "doc/cookbook/db-postgres-pool"/dist "doc/cookbook/db-sqlite-simple"/dist "doc/cookbook/file-upload"/dist "doc/cookbook/generic"/dist "doc/cookbook/https"/dist "doc/cookbook/sentry"/dist "doc/cookbook/testing"/dist "doc/cookbook/structuring-apis"/dist "doc/cookbook/using-custom-monad"/dist "doc/cookbook/using-free-client"/dist
- DISTDIR=$(mktemp -d /tmp/dist-test.XXXX)
# Here starts the actual work to be performed for the package under test;
@ -148,11 +160,11 @@ script:
- mv dist-newstyle/sdist/*.tar.gz ${DISTDIR}/
- cd ${DISTDIR} || false
- find . -maxdepth 1 -name '*.tar.gz' -exec tar -xvf '{}' \;
- "printf 'packages: servant-*/*.cabal servant-client-*/*.cabal servant-client-core-*/*.cabal servant-docs-*/*.cabal servant-foreign-*/*.cabal servant-server-*/*.cabal tutorial-*/*.cabal cookbook-basic-auth-*/*.cabal cookbook-curl-mock-*/*.cabal cookbook-db-postgres-pool-*/*.cabal cookbook-db-sqlite-simple-*/*.cabal cookbook-file-upload-*/*.cabal cookbook-generic-*/*.cabal cookbook-https-*/*.cabal cookbook-sentry-*/*.cabal cookbook-testing-*/*.cabal cookbook-structuring-apis-*/*.cabal cookbook-using-custom-monad-*/*.cabal cookbook-using-free-client-*/*.cabal\\n' > cabal.project"
- "printf 'packages: servant-*/*.cabal servant-client-*/*.cabal servant-client-core-*/*.cabal servant-docs-*/*.cabal servant-foreign-*/*.cabal servant-server-*/*.cabal tutorial-*/*.cabal servant-machines-*/*.cabal servant-conduit-*/*.cabal servant-pipes-*/*.cabal cookbook-basic-auth-*/*.cabal cookbook-curl-mock-*/*.cabal cookbook-basic-streaming-*/*.cabal cookbook-db-postgres-pool-*/*.cabal cookbook-db-sqlite-simple-*/*.cabal cookbook-file-upload-*/*.cabal cookbook-generic-*/*.cabal cookbook-https-*/*.cabal cookbook-sentry-*/*.cabal cookbook-testing-*/*.cabal cookbook-structuring-apis-*/*.cabal cookbook-using-custom-monad-*/*.cabal cookbook-using-free-client-*/*.cabal\\n' > cabal.project"
- "echo 'constraints: foundation >=0.0.14,memory <0.14.12 || >0.14.12' >> cabal.project"
- "echo 'allow-newer: servant-auth-server:http-types,servant-auth-server:servant-server,servant-js:base, servant-pagination:servant,servant-pagination:servant-server' >> cabal.project"
- touch cabal.project.local
- "if ! $NOINSTALLEDCONSTRAINTS; then for pkg in $($HCPKG list --simple-output); do echo $pkg | grep -vw -- servant | grep -vw -- servant-client | grep -vw -- servant-client-core | grep -vw -- servant-docs | grep -vw -- servant-foreign | grep -vw -- servant-server | grep -vw -- tutorial | grep -vw -- cookbook-basic-auth | grep -vw -- cookbook-curl-mock | grep -vw -- cookbook-db-postgres-pool | grep -vw -- cookbook-db-sqlite-simple | grep -vw -- cookbook-file-upload | grep -vw -- cookbook-generic | grep -vw -- cookbook-https | grep -vw -- cookbook-sentry | grep -vw -- cookbook-testing | grep -vw -- cookbook-structuring-apis | grep -vw -- cookbook-using-custom-monad | grep -vw -- cookbook-using-free-client | sed 's/^/constraints: /' | sed 's/-[^-]*$/ installed/' >> cabal.project.local; done; fi"
- "if ! $NOINSTALLEDCONSTRAINTS; then for pkg in $($HCPKG list --simple-output); do echo $pkg | grep -vw -- servant | grep -vw -- servant-client | grep -vw -- servant-client-core | grep -vw -- servant-docs | grep -vw -- servant-foreign | grep -vw -- servant-server | grep -vw -- tutorial | grep -vw -- servant-machines | grep -vw -- servant-conduit | grep -vw -- servant-pipes | grep -vw -- cookbook-basic-auth | grep -vw -- cookbook-curl-mock | grep -vw -- cookbook-basic-streaming | grep -vw -- cookbook-db-postgres-pool | grep -vw -- cookbook-db-sqlite-simple | grep -vw -- cookbook-file-upload | grep -vw -- cookbook-generic | grep -vw -- cookbook-https | grep -vw -- cookbook-sentry | grep -vw -- cookbook-testing | grep -vw -- cookbook-structuring-apis | grep -vw -- cookbook-using-custom-monad | grep -vw -- cookbook-using-free-client | sed 's/^/constraints: /' | sed 's/-[^-]*$/ installed/' >> cabal.project.local; done; fi"
- cat cabal.project || true
- cat cabal.project.local || true
- echo -en 'travis_fold:end:unpack\\r'

5
cabal.project
View file

@ -6,10 +6,15 @@ packages: servant/
servant-server/
doc/tutorial/
servant-machines/
servant-conduit/
servant-pipes/
-- doc/cookbook/*/*.cabal
doc/cookbook/basic-auth
doc/cookbook/curl-mock
doc/cookbook/basic-streaming
doc/cookbook/db-postgres-pool
doc/cookbook/db-sqlite-simple
doc/cookbook/file-upload

126
doc/cookbook/basic-streaming/Streaming.lhs Normal file
View file

@ -0,0 +1,126 @@
# Streaming out-of-the-box
In other words, without streaming libraries.
## Introduction
- Servant supports streaming
- Some basic usage doesn't require usage of streaming libraries,
like `conduit`, `pipes`, `machines` or `streaming`.
We have bindings for them though.
- This is similar example file, which is bundled with each of the packages (TODO: links)
- `SourceT` doesn't have *Prelude* with handy combinators, so we have to write
things ourselves. (Note to self: `mapM` and `foldM` would be handy to have).
## Code
```haskell
{-# LANGUAGE BangPatterns #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TypeOperators #-}
module Main (main) where
import Control.Concurrent
(threadDelay)
import Control.Monad.IO.Class
(MonadIO (..))
import qualified Data.ByteString as BS
import Data.Maybe
(fromMaybe)
import Network.HTTP.Client
(defaultManagerSettings, newManager)
import Network.Wai
(Application)
import System.Environment
(getArgs, lookupEnv)
import Text.Read
(readMaybe)
import Servant
import Servant.Client
import qualified Servant.Types.SourceT as S
import qualified Network.Wai.Handler.Warp as Warp
type FastAPI = "get" :> Capture "num" Int :> StreamGet NewlineFraming JSON (SourceIO Int)
type API = FastAPI
:<|> "slow" :> Capture "num" Int :> StreamGet NewlineFraming JSON (SourceIO Int)
-- monad can be ResourceT IO too.
:<|> "readme" :> StreamGet NoFraming OctetStream (SourceIO BS.ByteString)
-- we can have streaming request body
:<|> "proxy"
:> StreamBody NoFraming OctetStream (SourceIO BS.ByteString)
:> StreamPost NoFraming OctetStream (SourceIO BS.ByteString)
api :: Proxy API
api = Proxy
server :: Server API
server = fast :<|> slow :<|> readme :<|> proxy where
fast n = liftIO $ do
putStrLn $ "/get/" ++ show n
return $ fastSource n
slow n = liftIO $ do
putStrLn $ "/slow/" ++ show n
return $ slowSource n
readme = liftIO $ do
putStrLn "/proxy"
return (S.readFile "README.md")
proxy c = liftIO $ do
putStrLn "/proxy"
return c
-- for some reason unfold leaks?
fastSource = S.fromStepT . mk where
mk m
| m < 0 = S.Stop
| otherwise = S.Yield m (mk (m - 1))
slowSource m = S.mapStepT delay (fastSource m) where
delay S.Stop = S.Stop
delay (S.Error err) = S.Error err
delay (S.Skip s) = S.Skip (delay s)
delay (S.Effect ms) = S.Effect (fmap delay ms)
delay (S.Yield x s) = S.Effect $
S.Yield x (delay s) <$ threadDelay 1000000
app :: Application
app = serve api server
cli :: Client ClientM FastAPI
cli :<|> _ :<|> _ :<|> _ = client api
main :: IO ()
main = do
args <- getArgs
case args of
("server":_) -> do
putStrLn "Starting cookbook-basic-streaming at http://localhost:8000"
port <- fromMaybe 8000 . (>>= readMaybe) <$> lookupEnv "PORT"
Warp.run port app
("client":ns:_) -> do
n <- maybe (fail $ "not a number: " ++ ns) pure $ readMaybe ns
mgr <- newManager defaultManagerSettings
burl <- parseBaseUrl "http://localhost:8000/"
withClientM (cli n) (mkClientEnv mgr burl) $ \me -> case me of
Left err -> print err
Right src -> do
x <- S.unSourceT src (go (0 :: Int))
print x
where
go !acc S.Stop = return acc
go !acc (S.Error err) = print err >> return acc
go !acc (S.Skip s) = go acc s
go !acc (S.Effect ms) = ms >>= go acc
go !acc (S.Yield _ s) = go (acc + 1) s
_ -> do
putStrLn "Try:"
putStrLn "cabal new-run cookbook-basic-streaming server"
putStrLn "cabal new-run cookbook-basic-streaming client 10"
putStrLn "time curl -H 'Accept: application/json' localhost:8000/slow/5"
```

28
doc/cookbook/basic-streaming/basic-streaming.cabal Normal file
View file

@ -0,0 +1,28 @@
name: cookbook-basic-streaming
version: 2.1
synopsis: Streaming in servant without streaming libs
homepage: http://haskell-servant.readthedocs.org/
license: BSD3
license-file: ../../../servant/LICENSE
author: Servant Contributors
maintainer: haskell-servant-maintainers@googlegroups.com
build-type: Simple
cabal-version: >=1.10
tested-with: GHC==8.0.2, GHC==8.2.2, GHC==8.4.4, GHC==8.6.1
executable cookbook-basic-streaming
main-is: Streaming.lhs
build-tool-depends: markdown-unlit:markdown-unlit
default-language: Haskell2010
ghc-options: -Wall -pgmL markdown-unlit -threaded -rtsopts
hs-source-dirs: .
build-depends: base >= 4.8 && <4.13
, aeson
, bytestring
, servant
, servant-server
, servant-client
, http-client
, wai
, warp

17
doc/tutorial/ApiType.lhs
View file

@ -129,16 +129,23 @@ type UserAPI4 = "users" :> Get '[JSON] [User]
### `StreamGet` and `StreamPost`
*Note*: Streaming has changed considerably in `servant-0.15`.
The `StreamGet` and `StreamPost` combinators are defined in terms of the more general `Stream`
``` haskell ignore
data Stream (method :: k1) (framing :: *) (contentType :: *) a
type StreamGet = Stream 'GET
type StreamPost = Stream 'POST
data Stream (method :: k1) (status :: Nat) (framing :: *) (contentType :: *) (a :: *)
type StreamGet = Stream 'GET 200
type StreamPost = Stream 'POST 200
```
These describe endpoints that return a stream of values rather than just a single value. They not only take a single content type as a parameter, but also a framing strategy -- this specifies how the individual results are delineated from one another in the stream. The three standard strategies given with Servant are `NewlineFraming`, `NetstringFraming` and `NoFraming`, but others can be written to match other protocols.
These describe endpoints that return a stream of values rather than just a
single value. They not only take a single content type as a parameter, but also
a framing strategy -- this specifies how the individual results are delineated
from one another in the stream. The three standard strategies given with
Servant are `NewlineFraming`, `NetstringFraming` and `NoFraming`, but others
can be written to match other protocols.
### `Capture`

42
doc/tutorial/Client.lhs
View file

@ -20,6 +20,8 @@ import GHC.Generics
import Network.HTTP.Client (newManager, defaultManagerSettings)
import Servant.API
import Servant.Client
import Servant.Types.SourceT (foreach)
import Control.Monad.Codensity (Codensity)
```
Also, we need examples for some domain specific data types:
@ -217,41 +219,39 @@ getClients clientEnv
Consider the following streaming API type:
``` haskell
type StreamAPI = "positionStream" :> StreamGet NewlineFraming JSON (ResultStream Position)
type StreamAPI = "positionStream" :> StreamGet NewlineFraming JSON (SourceIO Position)
```
Note that when we declared an API to serve, we specified a `StreamGenerator` as a producer of streams. Now we specify our result type as a `ResultStream`. With types that can be used both ways, if appropriate adaptors are written (in the form of `ToStreamGenerator` and `BuildFromStream` instances), then this asymmetry isn't necessary. Otherwise, if you want to share the same API across clients and servers, you can parameterize it like so:
``` haskell ignore
type StreamAPI f = "positionStream" :> StreamGet NewlineFraming JSON (f Position)
type ServerStreamAPI = StreamAPI StreamGenerator
type ClientStreamAPI = StreamAPI ResultStream
```
Note that we use the same `SourceIO` type as on the server-side
(this is different from `servant-0.14`).
In any case, here's how we write a function to query our API:
``` haskell
```haskell
streamAPI :: Proxy StreamAPI
streamAPI = Proxy
posStream :: ClientM (ResultStream Position)
posStream :: ClientM (Codensity IO (SourceIO Position))
posStream = client streamAPI
```
And here's how to just print out all elements from a `ResultStream`, to give some idea of how to work with them.
We'll get back a `Codensity IO (SourceIO Position)`. The wrapping in
`Codensity` is generally necessary, as `Codensity` lets us `bracket` things
properly. This is best explained by an example. To consume `ClientM (Codentity
IO ...)` we can use `withClientM` helper: the underlying HTTP connection is
open only until the inner functions exits. Inside the block we can e.g just
print out all elements from a `SourceIO`, to give some idea of how to work with
them.
``` haskell
printResultStream :: Show a => ResultStream a -> IO ()
printResultStream (ResultStream k) = k $ \getResult ->
let loop = do
r <- getResult
case r of
Nothing -> return ()
Just x -> print x >> loop
in loop
printSourceIO :: Show a => ClientEnv -> ClientM (Codensity IO (SourceIO a)) -> IO ()
printSourceIO env c = withClientM c env $ \e -> case e of
Left err -> putStrLn $ "Error: " ++ show err
Right rs -> foreach fail print rs
```
The stream is parsed and provided incrementally. So the above loop prints out each result as soon as it is received on the stream, rather than waiting until they are all available to print them at once.
The stream is parsed and provided incrementally. So the above loop prints out
each result as soon as it is received on the stream, rather than waiting until
they are all available to print them at once.
You now know how to use **servant-client**!

30
doc/tutorial/Server.lhs
View file

@ -46,6 +46,7 @@ import Servant
import System.Directory
import Text.Blaze
import Text.Blaze.Html.Renderer.Utf8
import Servant.Types.SourceT (source)
import qualified Data.Aeson.Parser
import qualified Text.Blaze.Html
```
@ -1160,24 +1161,37 @@ app5 = serve readerAPI (hoistServer readerAPI funToHandler funServerT)
## Streaming endpoints
We can create endpoints that don't just give back a single result, but give back a *stream* of results, served one at a time. Stream endpoints only provide a single content type, and also specify what framing strategy is used to delineate the results. To serve these results, we need to give back a stream producer. Adapters can be written to `Pipes`, `Conduit` and the like, or written directly as `StreamGenerator`s. StreamGenerators are IO-based continuations that are handed two functions -- the first to write the first result back, and the second to write all subsequent results back. (This is to allow handling of situations where the entire stream is prefixed by a header, or where a boundary is written between elements, but not prior to the first element). The API of a streaming endpoint needs to explicitly specify which sort of generator it produces. Note that the generator itself is returned by a `Handler` action, so that additional IO may be done in the creation of one.
We can create endpoints that don't just give back a single result, but give
back a *stream* of results, served one at a time. Stream endpoints only provide
a single content type, and also specify what framing strategy is used to
delineate the results. To serve these results, we need to give back a stream
producer. Adapters can be written to *Pipes*, *Conduit* and the like, or
written directly as `SourceIO`s. SourceIO builts upon servant's own `SourceT`
stream type (it's simpler than *Pipes* or *Conduit*).
The API of a streaming endpoint needs to explicitly specify which sort of
generator it produces. Note that the generator itself is returned by a
`Handler` action, so that additional IO may be done in the creation of one.
``` haskell
type StreamAPI = "userStream" :> StreamGet NewlineFraming JSON (StreamGenerator User)
type StreamAPI = "userStream" :> StreamGet NewlineFraming JSON (SourceIO User)
streamAPI :: Proxy StreamAPI
streamAPI = Proxy
streamUsers :: StreamGenerator User
streamUsers = StreamGenerator $ \sendFirst sendRest -> do
sendFirst isaac
sendRest albert
sendRest albert
streamUsers :: SourceIO User
streamUsers = source [isaac, albert, albert]
app6 :: Application
app6 = serve streamAPI (return streamUsers)
```
This simple application returns a stream of `User` values encoded in JSON format, with each value separated by a newline. In this case, the stream will consist of the value of `isaac`, followed by the value of `albert`, then the value of `albert` a third time. Importantly, the stream is written back as results are produced, rather than all at once. This means first that results are delivered when they are available, and second, that if an exception interrupts production of the full stream, nonetheless partial results have already been written back.
This simple application returns a stream of `User` values encoded in JSON
format, with each value separated by a newline. In this case, the stream will
consist of the value of `isaac`, followed by the value of `albert`, then the
value of `albert` a second time. Importantly, the stream is written back as
results are produced, rather than all at once. This means first that results
are delivered when they are available, and second, that if an exception
interrupts production of the full stream, nonetheless partial results have
already been written back.
## Conclusion

1
doc/tutorial/tutorial.cabal
View file

@ -46,6 +46,7 @@ library
, http-client
, http-media
, http-types
, kan-extensions
, mtl
, string-conversions
, text

1
servant-client-core/servant-client-core.cabal
View file

@ -70,6 +70,7 @@ library
, http-api-data >= 0.3.8.1 && < 0.4
, http-media >= 0.7.1.2 && < 0.8
, http-types >= 0.12.1 && < 0.13
, kan-extensions >= 5.2 && < 5.3
, network-uri >= 2.6.1.0 && < 2.7
, safe >= 0.3.17 && < 0.4

2
servant-client-core/src/Servant/Client/Core.hs
View file

@ -44,7 +44,7 @@ module Servant.Client.Core
, GenResponse (..)
, RunClient(..)
, module Servant.Client.Core.Internal.BaseUrl
, StreamingResponse(..)
, StreamingResponse
-- * Writing HasClient instances
-- | These functions need not be re-exported by backend libraries.

93
servant-client-core/src/Servant/Client/Core/Internal/HasClient.hs
View file

@ -16,10 +16,7 @@ module Servant.Client.Core.Internal.HasClient where
import Prelude ()
import Prelude.Compat
import Control.Concurrent
(modifyMVar, newMVar)
import Control.Monad.IO.Class
(MonadIO (..))
import qualified Data.ByteString as BS
import qualified Data.ByteString.Lazy as BL
import Data.Foldable
(toList)
@ -27,8 +24,6 @@ import Data.List
(foldl')
import Data.Proxy
(Proxy (Proxy))
import Data.Semigroup
((<>))
import Data.Sequence
(fromList)
import Data.String
@ -40,19 +35,22 @@ import GHC.TypeLits
import qualified Network.HTTP.Types as H
import Servant.API
((:<|>) ((:<|>)), (:>), AuthProtect, BasicAuth, BasicAuthData,
BuildHeadersTo (..), ByteStringParser (..), Capture',
CaptureAll, Description, EmptyAPI, FramingUnrender (..),
FromResultStream (..), Header', Headers (..), HttpVersion,
IsSecure, MimeRender (mimeRender),
MimeUnrender (mimeUnrender), NoContent (NoContent), QueryFlag,
QueryParam', QueryParams, Raw, ReflectMethod (..), RemoteHost,
ReqBody', ResultStream (..), SBoolI, Stream, Summary,
ToHttpApiData, Vault, Verb, WithNamedContext, contentType,
getHeadersHList, getResponse, toQueryParam, toUrlPiece)
BuildHeadersTo (..), Capture', CaptureAll, Description,
EmptyAPI, FramingUnrender (..), FromSourceIO (..),
Header', Headers (..), HttpVersion, IsSecure,
MimeRender (mimeRender), MimeUnrender (mimeUnrender),
NoContent (NoContent), QueryFlag, QueryParam', QueryParams,
Raw, ReflectMethod (..), RemoteHost, ReqBody', SBoolI, Stream,
StreamBody, Summary, ToHttpApiData, Vault, Verb,
WithNamedContext, contentType, getHeadersHList, getResponse,
toQueryParam, toUrlPiece)
import Servant.API.ContentTypes
(contentTypes)
import Servant.API.Modifiers
(FoldRequired, RequiredArgument, foldRequiredArgument)
import Control.Monad.Codensity
(Codensity (..))
import qualified Servant.Types.SourceT as S
import Servant.Client.Core.Internal.Auth
import Servant.Client.Core.Internal.BasicAuth
@ -274,54 +272,25 @@ instance {-# OVERLAPPING #-}
hoistClientMonad _ _ f ma = f ma
instance {-# OVERLAPPABLE #-}
( RunClient m, MonadIO m, MimeUnrender ct a, ReflectMethod method,
FramingUnrender framing a, FromResultStream a b
) => HasClient m (Stream method status framing ct b) where
( RunClient m, MimeUnrender ct chunk, ReflectMethod method,
FramingUnrender framing, FromSourceIO chunk a
) => HasClient m (Stream method status framing ct a) where
type Client m (Stream method status framing ct b) = m b
type Client m (Stream method status framing ct a) = m (Codensity IO a)
hoistClientMonad _ _ f ma = f ma
clientWithRoute _pm Proxy req = do
sresp <- streamingRequest req
sresp <- streamingRequest req
{ requestAccept = fromList [contentType (Proxy :: Proxy ct)]
, requestMethod = reflectMethod (Proxy :: Proxy method)
}
liftIO $ fromResultStream $ ResultStream $ \k ->
runStreamingResponse sresp $ \gres -> do
let reader = responseBody gres
let unrender = unrenderFrames (Proxy :: Proxy framing) (Proxy :: Proxy a)
loop bs = do
res <- BL.fromStrict <$> reader
if BL.null res
then return $ parseEOF unrender res
else let sofar = (bs <> res)
in case parseIncremental unrender sofar of
Just x -> return x
Nothing -> loop sofar
(frameParser, remainder) <- loop BL.empty
state <- newMVar remainder
let frameLoop bs = do
res <- BL.fromStrict <$> reader
let addIsEmptyInfo (a, r) = (r, (a, BL.null r && BL.null res))
if BL.null res
then if BL.null bs
then return ("", (Right "", True))
else return . addIsEmptyInfo $ parseEOF frameParser bs
else let sofar = (bs <> res)
in case parseIncremental frameParser sofar of
Just x -> return $ addIsEmptyInfo x
Nothing -> frameLoop sofar
return $ do
let mimeUnrender' = mimeUnrender (Proxy :: Proxy ct) :: BL.ByteString -> Either String chunk
framingUnrender' = framingUnrender (Proxy :: Proxy framing) mimeUnrender'
gres <- sresp
return $ fromSourceIO $ framingUnrender' $ S.fromAction BS.null (responseBody gres)
go = processResult <$> modifyMVar state frameLoop
processResult (Right bs,isDone) =
if BL.null bs && isDone
then Nothing
else Just $ case mimeUnrender (Proxy :: Proxy ct) bs :: Either String a of
Left err -> Left err
Right x -> Right x
processResult (Left err, _) = Just (Left err)
k go
hoistClientMonad _ _ f ma = f ma
-- | If you use a 'Header' in one of your endpoints in your API,
-- the corresponding querying function will automatically take
@ -573,6 +542,18 @@ instance (MimeRender ct a, HasClient m api)
hoistClientMonad pm _ f cl = \a ->
hoistClientMonad pm (Proxy :: Proxy api) f (cl a)
instance
( HasClient m api
) => HasClient m (StreamBody framing ctype a :> api)
where
type Client m (StreamBody framing ctype a :> api) = a -> Client m api
hoistClientMonad pm _ f cl = \a ->
hoistClientMonad pm (Proxy :: Proxy api) f (cl a)
clientWithRoute _pm Proxy _req _body = error "HasClient @StreamBody"
-- | Make the querying function append @path@ to the request path.
instance (KnownSymbol path, HasClient m api) => HasClient m (path :> api) where
type Client m (path :> api) = Client m api

4
servant-client-core/src/Servant/Client/Core/Internal/Request.hs
View file

@ -38,6 +38,8 @@ import Network.HTTP.Media
import Network.HTTP.Types
(Header, HeaderName, HttpVersion, Method, QueryItem, Status,
http11, methodGet)
import Control.Monad.Codensity
(Codensity (..))
import Web.HttpApiData
(ToHttpApiData, toEncodedUrlPiece, toHeader)
@ -89,7 +91,7 @@ data GenResponse a = Response
} deriving (Eq, Show, Generic, Typeable, Functor, Foldable, Traversable)
type Response = GenResponse LBS.ByteString
newtype StreamingResponse = StreamingResponse { runStreamingResponse :: forall a. (GenResponse (IO BS.ByteString) -> IO a) -> IO a }
type StreamingResponse = Codensity IO (GenResponse (IO BS.ByteString))
-- A GET request to the top-level path
defaultRequest :: Request

2
servant-client-core/src/Servant/Client/Core/Internal/RunClient.hs
View file

@ -26,7 +26,7 @@ import Servant.API
import Servant.Client.Core.Internal.ClientF
import Servant.Client.Core.Internal.Request
(GenResponse (..), Request, Response, ServantError (..),
StreamingResponse (..))
StreamingResponse)
class Monad m => RunClient m where
-- | How to make a request.

15
servant-client/servant-client.cabal
View file

@ -50,7 +50,8 @@ library
-- Servant dependencies
build-depends:
servant-client-core == 0.14.*
servant == 0.14.*
, servant-client-core == 0.14.*
-- Other dependencies: Lower bound around what is in the latest Stackage LTS.
-- Here can be exceptions if we really need features from the newer versions.
@ -60,6 +61,7 @@ library
, http-media >= 0.7.1.2 && < 0.8
, http-types >= 0.12.1 && < 0.13
, exceptions >= 0.10.0 && < 0.11
, kan-extensions >= 5.2 && < 5.3
, monad-control >= 1.0.2.3 && < 1.1
, semigroupoids >= 5.2.2 && < 5.4
, stm >= 2.4.5.0 && < 2.6
@ -72,7 +74,7 @@ library
test-suite spec
type: exitcode-stdio-1.0
ghc-options: -Wall -rtsopts -with-rtsopts=-T
ghc-options: -Wall -rtsopts -threaded "-with-rtsopts=-T -N2"
default-language: Haskell2010
hs-source-dirs: test
main-is: Spec.hs
@ -90,6 +92,7 @@ test-suite spec
, http-client
, http-types
, mtl
, kan-extensions
, servant-client
, servant-client-core
, text
@ -100,16 +103,18 @@ test-suite spec
-- Additonal dependencies
build-depends:
generics-sop >= 0.4.0.1 && < 0.5
, hspec >= 2.5.1 && < 2.6
entropy >= 0.4.1.3 && < 0.5
, generics-sop >= 0.4.0.1 && < 0.5
, hspec >= 2.5.8 && < 2.6
, HUnit >= 1.6 && < 1.7
, network >= 2.8.0.0 && < 2.9
, QuickCheck >= 2.12.6.1 && < 2.13
, servant == 0.14.*
, servant-server == 0.14.*
, tdigest >= 0.2 && < 0.3
build-tool-depends:
hspec-discover:hspec-discover >= 2.5.1 && < 2.6
hspec-discover:hspec-discover >= 2.5.8 && < 2.6
test-suite readme
type: exitcode-stdio-1.0

1
servant-client/src/Servant/Client.hs
View file

@ -7,6 +7,7 @@ module Servant.Client
( client
, ClientM
, runClientM
, withClientM
, ClientEnv(..)
, mkClientEnv
, hoistClient

15
servant-client/src/Servant/Client/Internal/HttpClient.hs
View file

@ -21,6 +21,8 @@ import Control.Monad.Base
(MonadBase (..))
import Control.Monad.Catch
(MonadCatch, MonadThrow)
import Control.Monad.Codensity
(Codensity (..))
import Control.Monad.Error.Class
(MonadError (..))
import Control.Monad.Reader
@ -141,6 +143,17 @@ instance ClientLike (ClientM a) (ClientM a) where
runClientM :: ClientM a -> ClientEnv -> IO (Either ServantError a)
runClientM cm env = runExceptT $ flip runReaderT env $ unClientM cm
withClientM
:: ClientM (Codensity IO a) -- ^ client with codensity result
-> ClientEnv -- ^ environment
-> (Either ServantError a -> IO b) -- ^ continuation
-> IO b
withClientM cm env k = do
e <- runExceptT (runReaderT (unClientM cm) env)
case e of
Left err -> k (Left err)
Right cod -> runCodensity cod (k . Right)
performRequest :: Request -> ClientM Response
performRequest req = do
ClientEnv m burl cookieJar' <- ask
@ -178,7 +191,7 @@ performStreamingRequest req = do
m <- asks manager
burl <- asks baseUrl
let request = requestToClientRequest burl req
return $ StreamingResponse $
return $ Codensity $
\k -> Client.withResponse request m $
\r -> do
let status = Client.responseStatus r

160
servant-client/test/Servant/StreamSpec.hs
View file

@ -22,32 +22,38 @@
module Servant.StreamSpec (spec) where
import Control.Monad
(replicateM_, void)
import qualified Data.ByteString as BS
(when)
import Control.Monad.Codensity
(Codensity (..))
import Control.Monad.IO.Class
(MonadIO (..))
import Control.Monad.Trans.Except
import qualified Data.ByteString as BS
import Data.Proxy
import qualified Network.HTTP.Client as C
import qualified Data.TDigest as TD
import qualified Network.HTTP.Client as C
import Prelude ()
import Prelude.Compat
import System.IO
(IOMode (ReadMode), withFile)
import System.IO.Unsafe
(unsafePerformIO)
import Test.Hspec
import Test.QuickCheck
import Servant.API
((:<|>) ((:<|>)), (:>), JSON, NetstringFraming,
NewlineFraming, NoFraming, OctetStream, ResultStream (..),
StreamGenerator (..), StreamGet)
NewlineFraming, NoFraming, OctetStream, SourceIO, StreamGet)
import Servant.Client
import Servant.ClientSpec
(Person (..))
import qualified Servant.ClientSpec as CS
import qualified Servant.ClientSpec as CS
import Servant.Server
import Servant.Types.SourceT
import System.Entropy
(getEntropy, getHardwareEntropy)
import System.IO.Unsafe
(unsafePerformIO)
import System.Mem
(performGC)
import Test.Hspec
#if MIN_VERSION_base(4,10,0)
import GHC.Stats
(gc, gcdetails_mem_in_use_bytes, getRTSStats)
(gc, gcdetails_live_bytes, getRTSStats)
#else
import GHC.Stats
(currentBytesUsed, getGCStats)
@ -57,21 +63,17 @@ spec :: Spec
spec = describe "Servant.Stream" $ do
streamSpec
type StreamApi f =
"streamGetNewline" :> StreamGet NewlineFraming JSON (f Person)
:<|> "streamGetNetstring" :> StreamGet NetstringFraming JSON (f Person)
:<|> "streamALot" :> StreamGet NoFraming OctetStream (f BS.ByteString)
type StreamApi =
"streamGetNewline" :> StreamGet NewlineFraming JSON (SourceIO Person)
:<|> "streamGetNetstring" :> StreamGet NetstringFraming JSON (SourceIO Person)
:<|> "streamALot" :> StreamGet NoFraming OctetStream (SourceIO BS.ByteString)
api :: Proxy StreamApi
api = Proxy
capi :: Proxy (StreamApi ResultStream)
capi = Proxy
sapi :: Proxy (StreamApi StreamGenerator)
sapi = Proxy
getGetNL, getGetNS :: ClientM (ResultStream Person)
getGetALot :: ClientM (ResultStream BS.ByteString)
getGetNL :<|> getGetNS :<|> getGetALot = client capi
getGetNL, getGetNS :: ClientM (Codensity IO (SourceIO Person))
getGetALot :: ClientM (Codensity IO (SourceIO BS.ByteString))
getGetNL :<|> getGetNS :<|> getGetALot = client api
alice :: Person
alice = Person "Alice" 42
@ -80,25 +82,23 @@ bob :: Person
bob = Person "Bob" 25
server :: Application
server = serve sapi
$ return (StreamGenerator (\f r -> f alice >> r bob >> r alice))
:<|> return (StreamGenerator (\f r -> f alice >> r bob >> r alice))
:<|> return (StreamGenerator lotsGenerator)
server = serve api
$ return (source [alice, bob, alice])
:<|> return (source [alice, bob, alice])
-- 2 ^ (18 + 10) = 256M
:<|> return (SourceT ($ lots (powerOfTwo 18)))
where
lotsGenerator f r = do
void $ f ""
void $ withFile "/dev/urandom" ReadMode $
\handle -> streamFiveMBNTimes handle 1000 r
return ()
streamFiveMBNTimes handle left sink
| left <= (0 :: Int) = return ()
| otherwise = do
msg <- BS.hGet handle (megabytes 5)
_ <- sink msg
streamFiveMBNTimes handle (left - 1) sink
lots n
| n < 0 = Stop
| otherwise = Effect $ do
let size = powerOfTwo 10
mbs <- getHardwareEntropy size
bs <- maybe (getEntropy size) pure mbs
return (Yield bs (lots (n - 1)))
powerOfTwo :: Int -> Int
powerOfTwo = (2 ^)
{-# NOINLINE manager' #-}
manager' :: C.Manager
@ -107,41 +107,69 @@ manager' = unsafePerformIO $ C.newManager C.defaultManagerSettings
runClient :: ClientM a -> BaseUrl -> IO (Either ServantError a)
runClient x baseUrl' = runClientM x (mkClientEnv manager' baseUrl')
testRunResultStream :: ResultStream a
-> IO ( Maybe (Either String a)
, Maybe (Either String a)
, Maybe (Either String a)
, Maybe (Either String a))
testRunResultStream (ResultStream k)
= k $ \act -> (,,,) <$> act <*> act <*> act <*> act
testRunSourceIO :: Codensity IO (SourceIO a)
-> IO (Either String [a])
testRunSourceIO = runExceptT . runSourceT . joinCodensitySourceT
joinCodensitySourceT :: Codensity m (SourceT m a) -> SourceT m a
joinCodensitySourceT cod =
SourceT $ \r ->
runCodensity cod $ \src ->
unSourceT src r
streamSpec :: Spec
streamSpec = beforeAll (CS.startWaiApp server) $ afterAll CS.endWaiApp $ do
it "works with Servant.API.StreamGet.Newline" $ \(_, baseUrl) -> do
Right res <- runClient getGetNL baseUrl
let jra = Just (Right alice)
jrb = Just (Right bob)
testRunResultStream res `shouldReturn` (jra, jrb, jra, Nothing)
Right res <- runClient getGetNL baseUrl
testRunSourceIO res `shouldReturn` Right [alice, bob, alice]
it "works with Servant.API.StreamGet.Netstring" $ \(_, baseUrl) -> do
Right res <- runClient getGetNS baseUrl
let jra = Just (Right alice)
jrb = Just (Right bob)
testRunResultStream res `shouldReturn` (jra, jrb, jra, Nothing)
Right res <- runClient getGetNS baseUrl
testRunSourceIO res `shouldReturn` Right [alice, bob, alice]
{-
it "streams in constant memory" $ \(_, baseUrl) -> do
Right (ResultStream res) <- runClient getGetALot baseUrl
let consumeNChunks n = replicateM_ n (res void)
consumeNChunks 900
Right rs <- runClient getGetALot baseUrl
performGC
-- usage0 <- getUsage
-- putStrLn $ "Start: " ++ show usage0
tdigest <- memoryUsage $ joinCodensitySourceT rs
-- putStrLn $ "Median: " ++ show (TD.median tdigest)
-- putStrLn $ "Mean: " ++ show (TD.mean tdigest)
-- putStrLn $ "Stddev: " ++ show (TD.stddev tdigest)
-- forM_ [0.01, 0.1, 0.2, 0.5, 0.8, 0.9, 0.99] $ \q ->
-- putStrLn $ "q" ++ show q ++ ": " ++ show (TD.quantile q tdigest)
let Just stddev = TD.stddev tdigest
-- standard deviation of 100k is ok, we generate 256M of data after all.
-- On my machine deviation is 40k-50k
stddev `shouldSatisfy` (< 100000)
memoryUsage :: SourceT IO BS.ByteString -> IO (TD.TDigest 25)
memoryUsage src = unSourceT src $ loop mempty (0 :: Int)
where
loop !acc !_ Stop = return acc
loop !_ !_ (Error err) = fail err -- !
loop !acc !n (Skip s) = loop acc n s
loop !acc !n (Effect ms) = ms >>= loop acc n
loop !acc !n (Yield _bs s) = do
usage <- liftIO getUsage
-- We perform GC in between as we generate garbage.
when (n `mod` 1024 == 0) $ liftIO performGC
loop (TD.insert usage acc) (n + 1) s
getUsage :: IO Double
getUsage = fromIntegral .
#if MIN_VERSION_base(4,10,0)
memUsed <- gcdetails_mem_in_use_bytes . gc <$> getRTSStats
gcdetails_live_bytes . gc <$> getRTSStats
#else
memUsed <- currentBytesUsed <$> getGCStats
currentBytesUsed <$> getGCStats
#endif
memUsed `shouldSatisfy` (< megabytes 22)
-}
megabytes :: Num a => a -> a
megabytes n = n * (1000 ^ (2 :: Int))
-}

16
servant-conduit/CHANGELOG.md Normal file
View file

@ -0,0 +1,16 @@
0.10
----
* `EncodeOpts` and `DecodeOpts` merged into just `EncodeOpts`.
* Add `TabSeparatedOpts` for `text/tab-separated-values`.
0.9
---
* Refactorerd `CSV`' type.
0.8
---
* Removed DefaultEncodeOpts and DefaultDecodeOpts in favor of a single
(new) DefaultOpts

30
servant-conduit/LICENSE Normal file
View file

@ -0,0 +1,30 @@
Copyright (c) 2018, Servant Contributors
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 Servant Contributors 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.

3
servant-conduit/README.md Normal file
View file

@ -0,0 +1,3 @@
# servant-conduit - Servant Stream support for conduit
![servant](https://raw.githubusercontent.com/haskell-servant/servant/master/servant.png)

2
servant-conduit/Setup.hs Normal file
View file

@ -0,0 +1,2 @@
import Distribution.Simple
main = defaultMain

103
servant-conduit/example/Main.hs Normal file
View file

@ -0,0 +1,103 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TypeOperators #-}
module Main (main) where
import Prelude ()
import Prelude.Compat
import Control.Concurrent
(threadDelay)
import Control.Monad.IO.Class
(MonadIO (..))
import Control.Monad.Trans.Resource
(ResourceT)
import qualified Data.ByteString as BS
import Data.Maybe
(fromMaybe)
import Network.HTTP.Client
(defaultManagerSettings, newManager)
import Network.Wai
(Application)
import System.Environment
(getArgs, lookupEnv)
import Text.Read
(readMaybe)
import Data.Conduit
import qualified Data.Conduit.Combinators as C
import Servant
import Servant.Client
import Servant.Conduit ()
import qualified Network.Wai.Handler.Warp as Warp
type FastAPI = "get" :> Capture "num" Int :> StreamGet NewlineFraming JSON (ConduitT () Int IO ())
type API = FastAPI
:<|> "slow" :> Capture "num" Int :> StreamGet NewlineFraming JSON (ConduitT () Int IO ())
-- monad can be ResourceT IO too.
:<|> "readme" :> StreamGet NoFraming OctetStream (ConduitT () BS.ByteString (ResourceT IO) ())
-- we can have streaming request body
:<|> "proxy"
:> StreamBody NoFraming OctetStream (ConduitT () BS.ByteString IO ())
:> StreamPost NoFraming OctetStream (ConduitT () BS.ByteString IO ())
api :: Proxy API
api = Proxy
server :: Server API
server = fast :<|> slow :<|> readme :<|> proxy
where
fast n = liftIO $ do
putStrLn $ "/get/" ++ show n
return $ fastConduit n
slow n = liftIO $ do
putStrLn $ "/slow/" ++ show n
return $ slowConduit n
readme = liftIO $ do
putStrLn "/proxy"
return (C.sourceFile "README.md")
proxy c = liftIO $ do
putStrLn "/proxy"
return c
-- for some reason unfold leaks?
fastConduit = C.unfold mk where
mk m
| m < 0 = Nothing
| otherwise = Just (m, pred m)
slowConduit m = fastConduit m .| C.mapM (<$ threadDelay 1000000)
app :: Application
app = serve api server
cli :: Client ClientM FastAPI
cli :<|> _ :<|> _ :<|> _ = client api
main :: IO ()
main = do
args <- getArgs
case args of
("server":_) -> do
putStrLn "Starting servant-conduit:example at http://localhost:8000"
port <- fromMaybe 8000 . (>>= readMaybe) <$> lookupEnv "PORT"
Warp.run port app
("client":ns:_) -> do
n <- maybe (fail $ "not a number: " ++ ns) pure $ readMaybe ns
mgr <- newManager defaultManagerSettings
burl <- parseBaseUrl "http://localhost:8000/"
withClientM (cli n) (mkClientEnv mgr burl) $ \me -> case me of
Left err -> print err
Right c -> do
x <- connect c $ C.foldl (\p _ -> p + 1) (0 :: Int)
print x
_ -> do
putStrLn "Try:"
putStrLn "cabal new-run servant-conduit:example server"
putStrLn "cabal new-run servant-conduit:example client 10"
putStrLn "time curl -H 'Accept: application/json' localhost:8000/slow/5"

59
servant-conduit/servant-conduit.cabal Normal file
View file

@ -0,0 +1,59 @@
name: servant-conduit
version: 1
synopsis: Servant Stream support for conduit.
description: Servant Stream support for conduit.
homepage: http://haskell-servant.readthedocs.org/
license: BSD3
license-file: LICENSE
author: Servant Contributors
maintainer: haskell-servant-maintainers@googlegroups.com
copyright: 2018 Servant Contributors
category: Web, Servant, Enumerator
build-type: Simple
cabal-version: >=1.10
bug-reports: http://github.com/haskell-servant/servant-conduit/issues
tested-with:
GHC ==8.0.2
|| ==8.2.2
|| ==8.4.4
|| ==8.6.1
source-repository head
type: git
location: http://github.com/haskell-servant/servant-conduit.git
library
exposed-modules: Servant.Conduit
build-depends:
base >=4.9 && <5
, bytestring >=0.10.4.0 && <0.11
, conduit >=1.3.0.2 && <1.4
, mtl >=2.1 && <2.3
, resourcet >=1.2.1 && <1.3
, servant >=0.14 && <0.15
, unliftio-core >=0.1.1 && <0.2
hs-source-dirs: src
default-language: Haskell2010
ghc-options: -Wall
test-suite example
type: exitcode-stdio-1.0
main-is: Main.hs
hs-source-dirs:
example
ghc-options: -Wall -rtsopts -threaded
build-depends:
base
, base-compat
, bytestring
, conduit
, http-media
, resourcet
, servant
, servant-conduit
, servant-server >=0.14 && <0.15
, servant-client >=0.14 && <0.15
, wai >=3.0.3.0 && <3.3
, warp >=3.0.13.1 && <3.3
, http-client
default-language: Haskell2010

74
servant-conduit/src/Servant/Conduit.hs Normal file
View file

@ -0,0 +1,74 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE UndecidableInstances #-}
{-# OPTIONS_GHC -Wno-orphans #-}
-- | TBW
--
-- This module exports 'ToSourceIO' and 'FromSourceIO' instances.
module Servant.Conduit (
ConduitToSourceIO (..),
) where
import Control.Monad.IO.Class
(MonadIO (..))
import Control.Monad.IO.Unlift
(MonadUnliftIO (..))
import Control.Monad.Trans.Resource
(ResourceT, runResourceT)
import Data.Conduit.Internal
(ConduitT (..), Pipe (..))
import Servant.API.Stream
import qualified Servant.Types.SourceT as S
-- | Helper class to implement @'ToSourceIO' 'ConduitT'@ instance
-- for various monads.
class ConduitToSourceIO m where
conduitToSourceIO :: ConduitT i o m () -> SourceIO o
instance ConduitToSourceIO IO where
conduitToSourceIO (ConduitT con) = S.SourceT ($ go (con Done)) where
go p0 = case p0 of
Done () -> S.Stop
HaveOutput p o -> S.Yield o (go p)
NeedInput _ip up -> S.Skip (go (up ()))
PipeM m -> S.Effect $ fmap go m
Leftover p _l -> S.Skip (go p)
instance m ~ IO => ConduitToSourceIO (ResourceT m) where
conduitToSourceIO (ConduitT con) =
S.SourceT $ \k ->
runResourceT $ withRunInIO $ \runRes ->
k (go runRes (con Done))
where
go :: (forall x. ResourceT m x -> m x)
-> Pipe i i o () (ResourceT m) ()
-> S.StepT IO o
go _ (Done ()) = S.Stop
go runRes (HaveOutput p o) = S.Yield o (go runRes p)
go runRes (NeedInput _ip up) = S.Skip (go runRes (up ()))
go runRes (PipeM m) = S.Effect $ runRes $ fmap (go runRes) m
go runRes (Leftover p _l) = S.Skip (go runRes p)
instance (ConduitToSourceIO m, r ~ ())
=> ToSourceIO o (ConduitT i o m r)
where
toSourceIO = conduitToSourceIO
instance (MonadIO m, r ~ ()) => FromSourceIO o (ConduitT i o m r) where
fromSourceIO src =
ConduitT $ \con ->
PipeM $ liftIO $ S.unSourceT src $ \step ->
loop con step
where
loop :: MonadIO m => (() -> Pipe i i o () m b) -> S.StepT IO o -> IO (Pipe i i o () m b)
loop con S.Stop = return (con ())
loop _con (S.Error err) = fail err
loop con (S.Skip s) = loop con s
loop con (S.Effect ms) = ms >>= loop con
loop con (S.Yield x s) = return (HaveOutput (PipeM (liftIO $ loop con s)) x)
{-# SPECIALIZE INLINE fromSourceIO :: SourceIO o -> ConduitT i o IO () #-}

9
servant-docs/src/Servant/Docs/Internal.hs
View file

@ -853,9 +853,10 @@ instance {-# OVERLAPPABLE #-}
-- | TODO: mention the endpoint is streaming, its framing strategy
--
-- Also there are no samples.
--
-- TODO: AcceptFraming for content-type
instance {-# OVERLAPPABLE #-}
(MimeRender ct a, KnownNat status
, ReflectMethod method)
(Accept ct, KnownNat status, ReflectMethod method)
=> HasDocs (Stream method status framing ct a) where
docsFor Proxy (endpoint, action) DocOptions{..} =
single endpoint' action'
@ -866,7 +867,6 @@ instance {-# OVERLAPPABLE #-}
t = Proxy :: Proxy '[ct]
method' = reflectMethod (Proxy :: Proxy method)
status = fromInteger $ natVal (Proxy :: Proxy status)
p = Proxy :: Proxy a
instance {-# OVERLAPPING #-}
(ToSample a, AllMimeRender (ct ': cts) a, KnownNat status
@ -969,6 +969,9 @@ instance (ToSample a, AllMimeRender (ct ': cts) a, HasDocs api)
t = Proxy :: Proxy (ct ': cts)
p = Proxy :: Proxy a
instance HasDocs api => HasDocs (StreamBody framing ctype a :> api) where
docsFor Proxy _ _ = error "HasDocs @StreamBody"
instance (KnownSymbol path, HasDocs api) => HasDocs (path :> api) where
docsFor Proxy (endpoint, action) =

8
servant-foreign/src/Servant/Foreign/Internal.hs
View file

@ -322,6 +322,14 @@ instance (Elem JSON list, HasForeignType lang ftype a, HasForeign lang ftype api
foreignFor lang ftype (Proxy :: Proxy api) $
req & reqBody .~ (Just $ typeFor lang ftype (Proxy :: Proxy a))
instance
( HasForeign lang ftype api
) => HasForeign lang ftype (StreamBody framing ctype a :> api)
where
type Foreign ftype (StreamBody framing ctype a :> api) = Foreign ftype api
foreignFor _lang Proxy Proxy _req = error "HasForeign @StreamBody"
instance (KnownSymbol path, HasForeign lang ftype api)
=> HasForeign lang ftype (path :> api) where
type Foreign ftype (path :> api) = Foreign ftype api

5
servant-foreign/test/Servant/ForeignSpec.hs
View file

@ -16,6 +16,8 @@ import Data.Monoid
import Data.Proxy
import Servant.API.Internal.Test.ComprehensiveAPI
import Servant.Foreign
import Servant.Types.SourceT
(SourceT)
import Test.Hspec
@ -51,6 +53,9 @@ instance HasForeignType LangX String (Headers ctyps NoContent) where
instance HasForeignType LangX String Int where
typeFor _ _ _ = "intX"
instance HasForeignType LangX String (SourceT m a) where
typeFor _ _ _ = "streamTX"
instance HasForeignType LangX String Bool where
typeFor _ _ _ = "boolX"

16
servant-machines/CHANGELOG.md Normal file
View file

@ -0,0 +1,16 @@
0.10
----
* `EncodeOpts` and `DecodeOpts` merged into just `EncodeOpts`.
* Add `TabSeparatedOpts` for `text/tab-separated-values`.
0.9
---
* Refactorerd `CSV`' type.
0.8
---
* Removed DefaultEncodeOpts and DefaultDecodeOpts in favor of a single
(new) DefaultOpts

30
servant-machines/LICENSE Normal file
View file

@ -0,0 +1,30 @@
Copyright (c) 2018, Servant Contributors
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 Servant Contributors 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.

3
servant-machines/README.md Normal file
View file

@ -0,0 +1,3 @@
# servant-machines - Servant Stream support for machines
![servant](https://raw.githubusercontent.com/haskell-servant/servant/master/servant.png)

2
servant-machines/Setup.hs Normal file
View file

@ -0,0 +1,2 @@
import Distribution.Simple
main = defaultMain

98
servant-machines/example/Main.hs Normal file
View file

@ -0,0 +1,98 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TypeOperators #-}
module Main (main) where
import Prelude ()
import Prelude.Compat
import Control.Concurrent
(threadDelay)
import Control.Monad.IO.Class
(MonadIO (..))
import qualified Data.ByteString as BS
import Data.Maybe
(fromMaybe)
import Data.Void
(Void)
import Network.HTTP.Client
(defaultManagerSettings, newManager)
import Network.Wai
(Application)
import System.Environment
(getArgs, lookupEnv)
import Text.Read
(readMaybe)
import Data.Machine
import Servant
import Servant.Client
import Servant.Machines ()
import qualified Network.Wai.Handler.Warp as Warp
type FastAPI = "get" :> Capture "num" Int :> StreamGet NewlineFraming JSON (MachineT IO (Is Void) Int)
type API = FastAPI
:<|> "slow" :> Capture "num" Int :> StreamGet NewlineFraming JSON (MachineT IO (Is Void) Int)
:<|> "proxy"
:> StreamBody NoFraming OctetStream (MachineT IO (Is Void) BS.ByteString)
:> StreamPost NoFraming OctetStream (MachineT IO (Is Void) BS.ByteString)
api :: Proxy API
api = Proxy
server :: Server API
server = fast :<|> slow :<|> proxy
where
fast n = liftIO $ do
putStrLn ("/get/" ++ show n)
return $ fastMachine n
slow n = liftIO $ do
putStrLn ("/slow/" ++ show n)
return $ slowMachine n
proxy c = liftIO $ do
putStrLn "/proxy"
return c
-- for some reason unfold leaks?
fastMachine m
| m < 0 = MachineT (return Stop)
| otherwise = MachineT (return (Yield m (fastMachine (m - 1))))
slowMachine m
| m < 0 = MachineT (return Stop)
| otherwise = MachineT $ do
threadDelay 1000000
return (Yield m (slowMachine (m - 1)))
app :: Application
app = serve api server
cli :: Client ClientM FastAPI
cli :<|> _ = client api
main :: IO ()
main = do
args <- getArgs
case args of
("server":_) -> do
putStrLn "Starting servant-machines:example at http://localhost:8000"
port <- fromMaybe 8000 . (>>= readMaybe) <$> lookupEnv "PORT"
Warp.run port app
("client":ns:_) -> do
n <- maybe (fail $ "not a number: " ++ ns) pure $ readMaybe ns
mgr <- newManager defaultManagerSettings
burl <- parseBaseUrl "http://localhost:8000/"
withClientM (cli n) (mkClientEnv mgr burl) $ \me -> case me of
Left err -> print err
Right m -> do
x <- runT $ fold (\p _ -> p + 1) (0 :: Int) <~ m
print x
_ -> do
putStrLn "Try:"
putStrLn "cabal new-run servant-machines:example server"
putStrLn "cabal new-run servant-machines:example client 10"
putStrLn "time curl -H 'Accept: application/json' localhost:8000/slow/5"

56
servant-machines/servant-machines.cabal Normal file
View file

@ -0,0 +1,56 @@
name: servant-machines
version: 1
synopsis: Servant Stream support for machines
description: Servant Stream support for machines.
homepage: http://haskell-servant.readthedocs.org/
license: BSD3
license-file: LICENSE
author: Servant Contributors
maintainer: haskell-servant-maintainers@googlegroups.com
copyright: 2018 Servant Contributors
category: Web, Servant, Enumerator
build-type: Simple
cabal-version: >=1.10
bug-reports: http://github.com/haskell-servant/servant-machines/issues
tested-with:
GHC ==8.0.2
|| ==8.2.2
|| ==8.4.4
|| ==8.6.1
source-repository head
type: git
location: http://github.com/haskell-servant/servant-machines.git
library
exposed-modules: Servant.Machines
build-depends:
base >=4.9 && <5
, bytestring >=0.10.4.0 && <0.11
, machines >=0.6.3 && <0.7
, mtl >=2.1 && <2.3
, servant >=0.14 && <0.15
hs-source-dirs: src
default-language: Haskell2010
ghc-options: -Wall
test-suite example
type: exitcode-stdio-1.0
main-is: Main.hs
hs-source-dirs:
example
ghc-options: -Wall -rtsopts -threaded
build-depends:
base
, base-compat
, bytestring
, http-media
, servant
, machines
, servant-machines
, servant-server >=0.14 && <0.15
, servant-client >=0.14 && <0.15
, wai >=3.0.3.0 && <3.3
, warp >=3.0.13.1 && <3.3
, http-client
default-language: Haskell2010

48
servant-machines/src/Servant/Machines.hs Normal file
View file

@ -0,0 +1,48 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE UndecidableInstances #-}
{-# OPTIONS_GHC -Wno-orphans #-}
-- | TBW
--
-- This module exports 'ToSourceIO' and 'FromSourceIO' instances.
module Servant.Machines (
MachineToSourceIO (..),
) where
import Control.Monad.IO.Class
(MonadIO (..))
import Data.Machine
(MachineT (..), Step (..))
import Servant.API.Stream
import qualified Servant.Types.SourceT as S
-- | Helper class to implement @'ToSourceIO' 'MachineT'@ instance
-- for various monads.
class MachineToSourceIO m where
machineToSourceIO :: MachineT m k o -> S.SourceT IO o
instance MachineToSourceIO IO where
machineToSourceIO ma = S.SourceT ($ go ma) where
go (MachineT m) = S.Effect $ do
step <- m
case step of
Stop -> return S.Stop
Yield x m' -> return (S.Yield x (go m'))
Await _ _ m' -> return (S.Skip (go m'))
instance MachineToSourceIO m => ToSourceIO o (MachineT m k o) where
toSourceIO = machineToSourceIO
instance MonadIO m => FromSourceIO o (MachineT m k o) where
fromSourceIO src = MachineT $ liftIO $ S.unSourceT src go
where
go :: S.StepT IO o -> IO (Step k o (MachineT m k o))
go S.Stop = return Stop
go (S.Error err) = fail err
go (S.Skip s) = go s
go (S.Effect ms) = ms >>= go
go (S.Yield x s) = return (Yield x (MachineT (liftIO (go s))))
{-# SPECIALIZE INLINE fromSourceIO :: SourceIO o -> MachineT IO k o #-}

16
servant-pipes/CHANGELOG.md Normal file
View file

@ -0,0 +1,16 @@
0.10
----
* `EncodeOpts` and `DecodeOpts` merged into just `EncodeOpts`.
* Add `TabSeparatedOpts` for `text/tab-separated-values`.
0.9
---
* Refactorerd `CSV`' type.
0.8
---
* Removed DefaultEncodeOpts and DefaultDecodeOpts in favor of a single
(new) DefaultOpts

30
servant-pipes/LICENSE Normal file
View file

@ -0,0 +1,30 @@
Copyright (c) 2018, Servant Contributors
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 Servant Contributors 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.

3
servant-pipes/README.md Normal file
View file

@ -0,0 +1,3 @@
# servant-pipes - Servant Stream support for pipes
![servant](https://raw.githubusercontent.com/haskell-servant/servant/master/servant.png)

2
servant-pipes/Setup.hs Normal file
View file

@ -0,0 +1,2 @@
import Distribution.Simple
main = defaultMain

106
servant-pipes/example/Main.hs Normal file
View file

@ -0,0 +1,106 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TypeOperators #-}
module Main (main) where
import Prelude ()
import Prelude.Compat
import Control.Concurrent
(threadDelay)
import Control.Monad.IO.Class
(MonadIO (..))
import qualified Data.ByteString as BS
import Data.Maybe
(fromMaybe)
import Network.HTTP.Client
(defaultManagerSettings, newManager)
import Network.Wai
(Application)
import System.Environment
(getArgs, lookupEnv)
import System.IO (IOMode (..))
import Text.Read
(readMaybe)
import qualified Pipes as P
import qualified Pipes.Prelude as P
import Pipes.Safe
(SafeT)
import qualified Pipes.Safe.Prelude as P
import Servant
import Pipes.ByteString as PBS
import Servant.Client
import Servant.Pipes ()
import qualified Network.Wai.Handler.Warp as Warp
type FastAPI = "get" :> Capture "num" Int :> StreamGet NewlineFraming JSON (P.Producer Int IO ())
-- TODO: Change IO to something with MonadError ServantUnrenderError
type API = FastAPI
:<|> "slow" :> Capture "num" Int :> StreamGet NewlineFraming JSON (P.Producer Int IO ())
-- monad can be SafeT IO too.
:<|> "readme" :> StreamGet NoFraming OctetStream (P.Producer BS.ByteString (SafeT IO) ())
-- we can have streaming request body
:<|> "proxy"
:> StreamBody NoFraming OctetStream (P.Producer BS.ByteString IO ())
:> StreamPost NoFraming OctetStream (P.Producer BS.ByteString IO ())
api :: Proxy API
api = Proxy
server :: Server API
server = fast :<|> slow :<|> readme :<|> proxy
where
fast n = liftIO $ do
putStrLn ("/get/" ++ show n)
return $ fastPipe n
slow n = liftIO $ do
putStrLn ("/slow/" ++ show n)
return $ slowPipe n
readme = liftIO $ do
putStrLn "/readme"
return $ P.withFile "README.md" ReadMode PBS.fromHandle
proxy c = liftIO $ do
putStrLn "/proxy"
return c
-- for some reason unfold leaks?
fastPipe m
| m < 0 = return ()
| otherwise = P.yield m >> fastPipe (m - 1)
slowPipe m = fastPipe m P.>-> P.mapM (<$ threadDelay 1000000)
app :: Application
app = serve api server
cli :: Client ClientM FastAPI
cli :<|> _ :<|> _ :<|> _ = client api
main :: IO ()
main = do
args <- getArgs
case args of
("server":_) -> do
putStrLn "Starting servant-pipes:example at http://localhost:8000"
port <- fromMaybe 8000 . (>>= readMaybe) <$> lookupEnv "PORT"
Warp.run port app
("client":ns:_) -> do
n <- maybe (fail $ "not a number: " ++ ns) pure $ readMaybe ns
mgr <- newManager defaultManagerSettings
burl <- parseBaseUrl "http://localhost:8000/"
withClientM (cli n) (mkClientEnv mgr burl) $ \me -> case me of
Left err -> print err
Right p -> do
x <- P.fold (\c _ -> c + 1) (0 :: Int) id p
print x
_ -> do
putStrLn "Try:"
putStrLn "cabal new-run servant-pipes:example server"
putStrLn "cabal new-run servant-pipes:example client 10"
putStrLn "time curl -H 'Accept: application/json' localhost:8000/slow/5"

60
servant-pipes/servant-pipes.cabal Normal file
View file

@ -0,0 +1,60 @@
name: servant-pipes
version: 1
synopsis: Servant Stream support for pipes
description: Servant Stream support for pipes.
homepage: http://haskell-servant.readthedocs.org/
license: BSD3
license-file: LICENSE
author: Servant Contributors
maintainer: haskell-servant-maintainers@googlegroups.com
copyright: 2018 Servant Contributors
category: Web, Servant, Pipes
build-type: Simple
cabal-version: >=1.10
bug-reports: http://github.com/haskell-servant/servant-pipes/issues
tested-with:
GHC ==8.0.2
|| ==8.2.2
|| ==8.4.4
|| ==8.6.1
source-repository head
type: git
location: http://github.com/haskell-servant/servant-pipes.git
library
exposed-modules: Servant.Pipes
build-depends:
base >=4.9 && <5
, bytestring >=0.10.4.0 && <0.11
, pipes >=4.3.9 && <4.4
, pipes-safe >=2.3.1 && <2.4
, mtl >=2.1 && <2.3
, monad-control >=1.0.2.3 && <1.1
, servant >=0.14 && <0.15
hs-source-dirs: src
default-language: Haskell2010
ghc-options: -Wall
test-suite example
type: exitcode-stdio-1.0
main-is: Main.hs
hs-source-dirs:
example
ghc-options: -Wall -rtsopts -threaded
build-depends:
base
, base-compat
, bytestring
, http-media
, servant
, pipes
, pipes-safe
, servant-pipes
, pipes-bytestring >= 2.1.6 && <2.2
, servant-server >=0.14 && <0.15
, servant-client >=0.14 && <0.15
, wai >=3.0.3.0 && <3.3
, warp >=3.0.13.1 && <3.3
, http-client
default-language: Haskell2010

77
servant-pipes/src/Servant/Pipes.hs Normal file
View file

@ -0,0 +1,77 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE UndecidableInstances #-}
{-# OPTIONS_GHC -Wno-orphans #-}
-- | TBW
--
-- This module exports 'ToSourceIO' and 'FromSourceIO' instances.
module Servant.Pipes (
PipesToSourceIO (..),
) where
import Control.Monad.IO.Class
(MonadIO (..))
import Control.Monad.Trans.Control
(liftBaseWith)
import Pipes
(ListT (..))
import Pipes.Internal
(Proxy (..), X, closed)
import Pipes.Safe
(SafeT, runSafeT)
import Servant.API.Stream
import qualified Servant.Types.SourceT as S
-- | Helper class to implement @'ToSourceIO' 'Proxy'@ instance
-- for various monads.
class PipesToSourceIO m where
pipesToSourceIO :: Proxy X () () b m () -> SourceIO b
instance PipesToSourceIO IO where
pipesToSourceIO ma = S.SourceT ($ go ma) where
go :: Proxy X () () b IO () -> S.StepT IO b
go (Pure ()) = S.Stop
go (M p) = S.Effect (fmap go p)
go (Request v _) = closed v
go (Respond b n) = S.Yield b (go (n ()))
instance m ~ IO => PipesToSourceIO (SafeT m) where
pipesToSourceIO ma =
S.SourceT $ \k ->
runSafeT $ liftBaseWith $ \runSafe ->
k (go runSafe ma)
where
go :: (forall x. SafeT m x -> m x)
-> Proxy X () () b (SafeT m) ()
-> S.StepT IO b
go _ (Pure ()) = S.Stop
go runSafe (M p) = S.Effect $ runSafe $ fmap (go runSafe) p
go _ (Request v _) = closed v
go runSafe (Respond b n) = S.Yield b (go runSafe (n ()))
instance (PipesToSourceIO m, a' ~ X, a ~ (), b' ~ (), r ~ ())
=> ToSourceIO b (Proxy a' a b' b m r)
where
toSourceIO = pipesToSourceIO
instance PipesToSourceIO m => ToSourceIO a (ListT m a) where
toSourceIO = pipesToSourceIO . enumerate
instance (MonadIO m, a' ~ X, a ~ (), b' ~ (), r ~ ())
=> FromSourceIO b (Proxy a' a b' b m r)
where
fromSourceIO src = M $ liftIO $ S.unSourceT src (return . go) where
go :: S.StepT IO b -> Proxy X () () b m ()
go S.Stop = Pure ()
go (S.Error err) = M (fail err)
go (S.Skip s) = go s -- drives
go (S.Effect ms) = M (liftIO (fmap go ms))
go (S.Yield x s) = Respond x (const (go s))
{-# SPECIALIZE INLINE fromSourceIO :: SourceIO x -> Proxy X () () x IO () #-}
instance MonadIO m => FromSourceIO a (ListT m a) where
fromSourceIO = Select . fromSourceIO

105
servant-server/src/Servant/Server/Internal.hs
View file

@ -61,27 +61,27 @@ import Network.Socket
(SockAddr)
import Network.Wai
(Application, Request, httpVersion, isSecure, lazyRequestBody,
rawQueryString, remoteHost, requestHeaders, requestMethod,
responseLBS, responseStream, vault)
rawQueryString, remoteHost, requestBody, requestHeaders,
requestMethod, responseLBS, responseStream, vault)
import Prelude ()
import Prelude.Compat
import Servant.API
((:<|>) (..), (:>), Accept (..), BasicAuth,
BoundaryStrategy (..), Capture', CaptureAll, Description,
EmptyAPI, FramingRender (..), Header', If, IsSecure (..),
QueryFlag, QueryParam', QueryParams, Raw,
((:<|>) (..), (:>), Accept (..), BasicAuth, Capture',
CaptureAll, Description, EmptyAPI, FramingRender (..),
FramingUnrender (..), FromSourceIO (..), Header', If,
IsSecure (..), QueryFlag, QueryParam', QueryParams, Raw,
ReflectMethod (reflectMethod), RemoteHost, ReqBody',
SBool (..), SBoolI (..), Stream, StreamGenerator (..),
Summary, ToStreamGenerator (..), Vault, Verb,
WithNamedContext)
SBool (..), SBoolI (..), SourceIO, Stream, StreamBody,
Summary, ToSourceIO (..), Vault, Verb, WithNamedContext)
import Servant.API.ContentTypes
(AcceptHeader (..), AllCTRender (..), AllCTUnrender (..),
AllMime, MimeRender (..), canHandleAcceptH)
AllMime, MimeRender (..), MimeUnrender (..), canHandleAcceptH)
import Servant.API.Modifiers
(FoldLenient, FoldRequired, RequestArgument,
unfoldRequestArgument)
import Servant.API.ResponseHeaders
(GetHeaders, Headers, getHeaders, getResponse)
import qualified Servant.Types.SourceT as S
import Web.HttpApiData
(FromHttpApiData, parseHeader, parseQueryParam,
parseUrlPieceMaybe, parseUrlPieces)
@ -279,24 +279,25 @@ instance {-# OVERLAPPING #-}
instance {-# OVERLAPPABLE #-}
( MimeRender ctype a, ReflectMethod method, KnownNat status,
FramingRender framing ctype, ToStreamGenerator b a
) => HasServer (Stream method status framing ctype b) context where
( MimeRender ctype chunk, ReflectMethod method, KnownNat status,
FramingRender framing, ToSourceIO chunk a
) => HasServer (Stream method status framing ctype a) context where
type ServerT (Stream method status framing ctype b) m = m b
type ServerT (Stream method status framing ctype a) m = m a
hoistServerWithContext _ _ nt s = nt s
route Proxy _ = streamRouter ([],) method status (Proxy :: Proxy framing) (Proxy :: Proxy ctype)
where method = reflectMethod (Proxy :: Proxy method)
status = toEnum . fromInteger $ natVal (Proxy :: Proxy status)
instance {-# OVERLAPPING #-}
( MimeRender ctype a, ReflectMethod method, KnownNat status,
FramingRender framing ctype, ToStreamGenerator b a,
GetHeaders (Headers h b)
) => HasServer (Stream method status framing ctype (Headers h b)) context where
type ServerT (Stream method status framing ctype (Headers h b)) m = m (Headers h b)
instance {-# OVERLAPPING #-}
( MimeRender ctype chunk, ReflectMethod method, KnownNat status,
FramingRender framing, ToSourceIO chunk a,
GetHeaders (Headers h a)
) => HasServer (Stream method status framing ctype (Headers h a)) context where
type ServerT (Stream method status framing ctype (Headers h a)) m = m (Headers h a)
hoistServerWithContext _ _ nt s = nt s
route Proxy _ = streamRouter (\x -> (getHeaders x, getResponse x)) method status (Proxy :: Proxy framing) (Proxy :: Proxy ctype)
@ -304,8 +305,8 @@ instance {-# OVERLAPPING #-}
status = toEnum . fromInteger $ natVal (Proxy :: Proxy status)
streamRouter :: (MimeRender ctype a, FramingRender framing ctype, ToStreamGenerator b a) =>
(c -> ([(HeaderName, B.ByteString)], b))
streamRouter :: forall ctype a c chunk env framing. (MimeRender ctype chunk, FramingRender framing, ToSourceIO chunk a) =>
(c -> ([(HeaderName, B.ByteString)], a))
-> Method
-> Status
-> Proxy framing
@ -321,28 +322,19 @@ streamRouter splitHeaders method status framingproxy ctypeproxy action = leafRou
`addAcceptCheck` accCheck
) env request respond $ \ output ->
let (headers, fa) = splitHeaders output
k = getStreamGenerator . toStreamGenerator $ fa in
Route $ responseStream status (contentHeader : headers) $ \write flush -> do
write . BB.lazyByteString $ header framingproxy ctypeproxy
case boundary framingproxy ctypeproxy of
BoundaryStrategyBracket f ->
let go x = let bs = mimeRender ctypeproxy x
(before, after) = f bs
in write ( BB.lazyByteString before
<> BB.lazyByteString bs
<> BB.lazyByteString after) >> flush
in k go go
BoundaryStrategyIntersperse sep -> k
(\x -> do
write . BB.lazyByteString . mimeRender ctypeproxy $ x
flush)
(\x -> do
write . (BB.lazyByteString sep <>) . BB.lazyByteString . mimeRender ctypeproxy $ x
flush)
BoundaryStrategyGeneral f ->
let go = (>> flush) . write . BB.lazyByteString . f . mimeRender ctypeproxy
in k go go
write . BB.lazyByteString $ trailer framingproxy ctypeproxy
sourceT = toSourceIO fa
S.SourceT kStepLBS = framingRender framingproxy (mimeRender ctypeproxy :: chunk -> BL.ByteString) sourceT
in Route $ responseStream status (contentHeader : headers) $ \write flush -> do
let loop S.Stop = flush
loop (S.Error err) = fail err -- TODO: throw better error
loop (S.Skip s) = loop s
loop (S.Effect ms) = ms >>= loop
loop (S.Yield lbs s) = do
write (BB.lazyByteString lbs)
flush
loop s
kStepLBS loop
-- | If you use 'Header' in one of the endpoints for your API,
-- this automatically requires your server-side handler to be a function
@ -615,6 +607,31 @@ instance ( AllCTUnrender list a, HasServer api context, SBoolI (FoldLenient mods
Left e -> delayedFailFatal err400 { errBody = cs e }
Right v -> return v
instance
( FramingUnrender framing, FromSourceIO chunk a, MimeUnrender ctype chunk
, HasServer api context
) => HasServer (StreamBody framing ctype a :> api) context
where
type ServerT (StreamBody framing ctype a :> api) m = a -> ServerT api m
hoistServerWithContext _ pc nt s = hoistServerWithContext (Proxy :: Proxy api) pc nt . s
route Proxy context subserver = route (Proxy :: Proxy api) context $
addBodyCheck subserver ctCheck bodyCheck
where
ctCheck :: DelayedIO (SourceIO chunk -> a)
-- TODO: do content-type check
ctCheck = return fromSourceIO
bodyCheck :: (SourceIO chunk -> a) -> DelayedIO a
bodyCheck fromRS = withRequest $ \req -> do
let mimeUnrender' = mimeUnrender (Proxy :: Proxy ctype) :: BL.ByteString -> Either String chunk
let framingUnrender' = framingUnrender (Proxy :: Proxy framing) mimeUnrender' :: SourceIO B.ByteString -> SourceIO chunk
let body = requestBody req
let rs = S.fromAction B.null body
let rs' = fromRS $ framingUnrender' rs
return rs'
-- | Make sure the incoming request starts with @"/path"@, strip it and
-- pass the rest of the request path to @api@.
instance (KnownSymbol path, HasServer api context) => HasServer (path :> api) context where

7
servant-server/test/Servant/ServerSpec.hs
View file

@ -42,6 +42,7 @@ import Network.Wai
import Network.Wai.Test
(defaultRequest, request, runSession, simpleBody,
simpleHeaders, simpleStatus)
import qualified Servant.Types.SourceT as S
import Servant.API
((:<|>) (..), (:>), AuthProtect, BasicAuth,
BasicAuthData (BasicAuthData), Capture, CaptureAll, Delete,
@ -49,7 +50,7 @@ import Servant.API
JSON, NoContent (..), NoFraming, OctetStream, Patch,
PlainText, Post, Put, QueryFlag, QueryParam, QueryParams, Raw,
RemoteHost, ReqBody, StdMethod (..), Stream,
StreamGenerator (..), Verb, addHeader)
SourceIO, Verb, addHeader)
import Servant.API.Internal.Test.ComprehensiveAPI
import Servant.Server
(Context ((:.), EmptyContext), Handler, Server, Tagged (..),
@ -106,7 +107,7 @@ type VerbApi method status
:<|> "accept" :> ( Verb method status '[JSON] Person
:<|> Verb method status '[PlainText] String
)
:<|> "stream" :> Stream method status NoFraming OctetStream (StreamGenerator BS.ByteString)
:<|> "stream" :> Stream method status NoFraming OctetStream (SourceIO BS.ByteString)
verbSpec :: Spec
verbSpec = describe "Servant.API.Verb" $ do
@ -116,7 +117,7 @@ verbSpec = describe "Servant.API.Verb" $ do
:<|> return (addHeader 5 alice)
:<|> return (addHeader 10 NoContent)
:<|> (return alice :<|> return "B")
:<|> return (StreamGenerator $ \f _ -> f "bytestring")
:<|> return (S.source ["bytestring"])
get200 = Proxy :: Proxy (VerbApi 'GET 200)
post210 = Proxy :: Proxy (VerbApi 'POST 210)

19
servant/servant.cabal
View file

@ -24,6 +24,7 @@ tested-with:
GHC==8.6.1
extra-source-files:
CHANGELOG.md
source-repository head
type: git
location: http://github.com/haskell-servant/servant.git
@ -61,6 +62,13 @@ library
Servant.API.Vault
Servant.API.Verbs
Servant.API.WithNamedContext
-- Types
exposed-modules:
Servant.Types.SourceT
-- Safe links
exposed-modules:
Servant.Links
-- Deprecated modules, to be removed in late 2019
@ -75,6 +83,7 @@ library
base >= 4.9 && < 4.13
, bytestring >= 0.10.8.1 && < 0.11
, mtl >= 2.1 && < 2.3
, transformers >= 0.3.0.0 && < 0.6
, text >= 1.2.3.0 && < 1.3
if !impl(ghc >= 8.0)
@ -84,19 +93,19 @@ library
-- Other dependencies: Lower bound around what is in the latest Stackage LTS.
-- Here can be exceptions if we really need features from the newer versions.
build-depends:
base-compat >= 0.10.4 && < 0.11
, bifunctors >= 5.5.3 && < 5.6
base-compat >= 0.10.5 && < 0.11
, aeson >= 1.3.1.1 && < 1.5
, attoparsec >= 0.13.2.2 && < 0.14
, bifunctors >= 5.5.3 && < 5.6
, case-insensitive >= 1.2.0.11 && < 1.3
, http-api-data >= 0.3.8.1 && < 0.4
, http-media >= 0.7.1.2 && < 0.8
, http-types >= 0.12.1 && < 0.13
, mmorph >= 1.1.2 && < 1.2
, tagged >= 0.8.5 && < 0.9
, network-uri >= 2.6.1.0 && < 2.7
, singleton-bool >= 0.1.4 && < 0.2
, string-conversions >= 0.4.0.1 && < 0.5
, network-uri >= 2.6.1.0 && < 2.7
, tagged >= 0.8.5 && < 0.9
, vault >= 0.3.1.1 && < 0.4
hs-source-dirs: src
@ -132,6 +141,7 @@ test-suite spec
other-modules:
Servant.API.ContentTypesSpec
Servant.API.ResponseHeadersSpec
Servant.API.StreamSpec
Servant.LinksSpec
-- Dependencies inherited from the library. No need to specify bounds.
@ -143,6 +153,7 @@ test-suite spec
, servant
, string-conversions
, text
, transformers
-- Additonal dependencies
build-depends:

9
servant/src/Servant/API.hs
View file

@ -114,11 +114,10 @@ import Servant.API.ResponseHeaders
ResponseHeader (..), addHeader, getHeadersHList, getResponse,
noHeader)
import Servant.API.Stream
(BoundaryStrategy (..), ByteStringParser (..),
FramingRender (..), FramingUnrender (..),
FromResultStream (..), NetstringFraming, NewlineFraming,
NoFraming, ResultStream (..), Stream, StreamGenerator (..),
StreamGet, StreamPost, ToStreamGenerator (..))
(FramingRender (..), FramingUnrender (..),
FromSourceIO (..), NetstringFraming, NewlineFraming,
NoFraming, SourceIO, Stream, StreamBody, SourceIO,
StreamGet, StreamPost, ToSourceIO (..))
import Servant.API.Sub
((:>))
import Servant.API.Vault

4
servant/src/Servant/API/Internal/Test/ComprehensiveAPI.hs
View file

@ -9,6 +9,8 @@ module Servant.API.Internal.Test.ComprehensiveAPI where
import Data.Proxy
(Proxy (..))
import Servant.API
import Servant.Types.SourceT
(SourceT)
type GET = Get '[JSON] NoContent
@ -39,7 +41,7 @@ type ComprehensiveAPIWithoutRaw =
Vault :> GET :<|>
Verb 'POST 204 '[JSON] NoContent :<|>
Verb 'POST 204 '[JSON] Int :<|>
Stream 'GET 200 NetstringFraming JSON [Int] :<|>
StreamBody NetstringFraming JSON (SourceT IO Int) :> Stream 'GET 200 NetstringFraming JSON (SourceT IO Int) :<|>
WithNamedContext "foo" '[] GET :<|>
CaptureAll "foo" Int :> GET :<|>
Summary "foo" :> GET :<|>

330
servant/src/Servant/API/Stream.hs
View file

@ -1,6 +1,7 @@
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE FunctionalDependencies #-}
{-# LANGUAGE KindSignatures #-}
@ -8,39 +9,41 @@
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE PolyKinds #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE TupleSections #-}
{-# OPTIONS_HADDOCK not-home #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# OPTIONS_HADDOCK not-home #-}
module Servant.API.Stream (
Stream,
StreamGet,
StreamPost,
-- * Sources
--
-- | Both 'StreamGenerator' and 'ResultStream' are equivalent
-- to some *source* in streaming libraries.
StreamGenerator (..),
ToStreamGenerator (..),
ResultStream (..),
FromResultStream (..),
-- * Framing
FramingRender (..),
FramingUnrender (..),
BoundaryStrategy (..),
ByteStringParser (..),
-- ** Strategies
NoFraming,
NewlineFraming,
NetstringFraming,
) where
Stream,
StreamGet,
StreamPost,
StreamBody,
-- * Source
--
-- | 'SourceIO' are equivalent to some *source* in streaming libraries.
SourceIO,
ToSourceIO (..),
FromSourceIO (..),
-- ** Auxiliary classes
SourceToSourceIO (..),
-- * Framing
FramingRender (..),
FramingUnrender (..),
-- ** Strategies
NoFraming,
NewlineFraming,
NetstringFraming,
) where
import Control.Arrow
(first)
import Data.ByteString.Lazy
(ByteString, empty)
import qualified Data.ByteString.Lazy.Char8 as LB
import Data.Foldable
(traverse_)
import Control.Applicative
((<|>))
import Control.Monad.IO.Class
(MonadIO (..))
import qualified Data.Attoparsec.ByteString as A
import qualified Data.Attoparsec.ByteString.Char8 as A8
import qualified Data.ByteString as BS
import qualified Data.ByteString.Lazy as LBS
import qualified Data.ByteString.Lazy.Char8 as LBS8
import Data.List.NonEmpty
(NonEmpty (..))
import Data.Monoid
@ -55,158 +58,185 @@ import GHC.TypeLits
(Nat)
import Network.HTTP.Types.Method
(StdMethod (..))
import System.IO.Unsafe
(unsafeInterleaveIO)
import Text.Read
(readMaybe)
import Servant.Types.SourceT
-- | A Stream endpoint for a given method emits a stream of encoded values at a
-- given Content-Type, delimited by a framing strategy. Stream endpoints always
-- return response code 200 on success. Type synonyms are provided for standard
-- methods.
-- given @Content-Type@, delimited by a @framing@ strategy.
-- Type synonyms are provided for standard methods.
--
data Stream (method :: k1) (status :: Nat) (framing :: *) (contentType :: *) (a :: *)
deriving (Typeable, Generic)
type StreamGet = Stream 'GET 200
type StreamPost = Stream 'POST 200
-- | Stream endpoints may be implemented as producing a @StreamGenerator@ a
-- function that itself takes two emit functions the first to be used on the
-- first value the stream emits, and the second to be used on all subsequent
-- values (to allow interspersed framing strategies such as comma separation).
newtype StreamGenerator a = StreamGenerator { getStreamGenerator :: (a -> IO ()) -> (a -> IO ()) -> IO () }
-- | A stream request body.
--
-- TODO: add mods
data StreamBody (framing :: *) (contentType :: *) (a :: *)
deriving (Typeable, Generic)
-- | ToStreamGenerator is intended to be implemented for types such as Conduit, Pipe, etc. By implementing this class, all such streaming abstractions can be used directly as endpoints.
class ToStreamGenerator a b | a -> b where
toStreamGenerator :: a -> StreamGenerator b
-------------------------------------------------------------------------------
-- Sources
-------------------------------------------------------------------------------
instance ToStreamGenerator (StreamGenerator a) a where
toStreamGenerator x = x
-- | Stream endpoints may be implemented as producing a @'SourceIO' chunk@.
--
-- Clients reading from streaming endpoints can be implemented as consuming a
-- @'SourceIO' chunk@.
--
type SourceIO = SourceT IO
instance ToStreamGenerator (NonEmpty a) a where
toStreamGenerator (x :| xs) = StreamGenerator $ \f g -> f x >> traverse_ g xs
-- | 'ToSourceIO' is intended to be implemented for types such as Conduit, Pipe,
-- etc. By implementing this class, all such streaming abstractions can be used
-- directly as endpoints.
class ToSourceIO chunk a | a -> chunk where
toSourceIO :: a -> SourceIO chunk
instance ToStreamGenerator [a] a where
toStreamGenerator [] = StreamGenerator $ \_ _ -> return ()
toStreamGenerator (x : xs) = StreamGenerator $ \f g -> f x >> traverse_ g xs
-- | Auxiliary class for @'ToSourceIO' x ('SourceT' m x)@ instance.
class SourceToSourceIO m where
sourceToSourceIO :: SourceT m a -> SourceT IO a
-- | Clients reading from streaming endpoints can be implemented as producing a
-- @ResultStream@ that captures the setup, takedown, and incremental logic for
-- a read, being an IO continuation that takes a producer of Just either values
-- or errors that terminates with a Nothing.
newtype ResultStream a = ResultStream { runResultStream :: forall b. (IO (Maybe (Either String a)) -> IO b) -> IO b }
instance SourceToSourceIO IO where
sourceToSourceIO = id
-- | FromResultStream is intended to be implemented for types such as Conduit, Pipe, etc. By implementing this class, all such streaming abstractions can be used directly on the client side for talking to streaming endpoints.
class FromResultStream a b | b -> a where
fromResultStream :: ResultStream a -> IO b
-- | Relax to use auxiliary class, have m
instance SourceToSourceIO m => ToSourceIO chunk (SourceT m chunk) where
toSourceIO = sourceToSourceIO
instance FromResultStream a (ResultStream a) where
fromResultStream = return
instance ToSourceIO a (NonEmpty a) where
toSourceIO (x :| xs) = fromStepT (Yield x (foldr Yield Stop xs))
-- | Uses 'unsafeInterleaveIO'
instance FromResultStream a [a] where
fromResultStream x = runResultStream x lazyRead
instance ToSourceIO a [a] where
toSourceIO = source
-- | Uses 'unsafeInterleaveIO'
instance FromResultStream a (NonEmpty a) where
fromResultStream x = runResultStream x $ \r -> do
e <- r
case e of
Nothing -> fail "Empty stream"
Just (Left er) -> fail er
Just (Right y) -> do
ys <- lazyRead r
return (y :| ys)
-- | 'FromSourceIO' is intended to be implemented for types such as Conduit,
-- Pipe, etc. By implementing this class, all such streaming abstractions can
-- be used directly on the client side for talking to streaming endpoints.
class FromSourceIO chunk a | a -> chunk where
fromSourceIO :: SourceIO chunk -> a
lazyRead :: IO (Maybe (Either String a)) -> IO [a]
lazyRead r = go
instance MonadIO m => FromSourceIO a (SourceT m a) where
fromSourceIO = sourceFromSourceIO
sourceFromSourceIO :: forall m a. MonadIO m => SourceT IO a -> SourceT m a
sourceFromSourceIO src =
SourceT $ \k ->
k $ Effect $ liftIO $ unSourceT src (return . go)
where
go = unsafeInterleaveIO loop
go :: StepT IO a -> StepT m a
go Stop = Stop
go (Error err) = Error err
go (Skip s) = Skip (go s)
go (Effect ms) = Effect (liftIO (fmap go ms))
go (Yield x s) = Yield x (go s)
loop = do
e <- r
case e of
Nothing -> return []
Just (Left er) -> fail er
Just (Right y) -> do
ys <- go
return (y : ys)
-- This fires e.g. in Client.lhs
-- {-# OPTIONS_GHC -ddump-simpl -ddump-rule-firings -ddump-to-file #-}
{-# NOINLINE [2] sourceFromSourceIO #-}
{-# RULES "sourceFromSourceIO @IO" sourceFromSourceIO = id :: SourceT IO a -> SourceT IO a #-}
-- | The FramingRender class provides the logic for emitting a framing strategy. The strategy emits a header, followed by boundary-delimited data, and finally a termination character. For many strategies, some of these will just be empty bytestrings.
class FramingRender strategy a where
header :: Proxy strategy -> Proxy a -> ByteString
boundary :: Proxy strategy -> Proxy a -> BoundaryStrategy
trailer :: Proxy strategy -> Proxy a -> ByteString
-------------------------------------------------------------------------------
-- Framing
-------------------------------------------------------------------------------
-- | The bracketing strategy generates things to precede and follow the content, as with netstrings.
-- The intersperse strategy inserts seperators between things, as with newline framing.
-- Finally, the general strategy performs an arbitrary rewrite on the content, to allow escaping rules and such.
data BoundaryStrategy = BoundaryStrategyBracket (ByteString -> (ByteString,ByteString))
| BoundaryStrategyIntersperse ByteString
| BoundaryStrategyGeneral (ByteString -> ByteString)
-- | The 'FramingRender' class provides the logic for emitting a framing strategy.
-- The strategy transforms a @'SourceT' m a@ into @'SourceT' m 'LBS.ByteString'@,
-- therefore it can prepend, append and intercalate /framing/ structure
-- around chunks.
--
-- /Note:/ as the @'Monad' m@ is generic, this is pure transformation.
--
class FramingRender strategy where
framingRender :: Monad m => Proxy strategy -> (a -> LBS.ByteString) -> SourceT m a -> SourceT m LBS.ByteString
-- | A type of parser that can never fail, and has different parsing strategies (incremental, or EOF) depending if more input can be sent. The incremental parser should return `Nothing` if it would like to be sent a longer ByteString. If it returns a value, it also returns the remainder following that value.
data ByteStringParser a = ByteStringParser
{ parseIncremental :: ByteString -> Maybe (a, ByteString)
, parseEOF :: ByteString -> (a, ByteString)
}
-- | The 'FramingUnrender' class provides the logic for parsing a framing
-- strategy.
class FramingUnrender strategy where
framingUnrender :: Monad m => Proxy strategy -> (LBS.ByteString -> Either String a) -> SourceT m BS.ByteString -> SourceT m a
-- | The FramingUnrender class provides the logic for parsing a framing strategy. The outer @ByteStringParser@ strips the header from a stream of bytes, and yields a parser that can handle the remainder, stepwise. Each frame may be a ByteString, or a String indicating the error state for that frame. Such states are per-frame, so that protocols that can resume after errors are able to do so. Eventually this returns an empty ByteString to indicate termination.
class FramingUnrender strategy a where
unrenderFrames :: Proxy strategy -> Proxy a -> ByteStringParser (ByteStringParser (Either String ByteString))
-------------------------------------------------------------------------------
-- NoFraming
-------------------------------------------------------------------------------
-- | A framing strategy that does not do any framing at all, it just passes the input data
-- This will be used most of the time with binary data, such as files
-- | A framing strategy that does not do any framing at all, it just passes the
-- input data This will be used most of the time with binary data, such as
-- files
data NoFraming
instance FramingRender NoFraming a where
header _ _ = empty
boundary _ _ = BoundaryStrategyGeneral id
trailer _ _ = empty
instance FramingRender NoFraming where
framingRender _ = fmap
instance FramingUnrender NoFraming a where
unrenderFrames _ _ = ByteStringParser (Just . (go,)) (go,)
where go = ByteStringParser (Just . (, empty) . Right) ((, empty) . Right)
-- | As 'NoFraming' doesn't have frame separators, we take the chunks
-- as given and try to convert them one by one.
--
-- That works well when @a@ is a 'ByteString'.
instance FramingUnrender NoFraming where
framingUnrender _ f = mapStepT go
where
go Stop = Stop
go (Error err) = Error err
go (Skip s) = Skip (go s)
go (Effect ms) = Effect (fmap go ms)
go (Yield x s) = case f (LBS.fromStrict x) of
Right y -> Yield y (go s)
Left err -> Error err
-- | A simple framing strategy that has no header or termination, and inserts a newline character between each frame.
-- This assumes that it is used with a Content-Type that encodes without newlines (e.g. JSON).
-------------------------------------------------------------------------------
-- NewlineFraming
-------------------------------------------------------------------------------
-- | A simple framing strategy that has no header or termination, and inserts a
-- newline character between each frame. This assumes that it is used with a
-- Content-Type that encodes without newlines (e.g. JSON).
data NewlineFraming
instance FramingRender NewlineFraming a where
header _ _ = empty
boundary _ _ = BoundaryStrategyIntersperse "\n"
trailer _ _ = empty
instance FramingRender NewlineFraming where
framingRender _ f = mapStepT go0 where
go0 Stop = Stop
go0 (Error err) = Error err
go0 (Skip s) = Skip (go0 s)
go0 (Yield x s) = Yield (f x) (go s)
go0 (Effect ms) = Effect (fmap go0 ms)
instance FramingUnrender NewlineFraming a where
unrenderFrames _ _ = ByteStringParser (Just . (go,)) (go,)
where go = ByteStringParser
(\x -> case LB.break (== '\n') x of
(h,r) -> if not (LB.null r) then Just (Right h, LB.drop 1 r) else Nothing
)
(\x -> case LB.break (== '\n') x of
(h,r) -> (Right h, LB.drop 1 r)
)
-- | The netstring framing strategy as defined by djb: <http://cr.yp.to/proto/netstrings.txt>
go = fmap (\x -> "\n" <> f x)
instance FramingUnrender NewlineFraming where
framingUnrender _ f = transformWithAtto $ do
bs <- A.takeWhile (/= 10)
() <$ A.word8 10 <|> A.endOfInput
either fail pure (f (LBS.fromStrict bs))
-------------------------------------------------------------------------------
-- NetstringFraming
-------------------------------------------------------------------------------
-- | The netstring framing strategy as defined by djb:
-- <http://cr.yp.to/proto/netstrings.txt>
--
-- Any string of 8-bit bytes may be encoded as @[len]":"[string]","@. Here
-- @[string]@ is the string and @[len]@ is a nonempty sequence of ASCII digits
-- giving the length of @[string]@ in decimal. The ASCII digits are @<30>@ for
-- 0, @<31>@ for 1, and so on up through @<39>@ for 9. Extra zeros at the front
-- of @[len]@ are prohibited: @[len]@ begins with @<30>@ exactly when
-- @[string]@ is empty.
--
-- For example, the string @"hello world!"@ is encoded as
-- @<31 32 3a 68 65 6c 6c 6f 20 77 6f 72 6c 64 21 2c>@,
-- i.e., @"12:hello world!,"@.
-- The empty string is encoded as @"0:,"@.
--
data NetstringFraming
instance FramingRender NetstringFraming a where
header _ _ = empty
boundary _ _ = BoundaryStrategyBracket $ \b -> ((<> ":") . LB.pack . show . LB.length $ b, ",")
trailer _ _ = empty
instance FramingRender NetstringFraming where
framingRender _ f = fmap $ \x ->
let bs = f x
in LBS8.pack (show (LBS8.length bs)) <> ":" <> bs <> ","
instance FramingUnrender NetstringFraming a where
unrenderFrames _ _ = ByteStringParser (Just . (go,)) (go,)
where go = ByteStringParser
(\b -> let (i,r) = LB.break (==':') b
in case readMaybe (LB.unpack i) of
Just len -> if LB.length r > len
then Just . first Right . fmap (LB.drop 1) $ LB.splitAt len . LB.drop 1 $ r
else Nothing
Nothing -> Just (Left ("Bad netstring frame, couldn't parse value as integer value: " ++ LB.unpack i), LB.drop 1 . LB.dropWhile (/= ',') $ r))
(\b -> let (i,r) = LB.break (==':') b
in case readMaybe (LB.unpack i) of
Just len -> if LB.length r > len
then first Right . fmap (LB.drop 1) $ LB.splitAt len . LB.drop 1 $ r
else (Right $ LB.take len r, LB.empty)
Nothing -> (Left ("Bad netstring frame, couldn't parse value as integer value: " ++ LB.unpack i), LB.drop 1 . LB.dropWhile (/= ',') $ r))
instance FramingUnrender NetstringFraming where
framingUnrender _ f = transformWithAtto $ do
len <- A8.decimal
_ <- A8.char ':'
bs <- A.take len
_ <- A8.char ','
either fail pure (f (LBS.fromStrict bs))

6
servant/src/Servant/Links.hs
View file

@ -170,7 +170,7 @@ import Servant.API.RemoteHost
import Servant.API.ReqBody
(ReqBody')
import Servant.API.Stream
(Stream)
(Stream, StreamBody)
import Servant.API.Sub
(type (:>))
import Servant.API.TypeLevel
@ -482,6 +482,10 @@ instance HasLink sub => HasLink (ReqBody' mods ct a :> sub) where
type MkLink (ReqBody' mods ct a :> sub) r = MkLink sub r
toLink toA _ = toLink toA (Proxy :: Proxy sub)
instance HasLink sub => HasLink (StreamBody framing ct a :> sub) where
type MkLink (StreamBody framing ct a :> sub) r = MkLink sub r
toLink toA _ = toLink toA (Proxy :: Proxy sub)
instance (ToHttpApiData v, HasLink sub)
=> HasLink (Capture' mods sym v :> sub)
where

305
servant/src/Servant/Types/SourceT.hs Normal file
View file

@ -0,0 +1,305 @@
{-# LANGUAGE DeriveFunctor #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Servant.Types.SourceT where
import Control.Monad.Except
(ExceptT (..), runExceptT, throwError)
import Control.Monad.Morph
(MFunctor (..))
import Control.Monad.Trans.Class
(MonadTrans (..))
import qualified Data.Attoparsec.ByteString as A
import qualified Data.ByteString as BS
import Data.Functor.Classes
(Show1 (..), showsBinaryWith, showsPrec1, showsUnaryWith)
import Data.Functor.Identity
(Identity (..))
import Prelude hiding
(readFile)
import System.IO
(Handle, IOMode (..), withFile)
-- $setup
-- >>> :set -XOverloadedStrings
-- >>> import Control.Monad.Except (runExcept)
-- >>> import Data.Foldable (toList)
-- >>> import qualified Data.Attoparsec.ByteString.Char8 as A8
-- | This is CPSised ListT.
newtype SourceT m a = SourceT
{ unSourceT :: forall b. (StepT m a -> m b) -> m b
}
mapStepT :: (StepT m a -> StepT m b) -> SourceT m a -> SourceT m b
mapStepT f (SourceT m) = SourceT $ \k -> m (k . f)
{-# INLINE mapStepT #-}
-- | @ListT@ with additional constructors.
data StepT m a
= Stop
| Error String -- we can this argument configurable.
| Skip (StepT m a) -- Note: not sure about this constructor
| Yield a (StepT m a)
| Effect (m (StepT m a))
deriving Functor
-- | Create 'SourceT' from 'Step'.
--
-- /Note:/ often enough you want to use 'SourceT' directly.
fromStepT :: StepT m a -> SourceT m a
fromStepT s = SourceT ($ s)
-------------------------------------------------------------------------------
-- SourceT instances
-------------------------------------------------------------------------------
instance Functor m => Functor (SourceT m) where
fmap f = mapStepT (fmap f)
-- | >>> toList (source [1..10])
-- [1,2,3,4,5,6,7,8,9,10]
--
instance Identity ~ m => Foldable (SourceT m) where
foldr f z (SourceT m) = foldr f z (runIdentity (m Identity))
instance (Applicative m, Show1 m) => Show1 (SourceT m) where
liftShowsPrec sp sl d (SourceT m) = showsUnaryWith
(liftShowsPrec sp sl)
"fromStepT" d (Effect (m pure'))
where
pure' (Effect s) = s
pure' s = pure s
instance (Applicative m, Show1 m, Show a) => Show (SourceT m a) where
showsPrec = showsPrec1
-- | >>> hoist (Just . runIdentity) (source [1..3]) :: SourceT Maybe Int
-- fromStepT (Effect (Just (Yield 1 (Yield 2 (Yield 3 Stop)))))
instance MFunctor SourceT where
hoist f (SourceT m) = SourceT $ \k -> k $
Effect $ f $ fmap (hoist f) $ m return
-------------------------------------------------------------------------------
-- StepT instances
-------------------------------------------------------------------------------
instance Identity ~ m => Foldable (StepT m) where
foldr f z = go where
go Stop = z
go (Error _) = z
go (Skip s) = go s
go (Yield a s) = f a (go s)
go (Effect (Identity s)) = go s
instance (Applicative m, Show1 m) => Show1 (StepT m) where
liftShowsPrec sp sl = go where
go _ Stop = showString "Stop"
go d (Skip s) = showsUnaryWith
go
"Skip" d s
go d (Error err) = showsUnaryWith
showsPrec
"Error" d err
go d (Effect ms) = showsUnaryWith
(liftShowsPrec go goList)
"Effect" d ms
go d (Yield x s) = showsBinaryWith
sp go
"Yield" d x s
goList = liftShowList sp sl
instance (Applicative m, Show1 m, Show a) => Show (StepT m a) where
showsPrec = showsPrec1
-- | >>> lift [1,2,3] :: StepT [] Int
-- Effect [Yield 1 Stop,Yield 2 Stop,Yield 3 Stop]
--
instance MonadTrans StepT where
lift = Effect . fmap (`Yield` Stop)
instance MFunctor StepT where
hoist f = go where
go Stop = Stop
go (Error err) = Error err
go (Skip s) = Skip (go s)
go (Yield x s) = Yield x (go s)
go (Effect ms) = Effect (f (fmap go ms))
-------------------------------------------------------------------------------
-- Operations
-------------------------------------------------------------------------------
-- | Create pure 'SourceT'.
--
-- >>> source "foo" :: SourceT Identity Char
-- fromStepT (Effect (Identity (Yield 'f' (Yield 'o' (Yield 'o' Stop)))))
--
source :: [a] -> SourceT m a
source = fromStepT . foldr Yield Stop
-- | Get the answers.
--
-- >>> runSourceT (source "foo" :: SourceT Identity Char)
-- ExceptT (Identity (Right "foo"))
--
-- >>> runSourceT (source "foo" :: SourceT [] Char)
-- ExceptT [Right "foo"]
--
runSourceT :: Monad m => SourceT m a -> ExceptT String m [a]
runSourceT (SourceT m) = ExceptT (m (runExceptT . runStepT))
runStepT :: Monad m => StepT m a -> ExceptT String m [a]
runStepT Stop = return []
runStepT (Error err) = throwError err
runStepT (Skip s) = runStepT s
runStepT (Yield x s) = fmap (x :) (runStepT s)
runStepT (Effect ms) = lift ms >>= runStepT
{-
-- | >>> uncons (foldr Yield Stop "foo" :: StepT Identity Char)
-- Identity (Just ('f',Yield 'o' (Yield 'o' Stop)))
--
uncons :: Monad m => StepT m a -> m (Maybe (a, StepT m a))
uncons Stop = return Nothing
uncons (Skip s) = uncons s
uncons (Yield x s) = return (Just (x, s))
uncons (Effect ms) = ms >>= uncons
uncons (Error _) =
-}
-- | Filter values.
--
-- >>> toList $ mapMaybe (\x -> if odd x then Just x else Nothing) (source [0..10]) :: [Int]
-- [1,3,5,7,9]
--
-- >>> mapMaybe (\x -> if odd x then Just x else Nothing) (source [0..2]) :: SourceT Identity Int
-- fromStepT (Effect (Identity (Skip (Yield 1 (Skip Stop)))))
--
-- Illustrates why we need 'Skip'.
mapMaybe :: Functor m => (a -> Maybe b) -> SourceT m a -> SourceT m b
mapMaybe p (SourceT m) = SourceT $ \k -> m (k . mapMaybeStep p)
mapMaybeStep :: Functor m => (a -> Maybe b) -> StepT m a -> StepT m b
mapMaybeStep p = go where
go Stop = Stop
go (Error err) = Error err
go (Skip s) = Skip (go s)
go (Effect ms) = Effect (fmap go ms)
go (Yield x s) = case p x of
Nothing -> Skip (go s)
Just y -> Yield y (go s)
-- | Run action for each value in the 'SourceT'.
--
-- >>> foreach fail print (source "abc")
-- 'a'
-- 'b'
-- 'c'
--
foreach
:: Monad m
=> (String -> m ()) -- ^ error handler
-> (a -> m ())
-> SourceT m a
-> m ()
foreach f g src = unSourceT src (foreachStep f g)
-- | See 'foreach'.
foreachStep
:: Monad m
=> (String -> m ()) -- ^ error handler
-> (a -> m ())
-> StepT m a
-> m ()
foreachStep f g = go where
go Stop = return ()
go (Skip s) = go s
go (Yield x s) = g x >> go s
go (Error err) = f err
go (Effect ms) = ms >>= go
-------------------------------------------------------------------------------
-- Monadic
-------------------------------------------------------------------------------
fromAction :: Functor m => (a -> Bool) -> m a -> SourceT m a
fromAction stop action = SourceT ($ fromActionStep stop action)
{-# INLINE fromAction #-}
fromActionStep :: Functor m => (a -> Bool) -> m a -> StepT m a
fromActionStep stop action = loop where
loop = Effect $ fmap step action
step x
| stop x = Stop
| otherwise = Yield x loop
{-# INLINE fromActionStep #-}
-------------------------------------------------------------------------------
-- File
-------------------------------------------------------------------------------
-- | Read file.
--
-- >>> foreach fail BS.putStr (readFile "servant.cabal")
-- name: servant
-- ...
--
readFile :: FilePath -> SourceT IO BS.ByteString
readFile fp =
SourceT $ \k ->
withFile fp ReadMode $ \hdl ->
k (readHandle hdl)
where
readHandle :: Handle -> StepT IO BS.ByteString
readHandle hdl = fromActionStep BS.null (BS.hGet hdl 4096)
-------------------------------------------------------------------------------
-- Attoparsec
-------------------------------------------------------------------------------
-- | Transform using @attoparsec@ parser.
--
-- Note: @parser@ should not accept empty input!
--
-- >>> let parser = A.skipWhile A8.isSpace_w8 >> A.takeWhile1 A8.isDigit_w8
--
-- >>> runExcept $ runSourceT $ transformWithAtto parser (source ["1 2 3"])
-- Right ["1","2","3"]
--
-- >>> runExcept $ runSourceT $ transformWithAtto parser (source ["1", "2", "3"])
-- Right ["123"]
--
-- >>> runExcept $ runSourceT $ transformWithAtto parser (source ["1", "2 3", "4"])
-- Right ["12","34"]
--
-- >>> runExcept $ runSourceT $ transformWithAtto parser (source ["foobar"])
-- Left "Failed reading: takeWhile1"
--
transformWithAtto :: Monad m => A.Parser a -> SourceT m BS.ByteString -> SourceT m a
transformWithAtto parser = mapStepT (transformStepWithAtto parser)
transformStepWithAtto
:: forall a m. Monad m
=> A.Parser a -> StepT m BS.ByteString -> StepT m a
transformStepWithAtto parser = go (A.parse parser) where
p0 = A.parse parser
go :: (BS.ByteString -> A.Result a)
-> StepT m BS.ByteString -> StepT m a
go _ (Error err) = Error err
go p (Skip s) = Skip (go p s)
go p (Effect ms) = Effect (fmap (go p) ms)
go p Stop = case p mempty of
A.Fail _ _ err -> Error err
A.Done _ a -> Yield a Stop
A.Partial _ -> Stop
go p (Yield bs0 s) = loop p bs0 where
loop p' bs
| BS.null bs = Skip (go p' s)
| otherwise = case p' bs of
A.Fail _ _ err -> Error err
A.Done bs' a -> Yield a (loop p0 bs')
A.Partial p'' -> Skip (go p'' s)

84
servant/test/Servant/API/StreamSpec.hs Normal file
View file

@ -0,0 +1,84 @@
{-# LANGUAGE OverloadedStrings #-}
module Servant.API.StreamSpec where
import qualified Data.Aeson as Aeson
import qualified Data.ByteString as BS
import qualified Data.ByteString.Lazy as LBS
import Data.Functor.Identity
(Identity (..))
import Data.Proxy
(Proxy (..))
import Data.String
(fromString)
import Servant.API.Stream
import Servant.Types.SourceT
import Test.Hspec
import Test.QuickCheck
(Property, property, (===))
import Test.QuickCheck.Instances ()
spec :: Spec
spec = describe "Servant.API.Stream" $ do
describe "NoFraming" $ do
let framingUnrender' = framingUnrender (Proxy :: Proxy NoFraming) (Right . LBS.toStrict)
framingRender' = framingRender (Proxy :: Proxy NoFraming) LBS.fromStrict
it "framingUnrender" $
property $ \bss ->
runUnrenderFrames framingUnrender' bss === map Right (bss :: [BS.ByteString])
it "roundtrip" $
property $ roundtrip framingRender' framingUnrender'
describe "NewlineFraming" $ do
let tp = framingUnrender (Proxy :: Proxy NewlineFraming) (Right . LBS.toStrict)
it "framingUnrender examples" $ do
let expected n = map Right [fromString ("foo" ++ show (n :: Int)), "bar", "baz"]
runUnrenderFrames tp ["foo1\nbar\nbaz"] `shouldBe` expected 1
runUnrenderFrames tp ["foo2\n", "bar\n", "baz"] `shouldBe` expected 2
runUnrenderFrames tp ["foo3\nb", "ar\nbaz"] `shouldBe` expected 3
it "roundtrip" $ do
let framingUnrender' = framingUnrender (Proxy :: Proxy NewlineFraming) Aeson.eitherDecode
let framingRender' = framingRender (Proxy :: Proxy NewlineFraming) (Aeson.encode :: Int -> LBS.ByteString)
property $ roundtrip framingRender' framingUnrender'
-- it "fails if input doesn't contain newlines often" $
-- runUnrenderFrames tp ["foo", "bar"] `shouldSatisfy` any isLeft
describe "NetstringFraming" $ do
let tp = framingUnrender (Proxy :: Proxy NetstringFraming) (Right . LBS.toStrict)
it "framingUnrender examples" $ do
let expected n = map Right [fromString ("foo" ++ show (n :: Int)), "bar", "baz"]
runUnrenderFrames tp ["4:foo1,3:bar,3:baz,"] `shouldBe` expected 1
runUnrenderFrames tp ["4:foo2,", "3:bar,", "3:baz,"] `shouldBe` expected 2
runUnrenderFrames tp ["4:foo3,3:b", "ar,3:baz,"] `shouldBe` expected 3
it "roundtrip" $ do
let framingUnrender' = framingUnrender (Proxy :: Proxy NetstringFraming) Aeson.eitherDecode
let framingRender' = framingRender (Proxy :: Proxy NetstringFraming) (Aeson.encode :: Int -> LBS.ByteString)
property $ roundtrip framingRender' framingUnrender'
roundtrip
:: (Eq a, Show a)
=> (SourceT Identity a -> SourceT Identity LBS.ByteString)
-> (SourceT Identity BS.ByteString -> SourceT Identity a)
-> [a]
-> Property
roundtrip render unrender xs =
map Right xs === runUnrenderFrames (unrender . fmap LBS.toStrict . render) xs
runUnrenderFrames :: (SourceT Identity b -> SourceT Identity a) -> [b] -> [Either String a]
runUnrenderFrames f = go . Effect . flip unSourceT return . f . source where
go :: StepT Identity a -> [Either String a]
go Stop = []
go (Error err) = [Left err]
go (Skip s) = go s
go (Yield x s) = Right x : go s
go (Effect ms) = go (runIdentity ms)

4
stack.yaml
View file

@ -8,6 +8,10 @@ packages:
- servant-server/
- servant/
- servant-conduit
- servant-machines/
- servant-pipes/
# allow-newer: true # ignores all bounds, that's a sledgehammer
# - doc/tutorial/

268
streaming-benchmark.sh Normal file
View file

@ -0,0 +1,268 @@
#!/bin/sh
set -e
SIZE=10000000
SLOWURL="http://localhost:8000/slow/2"
FASTURL="http://localhost:8000/get/$SIZE"
PROXYURL="http://localhost:8000/proxy"
# we use binary, as there's some size
TESTFILE="$(cabal-plan list-bin servant-machines:test:example)"
TMPFILE="$(mktemp)"
CURLSTATS="time: %{time_total}, size: %{size_download}, download speed: %{speed_download}\n"
# cleanup
cleanup() {
if [ ! -z "$MACHINES_PID" ]; then
kill "$MACHINES_PID" || true
fi
if [ ! -z "$CONDUIT_PID" ]; then
kill "$CONDUIT_PID" || true
fi
if [ ! -z "$PIPES_PID" ]; then
kill "$PIPES_PID" || true
fi
if [ ! -z "$COOKBOOK_PID" ]; then
kill "$COOKBOOK_PID" || true
fi
rm -f "$TMPFILE"
}
trap cleanup EXIT
bench() {
echo "\e[36m=== $1 ===\e[0m"
}
# Server
#######################################################################
## Machines
bench "server machines"
$(cabal-plan list-bin servant-machines:test:example) server +RTS -sbench-machines-server-rts.txt &
MACHINES_PID=$!
echo "Starting servant-machines server. PID=$MACHINES_PID"
# Time to startup
sleep 1
# Run slow url to test & warm-up server
curl "$SLOWURL"
curl --silent --show-error "$FASTURL" --output /dev/null --write-out "$CURLSTATS" > bench-machines-server.txt
curl --silent --show-error "$PROXYURL" --request POST --data-binary @"$TESTFILE" --output "$TMPFILE" --write-out "$CURLSTATS" > bench-machines-server-proxy.txt
kill -INT $MACHINES_PID
unset MACHINES_PID
## Pipes
bench "server pipes"
$(cabal-plan list-bin servant-pipes:test:example) server +RTS -sbench-pipes-server-rts.txt &
PIPES_PID=$!
echo "Starting servant-pipes server. PID=$PIPES_PID"
# Time to startup
sleep 1
# Run slow url to test & warm-up server
curl "$SLOWURL"
curl --silent --show-error "$FASTURL" --output /dev/null --write-out "$CURLSTATS" > bench-pipes-server.txt
curl --silent --show-error "$PROXYURL" --request POST --data-binary @"$TESTFILE" --output "$TMPFILE" --write-out "$CURLSTATS" > bench-pipes-server-proxy.txt
kill -INT $PIPES_PID
unset PIPES_PID
## Cookbook
bench "server cookbook"
$(cabal-plan list-bin cookbook-basic-streaming) server +RTS -sbench-cookbook-server-rts.txt &
COOKBOOK_PID=$!
echo "Starting servant-cookbook server. PID=$COOKBOOK_PID"
# Time to startup
sleep 1
# Run slow url to test & warm-up server
curl "$SLOWURL"
curl --silent --show-error "$FASTURL" --output /dev/null --write-out "$CURLSTATS" > bench-cookbook-server.txt
curl --silent --show-error "$PROXYURL" --request POST --data-binary @"$TESTFILE" --output "$TMPFILE" --write-out "$CURLSTATS" > bench-cookbook-server-proxy.txt
kill -INT $COOKBOOK_PID
unset COOKBOOK_PID
## Conduit
bench "server conduit"
$(cabal-plan list-bin servant-conduit:test:example) server +RTS -sbench-conduit-server-rts.txt &
CONDUIT_PID=$!
echo "Starting servant-conduit server. PID=$CONDUIT_PID"
# Time to startup
sleep 1
# Run slow url to test & warm-up server
curl "$SLOWURL"
curl --silent --show-error "$FASTURL" --output /dev/null --write-out "$CURLSTATS" > bench-conduit-server.txt
curl --silent --show-error "$PROXYURL" --request POST --data-binary @"$TESTFILE" --output "$TMPFILE" --write-out "$CURLSTATS" > bench-conduit-server-proxy.txt
# kill -INT $CONDUIT_PID
# unset CONDUIT_PID
# Client
#######################################################################
# Uses conduit as server
## Machines
bench "client machines"
# Test run
$(cabal-plan list-bin servant-machines:test:example) client 10
# Real run
/usr/bin/time --verbose --output bench-machines-client-time.txt \
"$(cabal-plan list-bin servant-machines:test:example)" client "$SIZE" +RTS -sbench-machines-client-rts.txt
## Pipes
bench "server pipes"
# Test run
$(cabal-plan list-bin servant-pipes:test:example) client 10
# Real run
/usr/bin/time --verbose --output bench-pipes-client-time.txt \
"$(cabal-plan list-bin servant-pipes:test:example)" client "$SIZE" +RTS -sbench-pipes-client-rts.txt
## Conduit
bench "client conduit"
# Test run
$(cabal-plan list-bin servant-conduit:test:example) client 10
# Real run
/usr/bin/time --verbose --output bench-conduit-client-time.txt \
"$(cabal-plan list-bin servant-conduit:test:example)" client "$SIZE" +RTS -sbench-conduit-client-rts.txt
## Cookbook
bench "server cookbook"
# Test run
$(cabal-plan list-bin cookbook-basic-streaming) client 10
# Real run
/usr/bin/time --verbose --output bench-cookbook-client-time.txt \
"$(cabal-plan list-bin cookbook-basic-streaming)" client "$SIZE" +RTS -sbench-cookbook-client-rts.txt
## Kill server
kill -INT $CONDUIT_PID
unset CONDUIT_PID
sleep 1
# Exit
#######################################################################
header() {
{ echo "$1 $2";
echo ""
} >> bench.md
}
report() {
echo "\`\`\`" >> bench.md
cat "$1" >> bench.md
echo "\`\`\`" >> bench.md
echo "" >> bench.md
}
report2() {
echo "\`\`\`" >> bench.md
cat "$1" | sed 's/^\s*//' >> bench.md
echo "\`\`\`" >> bench.md
echo "" >> bench.md
}
note() {
echo "$1" >> bench.md
echo "" >> bench.md
}
rm -f bench.md
header "#" "Streaming test benchmark"
note "size parameter: $SIZE"
header "##" Server
note "- /fast/$SIZE\n- /proxy"
header "###" machines
report bench-machines-server.txt
report bench-machines-server-proxy.txt
report bench-machines-server-rts.txt
header "###" pipes
report bench-pipes-server.txt
report bench-pipes-server-proxy.txt
report bench-pipes-server-rts.txt
header "###" conduit
note "Conduit server is also used for client tests below"
report bench-conduit-server.txt
report bench-conduit-server-proxy.txt
report bench-conduit-server-rts.txt
header "###" cookbook
report bench-cookbook-server.txt
report bench-cookbook-server-proxy.txt
report bench-cookbook-server-rts.txt
header "##" Client
header "###" machines
report2 bench-machines-client-time.txt
report bench-machines-client-rts.txt
header "###" pipes
report2 bench-pipes-client-time.txt
report bench-pipes-client-rts.txt
header "###" conduit
report2 bench-conduit-client-time.txt
report bench-conduit-client-rts.txt
header "###" cookbook
report2 bench-cookbook-client-time.txt
report bench-cookbook-client-rts.txt
# Cleanup filepaths
sed -E -i 's/\/[^ ]*machines[^ ]*\/example/...machines:example/' bench.md
sed -E -i 's/\/[^ ]*conduit[^ ]*\/example/...conduit:example/' bench.md
sed -E -i 's/\/[^ ]*pipes[^ ]*\/example/...pipes:example/' bench.md
sed -E -i 's/\/[^ ]*\/cookbook-basic-streaming/...cookbook-basic-streaming/' bench.md
sleep 3