e75a3cc · typeable/servant

git.iko.soy/typeable/servant

on master help sign in

streaming client actually streams

e75a3ccGershom9 years ago

servant-client-core/src/Servant/Client/Core.hs+1

compare with master

unifiedsplit

@@ -43,6 +43,7 @@ module Servant.Client.Core

43 , Response(..)

44 , RunClient(..)

45 , module Servant.Client.Core.Internal.BaseUrl

46+ , StreamingResponse(..)

47 -- * Writing HasClient instances

48 -- * Writing HasClient instances

48 -- | These functions need not be re-exported by backend libraries.

49 -- | These functions need not be re-exported by backend libraries.

servant-client-core/src/Servant/Client/Core/Internal/HasClient.hs+44−29

@@ -4,6 +4,7 @@

4{-# LANGUAGE FlexibleInstances #-}

5{-# LANGUAGE InstanceSigs #-}

6{-# LANGUAGE MultiParamTypeClasses #-}

7+{-# LANGUAGE RankNTypes #-}

7{-# LANGUAGE OverloadedStrings #-}

8{-# LANGUAGE OverloadedStrings #-}

8{-# LANGUAGE PolyKinds #-}

9{-# LANGUAGE PolyKinds #-}

9{-# LANGUAGE ScopedTypeVariables #-}

10{-# LANGUAGE ScopedTypeVariables #-}

@@ -17,9 +18,12 @@ module Servant.Client.Core.Internal.HasClient where

17import Prelude ()

18import Prelude ()

18import Prelude.Compat

19import Prelude.Compat

21+import Control.Concurrent (newMVar, modifyMVar)

22+import Control.Monad (when)

20import Data.Foldable (toList)

23import Data.Foldable (toList)

21import qualified Data.ByteString.Lazy as BL

24import qualified Data.ByteString.Lazy as BL

22import Data.List (foldl')

25import Data.List (foldl')

26+import Data.Monoid ((<>))

23import Data.Proxy (Proxy (Proxy))

27import Data.Proxy (Proxy (Proxy))

24import Data.Sequence (fromList)

28import Data.Sequence (fromList)

25import Data.String (fromString)

29import Data.String (fromString)

@@ -31,6 +35,7 @@ import Servant.API ((:<|>) ((:<|>)), (:>),

31 BasicAuthData,

35 BasicAuthData,

32 BuildHeadersTo (..),

36 BuildHeadersTo (..),

33 BuildFromStream (..),

37 BuildFromStream (..),

38+ ByteStringParser (..),

34 Capture, CaptureAll,

39 Capture, CaptureAll,

35 Description, EmptyAPI,

40 Description, EmptyAPI,

36 FramingUnrender (..),

41 FramingUnrender (..),

@@ -248,44 +253,54 @@ instance OVERLAPPING_

248 , getHeadersHList = buildHeadersTo . toList $ responseHeaders response

253 , getHeadersHList = buildHeadersTo . toList $ responseHeaders response

249 }

254 }

250

255

256+data ResultStream a = ResultStream ((forall b. (IO (Maybe (Either String a)) -> IO b) -> IO b))

257+

251instance OVERLAPPABLE_

258instance OVERLAPPABLE_

252 ( RunClient m, MimeUnrender ct a, ReflectMethod method,

259 ( RunClient m, MimeUnrender ct a, ReflectMethod method,

253 FramingUnrender framing a, BuildFromStream a (f a)

260 FramingUnrender framing a, BuildFromStream a (f a)

254 ) => HasClient m (Stream method framing ct (f a)) where

261 ) => HasClient m (Stream method framing ct (f a)) where

255− type Client m (Stream method framing ct (f a)) = m (f a)

256− clientWithRoute _pm Proxy req = do

257− response <- runRequest req

258− { requestAccept = fromList [contentType (Proxy :: Proxy ct)]

259− , requestMethod = reflectMethod (Proxy :: Proxy method)

260− }

261− return $ decodeFramed (Proxy :: Proxy framing) (Proxy :: Proxy ct) (Proxy :: Proxy a) response

262

263−instance OVERLAPPING_

264− ( RunClient m, BuildHeadersTo ls, MimeUnrender ct a, ReflectMethod method,

265− FramingUnrender framing a, BuildFromStream a (f a)

266− ) => HasClient m (Stream method framing ct (Headers ls (f a))) where

267− type Client m (Stream method framing ct (Headers ls (f a))) = m (Headers ls (f a))

263+ type Client m (Stream method framing ct (f a)) = m (ResultStream a)

264+

268 clientWithRoute _pm Proxy req = do

265 clientWithRoute _pm Proxy req = do

269− response <- runRequest req

266+ sresp <- streamingRequest req

270 { requestAccept = fromList [contentType (Proxy :: Proxy ct)]

267 { requestAccept = fromList [contentType (Proxy :: Proxy ct)]

271 , requestMethod = reflectMethod (Proxy :: Proxy method)

268 , requestMethod = reflectMethod (Proxy :: Proxy method)

272 }

269 }

273− return Headers { getResponse = decodeFramed (Proxy :: Proxy framing) (Proxy :: Proxy ct) (Proxy :: Proxy a) response

270+ return $ ResultStream $ \k ->

274− , getHeadersHList = buildHeadersTo . toList $ responseHeaders response

271+ runStreamingResponse sresp $ \(status,_headers,_httpversion,reader) -> do

275− }

272+ when (H.statusCode status /= 200) $ error "bad status" --fixme

276−

273+ let unrender = unrenderFrames (Proxy :: Proxy framing) (Proxy :: Proxy a)

277−decodeFramed :: forall ctype strategy a b.

274+ loop bs = do

278− (MimeUnrender ctype a, FramingUnrender strategy a, BuildFromStream a b) =>

275+ res <- BL.fromStrict <$> reader

279− Proxy strategy -> Proxy ctype -> Proxy a -> Response -> b

276+ if BL.null res

280−decodeFramed framingproxy ctypeproxy typeproxy response =

277+ then return $ parseEOF unrender res

281− let (body, uncons) = unrenderFrames framingproxy typeproxy (responseBody response)

278+ else let sofar = (bs <> res)

282− loop b | BL.null b = []

279+ in case parseIncremental unrender sofar of

283− | otherwise = case uncons b of

280+ Just x -> return x

281+ Nothing -> loop sofar

282+ (frameParser, remainder) <- loop BL.empty

283+ state <- newMVar remainder

284+ let frameLoop bs = do

285+ res <- BL.fromStrict <$> reader

286+ let addIsEmptyInfo (a, r) = (r, (a, BL.null r && BL.null res))

287+ if BL.null res

288+ then return . addIsEmptyInfo $ parseEOF frameParser res

289+ else let sofar = (bs <> res)

290+ in case parseIncremental frameParser res of

291+ Just x -> return $ addIsEmptyInfo x

292+ Nothing -> frameLoop sofar

293+

294+ go = processResult <$> modifyMVar state frameLoop

295+ processResult (Right bs,isDone) =

296+ if BL.null bs && isDone

297+ then Nothing

284− (Right x, r) -> case mimeUnrender ctypeproxy x :: Either String a of

298+ else Just $ case mimeUnrender (Proxy :: Proxy ct) bs :: Either String a of

285− Left err -> Left err : loop r

299+ Left err -> Left err

286− Right x' -> Right x' : loop r

300+ Right x -> Right x

287− (Left err, r) -> Left err : loop r

301+ processResult (Left err, _) = Just (Left err)

288− in buildFromStream $ loop body

302+ k go

303+

289

304

290-- | If you use a 'Header' in one of your endpoints in your API,

305-- | If you use a 'Header' in one of your endpoints in your API,

291-- the corresponding querying function will automatically take

306-- the corresponding querying function will automatically take

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

@@ -4,6 +4,7 @@

4{-# LANGUAGE DeriveGeneric #-}

5{-# LANGUAGE GeneralizedNewtypeDeriving #-}

6{-# LANGUAGE MultiParamTypeClasses #-}

7+{-# LANGUAGE RankNTypes #-}

7{-# LANGUAGE OverloadedStrings #-}

8{-# LANGUAGE OverloadedStrings #-}

8{-# LANGUAGE ScopedTypeVariables #-}

9{-# LANGUAGE ScopedTypeVariables #-}

9{-# LANGUAGE TypeFamilies #-}

10{-# LANGUAGE TypeFamilies #-}

@@ -15,6 +16,7 @@ import Prelude.Compat

16import Control.Monad.Catch (Exception)

17import Control.Monad.Catch (Exception)

17import qualified Data.ByteString.Builder as Builder

18import qualified Data.ByteString.Builder as Builder

19+import qualified Data.ByteString as BS

18import qualified Data.ByteString.Lazy as LBS

20import qualified Data.ByteString.Lazy as LBS

19import Data.Semigroup ((<>))

21import Data.Semigroup ((<>))

20import qualified Data.Sequence as Seq

22import qualified Data.Sequence as Seq

@@ -70,6 +72,8 @@ data Response = Response

70 , responseHttpVersion :: HttpVersion

72 , responseHttpVersion :: HttpVersion

71 } deriving (Eq, Show, Generic, Typeable)

73 } deriving (Eq, Show, Generic, Typeable)

75+data StreamingResponse = StreamingResponse { runStreamingResponse :: forall a. ((Status, Seq.Seq Header, HttpVersion, IO BS.ByteString) -> IO a) -> IO a }

76+

73-- A GET request to the top-level path

77-- A GET request to the top-level path

74defaultRequest :: Request

78defaultRequest :: Request

75defaultRequest = Request

79defaultRequest = Request

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

@@ -19,11 +19,13 @@ import Servant.API (MimeUnrender,

19 contentTypes,

20 mimeUnrender)

21import Servant.Client.Core.Internal.Request (Request, Response (..),

22+ StreamingResponse (..),

22 ServantError (..))

23 ServantError (..))

24class (Monad m) => RunClient m where

25class (Monad m) => RunClient m where

25 -- | How to make a request.

26 -- | How to make a request.

26 runRequest :: Request -> m Response

27 runRequest :: Request -> m Response

28+ streamingRequest :: Request -> m StreamingResponse

27 throwServantError :: ServantError -> m a

29 throwServantError :: ServantError -> m a

28 catchServantError :: m a -> (ServantError -> m a) -> m a

30 catchServantError :: m a -> (ServantError -> m a) -> m a

servant-client/src/Servant/Client/Internal/HttpClient.hs+12

@@ -88,6 +88,7 @@ instance Alt ClientM where

89instance RunClient ClientM where

90 runRequest = performRequest

91+ streamingRequest = performStreamingRequest

91 throwServantError = throwError

92 throwServantError = throwError

92 catchServantError = catchError

93 catchServantError = catchError

@@ -115,6 +116,17 @@ performRequest req = do

115 throwError $ FailureResponse ourResponse

116 throwError $ FailureResponse ourResponse

116 return ourResponse

117 return ourResponse

117

118

119+performStreamingRequest :: Request -> ClientM StreamingResponse

120+performStreamingRequest req = do

121+ m <- asks manager

122+ burl <- asks baseUrl

123+ let request = requestToClientRequest burl req

124+

125+ return $ StreamingResponse $

126+ \k -> Client.withResponse request m $

127+ \r ->

128+ k (Client.responseStatus r, fromList $ Client.responseHeaders r, Client.responseVersion r, Client.responseBody r)

129+

118clientResponseToReponse :: Client.Response BSL.ByteString -> Response

130clientResponseToReponse :: Client.Response BSL.ByteString -> Response

119clientResponseToReponse r = Response

131clientResponseToReponse r = Response

120 { responseStatusCode = Client.responseStatus r

132 { responseStatusCode = Client.responseStatus r

servant/src/Servant/API.hs+1

@@ -85,6 +85,7 @@ import Servant.API.QueryParam (QueryFlag, QueryParam,

85import Servant.API.Raw (Raw)

86import Servant.API.Stream (Stream, StreamGenerator (..),

87 ToStreamGenerator (..), BuildFromStream (..),

88+ ByteStringParser (..),

88 FramingRender (..), BoundaryStrategy (..),

89 FramingRender (..), BoundaryStrategy (..),

89 FramingUnrender (..),

90 FramingUnrender (..),

90 NewlineFraming)

91 NewlineFraming)

servant/src/Servant/API/Stream.hs+19−6

@@ -11,13 +11,11 @@

12module Servant.API.Stream where

14−import Control.Arrow ((***), first)

15import Data.ByteString.Lazy (ByteString, empty)

14import Data.ByteString.Lazy (ByteString, empty)

16import qualified Data.ByteString.Lazy.Char8 as LB

15import qualified Data.ByteString.Lazy.Char8 as LB

17import Data.Proxy (Proxy)

16import Data.Proxy (Proxy)

18import Data.Typeable (Typeable)

17import Data.Typeable (Typeable)

19import GHC.Generics (Generic)

18import GHC.Generics (Generic)

20−import Text.Read (readMaybe)

21import Network.HTTP.Types.Method (StdMethod (..))

19import Network.HTTP.Types.Method (StdMethod (..))

23-- | A Stream endpoint for a given method emits a stream of encoded values at a given Content-Type, delimited by a framing strategy. Steam endpoints always return response code 200 on success. Type synonyms are provided for standard methods.

21-- | A Stream endpoint for a given method emits a stream of encoded values at a given Content-Type, delimited by a framing strategy. Steam endpoints always return response code 200 on success. Type synonyms are provided for standard methods.

@@ -57,10 +55,16 @@ data BoundaryStrategy = BoundaryStrategyBracket (ByteString -> (ByteString,ByteS

57 | BoundaryStrategyIntersperse ByteString

55 | BoundaryStrategyIntersperse ByteString

58 | BoundaryStrategyGeneral (ByteString -> ByteString)

56 | BoundaryStrategyGeneral (ByteString -> ByteString)

60−-- | The FramingUnrender class provides the logic for parsing a framing strategy. Given a ByteString, it strips the header, and returns a tuple of the remainder along with a step function that can progressively "uncons" elements from this remainder. The error state is presented per-frame so that protocols that can resume after errors are able to do so. Termination of the unrendering is indicated by return of an empty reminder.

58+-- | 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.

59+data ByteStringParser a = ByteStringParser {

60+ parseIncremental :: ByteString -> Maybe (a, ByteString),

61+ parseEOF :: ByteString -> (a, ByteString)

62+}

64+-- | 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.

62class FramingUnrender strategy a where

65class FramingUnrender strategy a where

63− unrenderFrames :: Proxy strategy -> Proxy a -> ByteString -> (ByteString, ByteString -> (Either String ByteString, ByteString))

66+ unrenderFrames :: Proxy strategy -> Proxy a -> ByteStringParser (ByteStringParser (Either String ByteString))

67+

65-- | A simple framing strategy that has no header or termination, and inserts a newline character between each frame.

69-- | A simple framing strategy that has no header or termination, and inserts a newline character between each frame.

66-- This assumes that it is used with a Content-Type that encodes without newlines (e.g. JSON).

70-- This assumes that it is used with a Content-Type that encodes without newlines (e.g. JSON).

@@ -72,8 +76,14 @@ instance FramingRender NewlineFraming a where

72 trailer _ _ = empty

76 trailer _ _ = empty

74instance FramingUnrender NewlineFraming a where

78instance FramingUnrender NewlineFraming a where

75− unrenderFrames _ _ = (, (Right *** LB.drop 1) . LB.break (== '\n'))

79+ unrenderFrames _ _ = ByteStringParser (Just . (go,)) (go,)

76−

80+ where go = ByteStringParser

81+ (\x -> case LB.break (== '\n') x of

82+ (h,r) -> if not (LB.null r) then Just (Right h, LB.drop 1 r) else Nothing

83+ )

84+ (\x -> case LB.break (== '\n') x of

85+ (h,r) -> (Right h, LB.drop 1 r)

86+ )

77-- | The netstring framing strategy as defined by djb: <http://cr.yp.to/proto/netstrings.txt>

87-- | The netstring framing strategy as defined by djb: <http://cr.yp.to/proto/netstrings.txt>

78data NetstringFraming

88data NetstringFraming

@@ -82,9 +92,12 @@ instance FramingRender NetstringFraming a where

82 boundary _ _ = BoundaryStrategyBracket $ \b -> (LB.pack . show . LB.length $ b, "")

92 boundary _ _ = BoundaryStrategyBracket $ \b -> (LB.pack . show . LB.length $ b, "")

83 trailer _ _ = empty

93 trailer _ _ = empty

95+{-

96+

85instance FramingUnrender NetstringFraming a where

97instance FramingUnrender NetstringFraming a where

86 unrenderFrames _ _ = (, \b -> let (i,r) = LB.break (==':') b

98 unrenderFrames _ _ = (, \b -> let (i,r) = LB.break (==':') b

87 in case readMaybe (LB.unpack i) of

99 in case readMaybe (LB.unpack i) of

88 Just len -> first Right $ LB.splitAt len . LB.drop 1 $ r

100 Just len -> first Right $ LB.splitAt len . LB.drop 1 $ r

89 Nothing -> (Left ("Bad netstring frame, couldn't parse value as integer value: " ++ LB.unpack i), LB.drop 1 . LB.dropWhile (/= ',') $ r)

101 Nothing -> (Left ("Bad netstring frame, couldn't parse value as integer value: " ++ LB.unpack i), LB.drop 1 . LB.dropWhile (/= ',') $ r)

90 )

102 )

103+-}