http2-client 0.8.0.2 → 0.10.0.2
raw patch · 9 files changed
Files
- ChangeLog.md +12/−0
- http2-client.cabal +16/−11
- src/Network/HTTP2/Client.hs +1046/−957
- src/Network/HTTP2/Client/Channels.hs +36/−28
- src/Network/HTTP2/Client/Dispatch.hs +161/−137
- src/Network/HTTP2/Client/Exceptions.hs +24/−0
- src/Network/HTTP2/Client/FrameConnection.hs +30/−23
- src/Network/HTTP2/Client/Helpers.hs +31/−15
- src/Network/HTTP2/Client/RawConnection.hs +57/−24
ChangeLog.md view
@@ -2,6 +2,18 @@ ## Unreleased changes +- Change `waitStream` so that it automatically credits the connection. This+ is a breaking change due to function signature change.++## v0.10.0.0++- Export frameHttp2RawConnection to convert an RawHttp2Connection into an Http2FrameConnection.+- Add newRawHttp2ConnectionUnix to create a RawHttp2Connection from a unix domain socket.++## v0.9.0.0++- introduce ClientIO as an error-carrying IO-monad for performing http2-calls+ ## v0.8.0.2 - first Changelog!
http2-client.cabal view
@@ -1,5 +1,5 @@ name: http2-client-version: 0.8.0.2+version: 0.10.0.2 synopsis: A native HTTP2 client library. description: Please read the README.md at the homepage. homepage: https://github.com/lucasdicioccio/http2-client@@ -16,21 +16,26 @@ library hs-source-dirs: src exposed-modules: Network.HTTP2.Client+ , Network.HTTP2.Client.Exceptions , Network.HTTP2.Client.FrameConnection , Network.HTTP2.Client.Helpers , Network.HTTP2.Client.RawConnection other-modules: Network.HTTP2.Client.Channels , Network.HTTP2.Client.Dispatch- build-depends: base >= 4.7 && < 5- , async >= 2.1 && < 3- , bytestring >= 0.10 && < 1- , containers >= 0.5 && < 1- , deepseq >= 1.4 && < 2- , http2 >= 1.6 && < 2- , network >= 2.6 && < 3- , stm >= 2.4 && < 3- , time >= 1.8 && < 2- , tls >= 1.4 && < 2+ build-depends: base >= 4.7 && < 4.20+ , async >= 2.1 && < 2.3+ , bytestring >= 0.11 && < 0.13+ , containers >= 0.5 && < 0.8+ , deepseq >= 1.4 && < 1.6+ , http2 >= 4.1 && < 6+ , lifted-async >= 0.10 && < 0.11+ , lifted-base >= 0.2 && < 0.3+ , mtl >= 2.2 && < 2.4+ , network >= 2.6 && < 3.2+ , stm >= 2.4 && < 2.8+ , time >= 1.8 && < 1.15+ , tls >= 1.8.0 && < 2.0.3+ , transformers-base >= 0.4 && < 0.5 default-language: Haskell2010 -- Commented-out to avoid distribution
src/Network/HTTP2/Client.hs view
@@ -1,957 +1,1046 @@-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE RankNTypes #-}-{-# LANGUAGE OverloadedStrings #-}---- | This module defines a set of low-level primitives for starting an HTTP2--- session and interacting with a server.------ For higher-level primitives, please refer to Network.HTTP2.Client.Helpers .-module Network.HTTP2.Client (- -- * Basics- runHttp2Client- , newHttp2Client- , withHttp2Stream- , headers- , trailers- , sendData- -- * Starting clients- , Http2Client(..)- , PushPromiseHandler- -- * Starting streams- , StreamDefinition(..)- , StreamStarter- , TooMuchConcurrency(..)- , StreamThread- , Http2Stream(..)- -- * Flow control- , IncomingFlowControl(..)- , OutgoingFlowControl(..)- -- * Exceptions- , linkAsyncs- , RemoteSentGoAwayFrame(..)- , GoAwayHandler- , defaultGoAwayHandler- -- * Misc.- , FallBackFrameHandler- , ignoreFallbackHandler- , FlagSetter- , Http2ClientAsyncs(..)- , _gtfo- -- * Convenience re-exports- , StreamEvent(..)- , module Network.HTTP2.Client.FrameConnection- , module Network.Socket- , module Network.TLS- ) where--import Control.Concurrent.Async (Async, async, race, withAsync, link)-import Control.Exception (bracket, throwIO, SomeException, catch)-import Control.Concurrent.MVar (newEmptyMVar, newMVar, putMVar, takeMVar, tryPutMVar)-import Control.Concurrent (threadDelay)-import Control.Monad (forever, void, when, forM_)-import Data.ByteString (ByteString)-import qualified Data.ByteString as ByteString-import Data.IORef (newIORef, atomicModifyIORef', readIORef)-import Data.Maybe (fromMaybe)-import Network.HPACK as HPACK-import Network.HTTP2 as HTTP2-import Network.Socket (HostName, PortNumber)-import Network.TLS (ClientParams)--import Network.HTTP2.Client.Channels-import Network.HTTP2.Client.Dispatch-import Network.HTTP2.Client.FrameConnection---- | Offers credit-based flow-control.------ Any mutable changes are atomic and hence work as intended in a multithreaded--- setup.------ The design of the flow-control mechanism is subject to changes. One--- important thing to keep in mind with current implementation is that both the--- connection and streams are credited with '_addCredit' as soon as DATA frames--- arrive, hence no-need to account for the DATA frames (but you can account--- for delay-bandwidth product for instance).-data IncomingFlowControl = IncomingFlowControl {- _addCredit :: WindowSize -> IO ()- -- ^ Add credit (using a hidden mutable reference underneath). This function- -- only does accounting, the IO only does mutable changes. See '_updateWindow'.- , _consumeCredit :: WindowSize -> IO Int- -- ^ Consumes some credit and returns the credit left.- , _updateWindow :: IO Bool- -- ^ Sends a WINDOW_UPDATE frame crediting it with the whole amount credited- -- since the last _updateWindow call. The boolean tells whether an update was- -- actually sent or not. A reason for not sending an update is if there is no- -- credit in the flow-control system.- }---- | Receives credit-based flow-control or block.------ There is no way to observe the total amount of credit and receive/withdraw--- are atomic hence this object is thread-safe. However we plan to propose an--- STM-based API to allow withdrawing atomically from both the connection and a--- per-stream 'OutgoingFlowControl' objects at a same time. Without such--- atomicity one must ensure consumers do not exhaust the connection credit--- before taking the per-stream credit (else they might prevent others sending--- data without taking any).------ Longer term we plan to hide outgoing-flow-control increment/decrement--- altogether because exception between withdrawing credit and sending DATA--- could mean lost credit (and hence hanging streams).-data OutgoingFlowControl = OutgoingFlowControl {- _receiveCredit :: WindowSize -> IO ()- -- ^ Add credit (using a hidden mutable reference underneath).- , _withdrawCredit :: WindowSize -> IO WindowSize- -- ^ Wait until we can take credit from stash. The returned value correspond- -- to the amount that could be withdrawn, which is min(current, wanted). A- -- caller should withdraw credit to send DATA chunks and put back any unused- -- credit with _receiveCredit.- }---- | Defines a client stream.------ Please red the doc for this record fields and then see 'StreamStarter'.-data StreamDefinition a = StreamDefinition {- _initStream :: IO StreamThread- -- ^ Function to initialize a new client stream. This function runs in a- -- exclusive-access section of the code and may prevent other threads to- -- initialize new streams. Hence, you should ensure this IO does not wait for- -- long periods of time.- , _handleStream :: IncomingFlowControl -> OutgoingFlowControl -> IO a- -- ^ Function to operate with the stream. IncomingFlowControl currently is- -- credited on your behalf as soon as a DATA frame arrives (and before you- -- handle it with '_waitData'). However we do not send WINDOW_UPDATE with- -- '_updateWindow'. This design may change in the future to give more leeway- -- to library users.- }---- | Type alias for callback-based functions starting new streams.------ The callback a user must provide takes an 'Http2Stream' and returns a--- 'StreamDefinition'. This construction may seem wrong because a 'StreamDefinition'--- contains an initialization and a handler functions. The explanation for this--- twistedness is as follows: in HTTP2 stream-ids must be monotonically--- increasing, if we want to support multi-threaded clients we need to--- serialize access to a critical region of the code when clients send--- HEADERS+CONTINUATIONs frames.------ Passing the 'Http2Stream' object as part of the callback avoids leaking the--- implementation of the critical region, meanwhile, the 'StreamDefinition'--- delimits this critical region.-type StreamStarter a =- (Http2Stream -> StreamDefinition a) -> IO (Either TooMuchConcurrency a)---- | Whether or not the client library believes the server will reject the new--- stream. The Int content corresponds to the number of streams that should end--- before accepting more streams. A reason this number can be more than zero is--- that servers can change (and hence reduce) the advertised number of allowed--- 'maxConcurrentStreams' at any time.-newtype TooMuchConcurrency = TooMuchConcurrency { _getStreamRoomNeeded :: Int }- deriving Show---- | Record holding functions one can call while in an HTTP2 client session.-data Http2Client = Http2Client {- _ping :: ByteString -> IO (IO (FrameHeader, FramePayload))- -- ^ Send a PING, the payload size must be exactly eight bytes.- -- Returns an IO to wait for a ping reply. No timeout is provided. Only the- -- first call to this IO will return if a reply is received. Hence we- -- recommend wrapping this IO in an Async (e.g., with @race (threadDelay- -- timeout)@.)- , _settings :: SettingsList -> IO (IO (FrameHeader, FramePayload))- -- ^ Sends a SETTINGS. Returns an IO to wait for a settings reply. No timeout- -- is provided. Only the first call to this IO will return if a reply is- -- received. Hence we recommend wrapping this IO in an Async (e.g., with- -- @race (threadDelay timeout)@.)- , _goaway :: ErrorCodeId -> ByteString -> IO ()- -- ^ Sends a GOAWAY.- , _startStream :: forall a. StreamStarter a- -- ^ Spawns new streams. See 'StreamStarter'.- , _incomingFlowControl :: IncomingFlowControl- -- ^ Simple getter for the 'IncomingFlowControl' for the whole client- -- connection.- , _outgoingFlowControl :: OutgoingFlowControl- -- ^ Simple getter for the 'OutgoingFlowControl' for the whole client- -- connection.- , _payloadSplitter :: IO PayloadSplitter- -- ^ Returns a function to split a payload.- , _asyncs :: !Http2ClientAsyncs- -- ^ Asynchronous operations threads.- , _close :: IO ()- -- ^ Immediately stop processing incoming frames and closes the network- -- connection.- }--data InitHttp2Client = InitHttp2Client {- _initPing :: ByteString -> IO (IO (FrameHeader, FramePayload))- , _initSettings :: SettingsList -> IO (IO (FrameHeader, FramePayload))- , _initGoaway :: ErrorCodeId -> ByteString -> IO ()- , _initStartStream :: forall a. StreamStarter a- , _initIncomingFlowControl :: IncomingFlowControl- , _initOutgoingFlowControl :: OutgoingFlowControl- , _initPaylodSplitter :: IO PayloadSplitter- , _initClose :: IO ()- -- ^ Immediately closes the connection.- , _initStop :: IO Bool- -- ^ Stops receiving frames.- }---- | Set of Async threads running an Http2Client.------ This asyncs are linked to the thread where the Http2Client is created.--- If you modify this structure to add more Async, please also modify--- 'linkAsyncs' accordingly.-data Http2ClientAsyncs = Http2ClientAsyncs {- _waitSettingsAsync :: Async (FrameHeader, FramePayload)- -- ^ Async waiting for the initial settings ACK.- , _incomingFramesAsync :: Async ()- -- ^ Async responsible for ingesting all frames, increasing the- -- maximum-received streamID and starting the frame dispatch. See- -- 'dispatchFrames'.- }---- | Links all client's asyncs to current thread using:--- @ link someUnderlyingAsync @ .-linkAsyncs :: Http2Client -> IO ()-linkAsyncs client =- let Http2ClientAsyncs{..} = _asyncs client in do- link _waitSettingsAsync- link _incomingFramesAsync---- | Synonym of '_goaway'.------ https://github.com/http2/http2-spec/pull/366-_gtfo :: Http2Client -> ErrorCodeId -> ByteString -> IO ()-_gtfo = _goaway---- | Opaque proof that a client stream was initialized.------ This type is only useful to force calling '_headers' in '_initStream' and--- contains no information.-data StreamThread = CST---- | Record holding functions one can call while in an HTTP2 client stream.-data Http2Stream = Http2Stream {- _headers :: HPACK.HeaderList- -> (FrameFlags -> FrameFlags)- -> IO StreamThread- -- ^ Starts the stream with HTTP headers. Flags modifier can use- -- 'setEndStream' if no data is required passed the last block of headers.- -- Usually, this is the only call needed to build an '_initStream'.- , _prio :: Priority -> IO ()- -- ^ Changes the PRIORITY of this stream.- , _rst :: ErrorCodeId -> IO ()- -- ^ Resets this stream with a RST frame. You should not use this stream past this call.- , _waitEvent :: IO StreamEvent- -- ^ Waits for the next event on the stream.- , _sendDataChunk :: (FrameFlags -> FrameFlags) -> ByteString -> IO ()- -- ^ Sends a DATA frame chunk. You can use send empty frames with only- -- headers modifiers to close streams. This function is oblivious to framing- -- and hence does not respect the RFC if sending large blocks. Use 'sendData'- -- to chunk and send naively according to server\'s preferences. This function- -- can be useful if you intend to handle the framing yourself.- , _handlePushPromise :: StreamId -> HeaderList -> PushPromiseHandler -> IO ()- }---- | Sends HTTP trailers.------ Trailers should be the last thing sent over a stream.-trailers :: Http2Stream -> HPACK.HeaderList -> (FrameFlags -> FrameFlags) -> IO ()-trailers stream hdrs flagmod = void $ _headers stream hdrs flagmod---- | Handler upon receiving a PUSH_PROMISE from the server.------ The functions for 'Http2Stream' are similar to those used in ''. But callers--- shall not use '_headers' to initialize the PUSH_PROMISE stream. Rather,--- callers should 'waitHeaders' or '_rst' to reject the PUSH_PROMISE.------ The StreamId corresponds to the parent stream as PUSH_PROMISEs are tied to a--- client-initiated stream. Longer term we may move passing this handler to the--- '_startStream' instead of 'newHttp2Client' (as it is for now).-type PushPromiseHandler =- StreamId -> Http2Stream -> HeaderList -> IncomingFlowControl -> OutgoingFlowControl -> IO ()---- | Starts a new stream (i.e., one HTTP request + server-pushes).------ You will typically call the returned 'StreamStarter' immediately to define--- what you want to do with the Http2Stream.------ @ _ <- (withHttp2Stream myClient $ \stream -> StreamDefinition _ _) @------ Please refer to 'StreamStarter' and 'StreamDefinition' for more.-withHttp2Stream :: Http2Client -> StreamStarter a-withHttp2Stream = _startStream---- | Type synonym for functions that modify flags.------ Typical FlagSetter for library users are HTTP2.setEndHeader when sending--- headers HTTP2.setEndStream to signal that there the client is not willing to--- send more data.------ We might use Endo in the future.-type FlagSetter = FrameFlags -> FrameFlags---- | Sends the HTTP2+HTTP headers of your chosing.------ You must add HTTP2 pseudo-headers first, followed by your typical HTTP--- headers. This function makes no verification of this--- ordering/exhaustinevess.------ HTTP2 pseudo-headers replace the HTTP verb + parsed url as follows:--- ":method" such as "GET",--- ":scheme" such as "https",--- ":path" such as "/blog/post/1234?foo=bar",--- ":authority" such as "haskell.org"------ Note that we currently enforce the 'HTTP2.setEndHeader' but this design--- choice may change in the future. Hence, we recommend you use--- 'HTTP2.setEndHeader' as well.-headers :: Http2Stream -> HeaderList -> FlagSetter -> IO StreamThread-headers = _headers---- | Starts a new Http2Client around a frame connection.------ This function is slightly safer than 'startHttp2Client' because it uses--- 'Control.Concurrent.Async.withAsync' instead of--- 'Control.Concurrent.Async.async'; plus this function calls 'linkAsyncs' to--- make sure that a network error kills the controlling thread. However, this--- with-pattern takes the control of the thread and can be annoying at times.------ This function tries to finalize the client with a call to `_close`, a second--- call to `_close` will trigger an IOException because the Handle representing--- the TCP connection will be closed.-runHttp2Client- :: Http2FrameConnection- -- ^ A frame connection.- -> Int- -- ^ The buffersize for the Network.HPACK encoder.- -> Int- -- ^ The buffersize for the Network.HPACK decoder.- -> SettingsList- -- ^ Initial SETTINGS that are sent as first frame.- -> GoAwayHandler- -- ^ Actions to run when the remote sends a GoAwayFrame- -> FallBackFrameHandler- -- ^ Actions to run when a control frame is not yet handled in http2-client- -- lib (e.g., PRIORITY frames).- -> (Http2Client -> IO a)- -- ^ Actions to run on the client.- -> IO a-runHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler mainHandler = do- (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler- withAsync incomingLoop $ \aIncoming -> do- settsIO <- _initSettings initClient initSettings- withAsync settsIO $ \aSettings -> do- let client = Http2Client {- _settings = _initSettings initClient- , _ping = _initPing initClient- , _goaway = _initGoaway initClient- , _close =- _initStop initClient >> _initClose initClient- , _startStream = _initStartStream initClient- , _incomingFlowControl = _initIncomingFlowControl initClient- , _outgoingFlowControl = _initOutgoingFlowControl initClient- , _payloadSplitter = _initPaylodSplitter initClient- , _asyncs = Http2ClientAsyncs aSettings aIncoming- }- linkAsyncs client- ret <- mainHandler client- _close client- return ret---- | Starts a new Http2Client around a frame connection.------ You may want to 'linkAsyncs' for a proper and automated cleanup of the--- underlying threads.-newHttp2Client- :: Http2FrameConnection- -- ^ A frame connection.- -> Int- -- ^ The buffersize for the Network.HPACK encoder.- -> Int- -- ^ The buffersize for the Network.HPACK decoder.- -> SettingsList- -- ^ Initial SETTINGS that are sent as first frame.- -> GoAwayHandler- -- ^ Actions to run when the remote sends a GoAwayFrame- -> FallBackFrameHandler- -- ^ Actions to run when a control frame is not yet handled in http2-client- -- lib (e.g., PRIORITY frames).- -> IO Http2Client-newHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler = do- (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler- aIncoming <- async incomingLoop- settsIO <- _initSettings initClient initSettings- aSettings <- async settsIO- return $ Http2Client {- _settings = _initSettings initClient- , _ping = _initPing initClient- , _goaway = _initGoaway initClient- , _close =- _initStop initClient >> _initClose initClient- , _startStream = _initStartStream initClient- , _incomingFlowControl = _initIncomingFlowControl initClient- , _outgoingFlowControl = _initOutgoingFlowControl initClient- , _payloadSplitter = _initPaylodSplitter initClient- , _asyncs = Http2ClientAsyncs aSettings aIncoming- }--initHttp2Client- :: Http2FrameConnection- -> Int- -> Int- -> GoAwayHandler- -> FallBackFrameHandler- -> IO (IO (), InitHttp2Client)-initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler = do- let controlStream = makeFrameClientStream conn 0- let ackPing = sendPingFrame controlStream HTTP2.setAck- let ackSettings = sendSettingsFrame controlStream HTTP2.setAck []-- {- Setup for initial thread receiving server frames. -}- dispatch <- newDispatchIO- dispatchControl <- newDispatchControlIO encoderBufSize- ackPing- ackSettings- goAwayHandler- fallbackHandler-- let baseWindowSize = return HTTP2.defaultInitialWindowSize- _initIncomingFlowControl <- newIncomingFlowControl dispatchControl baseWindowSize (sendWindowUpdateFrame controlStream)- windowUpdatesChan <- newChan- _initOutgoingFlowControl <- newOutgoingFlowControl dispatchControl windowUpdatesChan baseWindowSize-- dispatchHPACK <- newDispatchHPACKIO decoderBufSize- (incomingLoop,endIncomingLoop) <- dispatchLoop conn dispatch dispatchControl windowUpdatesChan _initIncomingFlowControl dispatchHPACK-- {- Setup for client-initiated streams. -}- conccurentStreams <- newIORef 0- -- prepare client streams- clientStreamIdMutex <- newMVar 0- let withClientStreamId h = bracket (takeMVar clientStreamIdMutex)- (putMVar clientStreamIdMutex . succ)- (\k -> h (2 * k + 1)) -- Note: client StreamIds MUST be odd-- let _initStartStream getWork = do- maxConcurrency <- fromMaybe 100 . maxConcurrentStreams . _serverSettings <$> readSettings dispatchControl- roomNeeded <- atomicModifyIORef' conccurentStreams- (\n -> if n < maxConcurrency then (n + 1, 0) else (n, 1 + n - maxConcurrency))- if roomNeeded > 0- then- return $ Left $ TooMuchConcurrency roomNeeded- else Right <$> do- windowUpdatesChan <- newChan- cont <- withClientStreamId $ \sid -> do- dispatchStream <- newDispatchStreamIO sid- initializeStream conn- dispatch- dispatchControl- dispatchStream- windowUpdatesChan- getWork- Idle- v <- cont- atomicModifyIORef' conccurentStreams (\n -> (n - 1, ()))- pure v-- let _initPing dat = do- handler <- registerPingHandler dispatchControl dat- sendPingFrame controlStream id dat- return $ waitPingReply handler-- let _initSettings settslist = do- handler <- registerSetSettingsHandler dispatchControl- sendSettingsFrame controlStream id settslist- return $ do- ret <- waitSetSettingsReply handler- modifySettings dispatchControl- (\(ConnectionSettings cli srv) ->- (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ()))- return ret- let _initGoaway err errStr = do- sId <- readMaxReceivedStreamIdIO dispatch- sendGTFOFrame controlStream sId err errStr-- let _initPaylodSplitter = settingsPayloadSplitter <$> readSettings dispatchControl-- let _initStop = endIncomingLoop-- let _initClose = closeConnection conn-- return (incomingLoop, InitHttp2Client{..})--initializeStream- :: Http2FrameConnection- -> Dispatch- -> DispatchControl- -> DispatchStream- -> Chan (FrameHeader, FramePayload)- -> (Http2Stream -> StreamDefinition a)- -> StreamFSMState- -> IO (IO a)-initializeStream conn dispatch control stream windowUpdatesChan getWork initialState = do- let sid = _dispatchStreamId stream- let frameStream = makeFrameClientStream conn sid-- let events = _dispatchStreamReadEvents stream-- -- Prepare handlers.- let _headers headersList flags = do- splitter <- settingsPayloadSplitter <$> readSettings control- cst <- sendHeaders frameStream (_dispatchControlHpackEncoder control) headersList splitter flags- when (testEndStream $ flags 0) $ do- closeLocalStream dispatch sid- return cst- let _waitEvent = readChan events- let _sendDataChunk flags dat = do- sendDataFrame frameStream flags dat- when (testEndStream $ flags 0) $ do- closeLocalStream dispatch sid- let _rst = \err -> do- sendResetFrame frameStream err- closeReleaseStream dispatch sid- let _prio = sendPriorityFrame frameStream- let _handlePushPromise ppSid ppHeaders ppHandler = do- let mkStreamActions s = StreamDefinition (return CST) (ppHandler sid s ppHeaders)- newStream <- newDispatchStreamIO ppSid- ppWindowsUpdatesChan <- newChan- ppCont <- initializeStream conn- dispatch- control- newStream- ppWindowsUpdatesChan- mkStreamActions- ReservedRemote- ppCont-- let streamActions = getWork $ Http2Stream{..}-- -- Register handlers for receiving frames and perform the 1st action, the- -- stream won't be idle anymore.- registerStream dispatch sid (StreamState windowUpdatesChan events initialState)- _ <- _initStream streamActions-- -- Returns 2nd action.- -- We build the flow control contexts in this action outside the exclusive- -- lock on clientStreamIdMutex will be released.- return $ do- let baseIncomingWindowSize = initialWindowSize . _clientSettings <$> readSettings control- isfc <- newIncomingFlowControl control baseIncomingWindowSize (sendWindowUpdateFrame frameStream)- let baseOutgoingWindowSize = initialWindowSize . _serverSettings <$> readSettings control- osfc <- newOutgoingFlowControl control windowUpdatesChan baseOutgoingWindowSize- _handleStream streamActions isfc osfc--dispatchLoop- :: Http2FrameConnection- -> Dispatch- -> DispatchControl- -> Chan (FrameHeader, FramePayload)- -> IncomingFlowControl- -> DispatchHPACK- -> IO (IO (), IO Bool)-dispatchLoop conn d dc windowUpdatesChan inFlowControl dh = do- let getNextFrame = next conn- let go = delayException . forever $ do- frame <- getNextFrame- dispatchFramesStep frame d- whenFrame (hasStreamId 0) frame $ \got ->- dispatchControlFramesStep windowUpdatesChan got dc- whenFrame (hasTypeId [FrameData]) frame $ \got ->- creditDataFramesStep d inFlowControl got- whenFrame (hasTypeId [FrameWindowUpdate]) frame $ \got -> do- updateWindowsStep d got- whenFrame (hasTypeId [FramePushPromise, FrameHeaders]) frame $ \got -> do- let hpackLoop (FinishedWithHeaders curFh sId mkNewHdrs) = do- newHdrs <- mkNewHdrs- chan <- fmap _streamStateEvents <$> lookupStreamState d sId- let msg = StreamHeadersEvent curFh newHdrs- maybe (return ()) (flip writeChan msg) chan- hpackLoop (FinishedWithPushPromise curFh parentSid newSid mkNewHdrs) = do- newHdrs <- mkNewHdrs- chan <- fmap _streamStateEvents <$> lookupStreamState d parentSid- let msg = StreamPushPromiseEvent curFh newSid newHdrs- maybe (return ()) (flip writeChan msg) chan- hpackLoop (WaitContinuation act) =- getNextFrame >>= act >>= hpackLoop- hpackLoop (FailedHeaders curFh sId err) = do- chan <- fmap _streamStateEvents <$> lookupStreamState d sId- let msg = StreamErrorEvent curFh err- maybe (return ()) (flip writeChan msg) chan- hpackLoop (dispatchHPACKFramesStep got dh)- whenFrame (hasTypeId [FrameRSTStream]) frame $ \got -> do- handleRSTStep d got- finalizeFramesStep frame d- end <- newEmptyMVar- let run = void $ race go (takeMVar end)- let stop = tryPutMVar end ()- return (run, stop)--handleRSTStep- :: Dispatch- -> (FrameHeader, FramePayload)- -> IO ()-handleRSTStep d (fh, payload) = do- let sid = streamId fh- case payload of- (RSTStreamFrame err) -> do- chan <- fmap _streamStateEvents <$> lookupStreamState d sid- let msg = StreamErrorEvent fh (HTTP2.fromErrorCodeId err)- maybe (return ()) (flip writeChan msg) chan- closeReleaseStream d sid- _ ->- error $ "expecting RSTFrame but got " ++ show payload--dispatchFramesStep- :: (FrameHeader, Either HTTP2Error FramePayload)- -> Dispatch- -> IO ()-dispatchFramesStep (fh,_) d = do- let sid = streamId fh- -- Remember highest streamId.- atomicModifyIORef' (_dispatchMaxStreamId d) (\n -> (max n sid, ()))--finalizeFramesStep- :: (FrameHeader, Either HTTP2Error FramePayload)- -> Dispatch- -> IO ()-finalizeFramesStep (fh,_) d = do- let sid = streamId fh- -- Remote-close streams that match.- when (testEndStream $ flags fh) $ do- closeRemoteStream d sid--dispatchControlFramesStep- :: Chan (FrameHeader, FramePayload)- -> (FrameHeader, FramePayload)- -> DispatchControl- -> IO ()-dispatchControlFramesStep windowUpdatesChan controlFrame@(fh, payload) control@(DispatchControl{..}) = do- case payload of- (SettingsFrame settsList)- | not . testAck . flags $ fh -> do- atomicModifyIORef' _dispatchControlConnectionSettings- (\(ConnectionSettings cli srv) ->- (ConnectionSettings cli (HTTP2.updateSettings srv settsList), ()))- maybe (return ())- (_applySettings _dispatchControlHpackEncoder)- (lookup SettingsHeaderTableSize settsList)- _dispatchControlAckSettings- | otherwise -> do- handler <- lookupAndReleaseSetSettingsHandler control- maybe (return ()) (notifySetSettingsHandler controlFrame) handler- (PingFrame pingMsg)- | not . testAck . flags $ fh ->- _dispatchControlAckPing pingMsg- | otherwise -> do- handler <- lookupAndReleasePingHandler control pingMsg- maybe (return ()) (notifyPingHandler controlFrame) handler- (WindowUpdateFrame _ ) ->- writeChan windowUpdatesChan controlFrame- (GoAwayFrame lastSid errCode reason) ->- _dispatchControlOnGoAway $ RemoteSentGoAwayFrame lastSid errCode reason-- _ ->- _dispatchControlOnFallback controlFrame---- | We currently need a specific step in the main loop for crediting streams--- because a client user may programmatically reset and stop listening for a--- stream and stop calling waitData (which credits streams).------ TODO: modify the '_rst' function to wait and credit all the remaining data--- that could have been sent in flight-creditDataFramesStep- :: Dispatch- -> IncomingFlowControl- -> (FrameHeader, FramePayload)- -> IO ()-creditDataFramesStep d flowControl (fh,payload) = do- -- TODO: error if detect over-run. Current implementation credits- -- everything back. Hence, over-run should never happen.- _ <- _consumeCredit flowControl (HTTP2.payloadLength fh)- _addCredit flowControl (HTTP2.payloadLength fh)-- -- Write to the interested streams.- let sid = streamId fh- case payload of- (DataFrame dat) -> do- chan <- fmap _streamStateEvents <$> lookupStreamState d sid- maybe (return ()) (flip writeChan $ StreamDataEvent fh dat) chan- _ ->- error $ "expecting DataFrame but got " ++ show payload--updateWindowsStep- :: Dispatch- -> (FrameHeader, FramePayload)- -> IO ()-updateWindowsStep d got@(fh,_) = do- let sid = HTTP2.streamId fh- chan <- fmap _streamStateWindowUpdatesChan <$> lookupStreamState d sid- maybe (return ()) (flip writeChan got) chan --TODO: refer to RFC for erroring on idle/closed streams--data HPACKLoopDecision =- ForwardHeader !StreamId- | OpenPushPromise !StreamId !StreamId--data HPACKStepResult =- WaitContinuation !((FrameHeader, Either HTTP2Error FramePayload) -> IO HPACKStepResult)- | FailedHeaders !FrameHeader !StreamId ErrorCode- | FinishedWithHeaders !FrameHeader !StreamId (IO HeaderList)- | FinishedWithPushPromise !FrameHeader !StreamId !StreamId (IO HeaderList)--dispatchHPACKFramesStep- :: (FrameHeader, FramePayload)- -> DispatchHPACK- -> HPACKStepResult-dispatchHPACKFramesStep (fh,fp) (DispatchHPACK{..}) =- let (decision, pattern) = case fp of- PushPromiseFrame ppSid hbf -> do- (OpenPushPromise sid ppSid, Right hbf)- HeadersFrame _ hbf -> -- TODO: handle priority- (ForwardHeader sid, Right hbf)- RSTStreamFrame err ->- (ForwardHeader sid, Left err)- _ ->- error "wrong TypeId"- in go fh decision pattern- where- sid :: StreamId- sid = HTTP2.streamId fh-- go :: FrameHeader -> HPACKLoopDecision -> Either ErrorCodeId ByteString -> HPACKStepResult- go curFh decision (Right buffer) =- if not $ HTTP2.testEndHeader (HTTP2.flags curFh)- then WaitContinuation $ \frame -> do- let interrupted fh2 fp2 =- not $ hasTypeId [ FrameRSTStream , FrameContinuation ] fh2 fp2- whenFrameElse interrupted frame (\_ ->- error "invalid frame type while waiting for CONTINUATION")- (\(lastFh, lastFp) ->- case lastFp of- ContinuationFrame chbf ->- return $ go lastFh decision (Right (ByteString.append buffer chbf))- RSTStreamFrame err ->- return $ go lastFh decision (Left err)- _ ->- error "continued frame has invalid type")- else case decision of- ForwardHeader sId ->- FinishedWithHeaders curFh sId (decodeHeader _dispatchHPACKDynamicTable buffer)- OpenPushPromise parentSid newSid ->- FinishedWithPushPromise curFh parentSid newSid (decodeHeader _dispatchHPACKDynamicTable buffer)- go curFh _ (Left err) =- FailedHeaders curFh sid (HTTP2.fromErrorCodeId err)---newIncomingFlowControl- :: DispatchControl- -> IO Int- -- ^ Action to get the base window size.- -> (WindowSize -> IO())- -- ^ Action to send the window update to the server.- -> IO IncomingFlowControl-newIncomingFlowControl control getBase doSendUpdate = do- creditAdded <- newIORef 0- creditConsumed <- newIORef 0- let _addCredit n = atomicModifyIORef' creditAdded (\c -> (c + n, ()))- let _consumeCredit n = do- conso <- atomicModifyIORef' creditConsumed (\c -> (c + n, c + n))- base <- getBase- extra <- readIORef creditAdded- return $ base + extra - conso- let _updateWindow = do- base <- initialWindowSize . _clientSettings <$> readSettings control- added <- readIORef creditAdded- consumed <- readIORef creditConsumed-- let transferred = min added (HTTP2.maxWindowSize - base + consumed)- let shouldUpdate = transferred > 0-- _addCredit (negate transferred)- _ <- _consumeCredit (negate transferred)- when shouldUpdate (doSendUpdate transferred)-- return shouldUpdate- return $ IncomingFlowControl _addCredit _consumeCredit _updateWindow--newOutgoingFlowControl ::- DispatchControl- -- ^ Control dispatching reference.- -> Chan (FrameHeader, FramePayload)- -> IO Int- -- ^ Action to get the instantaneous base window-size.- -> IO OutgoingFlowControl-newOutgoingFlowControl control frames getBase = do- credit <- newIORef 0- let receive n = atomicModifyIORef' credit (\c -> (c + n, ()))- let withdraw 0 = return 0- withdraw n = do- base <- getBase- got <- atomicModifyIORef' credit (\c ->- if base + c >= n- then (c - n, n)- else (0 - base, base + c))- if got > 0- then return got- else do- amount <- race (waitSettingsChange base) (waitSomeCredit frames)- receive (either (const 0) id amount)- withdraw n- return $ OutgoingFlowControl receive withdraw- where- -- TODO: broadcast settings changes from ConnectionSettings using a better data type- -- than IORef+busy loop. Currently the busy loop is fine because- -- SettingsInitialWindowSize is typically set at the first frame and hence- -- waiting one second for an update that is likely to never come is- -- probably not an issue. There still is an opportunity risk, however, that- -- an hasted client asks for X > initialWindowSize before the server has- -- sent its initial SETTINGS frame.- waitSettingsChange prev = do- new <- initialWindowSize . _serverSettings <$> readSettings control- if new == prev then threadDelay 1000000 >> waitSettingsChange prev else return ()- waitSomeCredit frames = do- got <- readChan frames- case got of- (_, WindowUpdateFrame amt) ->- return amt- _ ->- error "got forwarded an unknown frame"--sendHeaders- :: Http2FrameClientStream- -> HpackEncoderContext- -> HeaderList- -> PayloadSplitter- -> (FrameFlags -> FrameFlags)- -> IO StreamThread-sendHeaders s enc hdrs blockSplitter flagmod = do- _sendFrames s mkFrames- return CST- where- mkFrames = do- headerBlockFragments <- blockSplitter <$> _encodeHeaders enc hdrs- let framers = (HeadersFrame Nothing) : repeat ContinuationFrame- let frames = zipWith ($) framers headerBlockFragments- let modifiersReversed = (HTTP2.setEndHeader . flagmod) : repeat id- let arrangedFrames = reverse $ zip modifiersReversed (reverse frames)- return arrangedFrames---- | A function able to split a header block into multiple fragments.-type PayloadSplitter = ByteString -> [ByteString]---- | Split headers like so that no payload exceeds server's maxFrameSize.-settingsPayloadSplitter :: ConnectionSettings -> PayloadSplitter-settingsPayloadSplitter (ConnectionSettings _ srv) =- fixedSizeChunks (maxFrameSize srv)---- | Breaks a ByteString into fixed-sized chunks.------ @ fixedSizeChunks 2 "hello" = ["he", "ll", "o"] @-fixedSizeChunks :: Int -> ByteString -> [ByteString]-fixedSizeChunks 0 _ = error "cannot chunk by zero-length blocks"-fixedSizeChunks _ "" = []-fixedSizeChunks len bstr =- let- (chunk, rest) = ByteString.splitAt len bstr- in- chunk : fixedSizeChunks len rest---- | Sends data, chunked according to the server's preferred chunk size.------ This function does not respect HTTP2 flow-control and send chunks--- sequentially. Hence, you should first ensure that you have enough--- flow-control credit (with '_withdrawCredit') or risk a connection failure.--- When you call _withdrawCredit keep in mind that HTTP2 has flow control at--- the stream and at the connection level. If you use `http2-client` in a--- multithreaded conext, you should avoid starving the connection-level--- flow-control.------ If you want to send bytestrings that fit in RAM, you can use--- 'Network.HTTP2.Client.Helpers.upload' as a function that implements--- flow-control.------ This function does not send frames back-to-back, that is, other frames may--- get interleaved between two chunks (for instance, to give priority to other--- streams, although no priority queue exists in `http2-client` so far).------ Please refer to '_sendDataChunk' and '_withdrawCredit' as well.-sendData :: Http2Client -> Http2Stream -> FlagSetter -> ByteString -> IO ()-sendData conn stream flagmod dat = do- splitter <- _payloadSplitter conn- let chunks = splitter dat- let pairs = reverse $ zip (flagmod : repeat id) (reverse chunks)- when (null chunks) $ _sendDataChunk stream flagmod ""- forM_ pairs $ \(flags, chunk) -> _sendDataChunk stream flags chunk--sendDataFrame- :: Http2FrameClientStream- -> (FrameFlags -> FrameFlags) -> ByteString -> IO ()-sendDataFrame s flagmod dat = do- sendOne s flagmod (DataFrame dat)--sendResetFrame :: Http2FrameClientStream -> ErrorCodeId -> IO ()-sendResetFrame s err = do- sendOne s id (RSTStreamFrame err)--sendGTFOFrame- :: Http2FrameClientStream- -> StreamId -> ErrorCodeId -> ByteString -> IO ()-sendGTFOFrame s lastStreamId err errStr = do- sendOne s id (GoAwayFrame lastStreamId err errStr)--rfcError :: String -> a-rfcError msg = error (msg ++ "draft-ietf-httpbis-http2-17")--sendPingFrame- :: Http2FrameClientStream- -> (FrameFlags -> FrameFlags)- -> ByteString- -> IO ()-sendPingFrame s flags dat- | _getStreamId s /= 0 =- rfcError "PING frames are not associated with any individual stream."- | ByteString.length dat /= 8 =- rfcError "PING frames MUST contain 8 octets"- | otherwise = sendOne s flags (PingFrame dat)--sendWindowUpdateFrame- :: Http2FrameClientStream -> WindowSize -> IO ()-sendWindowUpdateFrame s amount = do- let payload = WindowUpdateFrame amount- sendOne s id payload- return ()--sendSettingsFrame- :: Http2FrameClientStream- -> (FrameFlags -> FrameFlags) -> SettingsList -> IO ()-sendSettingsFrame s flags setts- | _getStreamId s /= 0 =- rfcError "The stream identifier for a SETTINGS frame MUST be zero (0x0)."- | otherwise = do- let payload = SettingsFrame setts- sendOne s flags payload- return ()--sendPriorityFrame :: Http2FrameClientStream -> Priority -> IO ()-sendPriorityFrame s p = do- let payload = PriorityFrame p- sendOne s id payload- return ()---- | Runs an action, rethrowing exception 50ms later.------ In a context where asynchronous are likely to occur this function gives a--- chance to other threads to do some work before Async linking reaps them all.------ In particular, servers are likely to close their TCP connection soon after--- sending a GoAwayFrame and we want to give a better chance to clients to--- observe the GoAwayFrame in their handlers to distinguish GoAwayFrames--- followed by TCP disconnection and plain TCP resets. As a result, this--- function is mostly used to delay 'dispatchFrames'. A more involved--- and future design will be to inline the various loop-processes for--- dispatchFrames and GoAwayHandlers in a same thread (e.g., using--- pipe/conduit to retain composability).-delayException :: IO a -> IO a-delayException act = act `catch` slowdown- where- slowdown :: SomeException -> IO a- slowdown e = threadDelay 50000 >> throwIO e-+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PackageImports #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}++{- | This module defines a set of low-level primitives for starting an HTTP2+session and interacting with a server.++For higher-level primitives, please refer to Network.HTTP2.Client.Helpers .+-}+module Network.HTTP2.Client (+ -- * Basics+ runHttp2Client,+ newHttp2Client,+ withHttp2Stream,+ headers,+ trailers,+ sendData,++ -- * Starting clients+ Http2Client (..),+ PushPromiseHandler,++ -- * Starting streams+ StreamDefinition (..),+ StreamStarter,+ TooMuchConcurrency (..),+ StreamThread,+ Http2Stream (..),++ -- * Flow control+ IncomingFlowControl (..),+ OutgoingFlowControl (..),++ -- * Exceptions+ linkAsyncs,+ RemoteSentGoAwayFrame (..),+ GoAwayHandler,+ defaultGoAwayHandler,++ -- * Misc.+ FallBackFrameHandler,+ ignoreFallbackHandler,+ FlagSetter,+ Http2ClientAsyncs (..),+ _gtfo,++ -- * Convenience re-exports+ StreamEvent (..),+ module Network.HTTP2.Client.FrameConnection,+ module Network.HTTP2.Client.Exceptions,+ module Network.Socket,+ module Network.TLS,+) where++import Control.Concurrent.Async.Lifted (Async, async, link, race, withAsync)+import Control.Concurrent.Lifted (threadDelay)+import Control.Concurrent.MVar.Lifted (newEmptyMVar, newMVar, putMVar, takeMVar, tryPutMVar)+import Control.Exception.Lifted (SomeException, bracket, catch, throwIO)+import Control.Monad (forM_, forever, void, when)+import Control.Monad.IO.Class (liftIO)+import Data.ByteString (ByteString)+import qualified Data.ByteString as ByteString+import Data.Foldable (foldl')+import Data.IORef.Lifted (atomicModifyIORef', newIORef, readIORef)+import Data.Maybe (fromMaybe)+import Network.HPACK as HPACK+import Network.HTTP2.Frame as HTTP2+import Network.Socket (HostName, PortNumber)+import Network.TLS (ClientParams)++import Network.HTTP2.Client.Channels+import Network.HTTP2.Client.Dispatch+import Network.HTTP2.Client.Exceptions+import Network.HTTP2.Client.FrameConnection++#if MIN_VERSION_http2(5,0,0)+import "http2" Network.HTTP2.Client (Settings, maxFrameSize, initialWindowSize, maxConcurrentStreams, headerTableSize, enablePush, maxHeaderListSize)+#endif++{- | Offers credit-based flow-control.++Any mutable changes are atomic and hence work as intended in a multithreaded+setup.++The design of the flow-control mechanism is subject to changes. One+important thing to keep in mind with current implementation is that both the+connection and streams are credited with '_addCredit' as soon as DATA frames+arrive, hence no-need to account for the DATA frames (but you can account+for delay-bandwidth product for instance).+-}+data IncomingFlowControl = IncomingFlowControl+ { _addCredit :: WindowSize -> IO ()+ -- ^ Add credit (using a hidden mutable reference underneath). This function+ -- only does accounting, the IO only does mutable changes. See '_updateWindow'.+ , _consumeCredit :: WindowSize -> IO Int+ -- ^ Consumes some credit and returns the credit left.+ , _updateWindow :: ClientIO Bool+ -- ^ Sends a WINDOW_UPDATE frame crediting it with the whole amount credited+ -- since the last _updateWindow call. The boolean tells whether an update was+ -- actually sent or not. A reason for not sending an update is if there is no+ -- credit in the flow-control system.+ }++{- | Receives credit-based flow-control or block.++There is no way to observe the total amount of credit and receive/withdraw+are atomic hence this object is thread-safe. However we plan to propose an+STM-based API to allow withdrawing atomically from both the connection and a+per-stream 'OutgoingFlowControl' objects at a same time. Without such+atomicity one must ensure consumers do not exhaust the connection credit+before taking the per-stream credit (else they might prevent others sending+data without taking any).++Longer term we plan to hide outgoing-flow-control increment/decrement+altogether because exception between withdrawing credit and sending DATA+could mean lost credit (and hence hanging streams).+-}+data OutgoingFlowControl = OutgoingFlowControl+ { _receiveCredit :: WindowSize -> IO ()+ -- ^ Add credit (using a hidden mutable reference underneath).+ , _withdrawCredit :: WindowSize -> ClientIO WindowSize+ -- ^ Wait until we can take credit from stash. The returned value correspond+ -- to the amount that could be withdrawn, which is min(current, wanted). A+ -- caller should withdraw credit to send DATA chunks and put back any unused+ -- credit with _receiveCredit.+ }++{- | Defines a client stream.++Please red the doc for this record fields and then see 'StreamStarter'.+-}+data StreamDefinition a = StreamDefinition+ { _initStream :: ClientIO StreamThread+ -- ^ Function to initialize a new client stream. This function runs in a+ -- exclusive-access section of the code and may prevent other threads to+ -- initialize new streams. Hence, you should ensure this IO does not wait for+ -- long periods of time.+ , _handleStream :: IncomingFlowControl -> OutgoingFlowControl -> ClientIO a+ -- ^ Function to operate with the stream. IncomingFlowControl currently is+ -- credited on your behalf as soon as a DATA frame arrives (and before you+ -- handle it with '_waitData'). However we do not send WINDOW_UPDATE with+ -- '_updateWindow'. This design may change in the future to give more leeway+ -- to library users.+ }++{- | Type alias for callback-based functions starting new streams.++The callback a user must provide takes an 'Http2Stream' and returns a+'StreamDefinition'. This construction may seem wrong because a 'StreamDefinition'+contains an initialization and a handler functions. The explanation for this+twistedness is as follows: in HTTP2 stream-ids must be monotonically+increasing, if we want to support multi-threaded clients we need to+serialize access to a critical region of the code when clients send+HEADERS+CONTINUATIONs frames.++Passing the 'Http2Stream' object as part of the callback avoids leaking the+implementation of the critical region, meanwhile, the 'StreamDefinition'+delimits this critical region.+-}+type StreamStarter a =+ (Http2Stream -> StreamDefinition a) -> ClientIO (Either TooMuchConcurrency a)++{- | Whether or not the client library believes the server will reject the new+stream. The Int content corresponds to the number of streams that should end+before accepting more streams. A reason this number can be more than zero is+that servers can change (and hence reduce) the advertised number of allowed+'maxConcurrentStreams' at any time.+-}+newtype TooMuchConcurrency = TooMuchConcurrency {_getStreamRoomNeeded :: Int}+ deriving (Show)++-- | Record holding functions one can call while in an HTTP2 client session.+data Http2Client = Http2Client+ { _ping :: ByteString -> ClientIO (ClientIO (FrameHeader, FramePayload))+ -- ^ Send a PING, the payload size must be exactly eight bytes.+ -- Returns an IO to wait for a ping reply. No timeout is provided. Only the+ -- first call to this IO will return if a reply is received. Hence we+ -- recommend wrapping this IO in an Async (e.g., with @race (threadDelay+ -- timeout)@.)+ , _settings :: SettingsList -> ClientIO (ClientIO (FrameHeader, FramePayload))+ -- ^ Sends a SETTINGS. Returns an IO to wait for a settings reply. No timeout+ -- is provided. Only the first call to this IO will return if a reply is+ -- received. Hence we recommend wrapping this IO in an Async (e.g., with+ -- @race (threadDelay timeout)@.)+ , _goaway :: ErrorCode -> ByteString -> ClientIO ()+ -- ^ Sends a GOAWAY.+ , _startStream :: forall a. StreamStarter a+ -- ^ Spawns new streams. See 'StreamStarter'.+ , _incomingFlowControl :: IncomingFlowControl+ -- ^ Simple getter for the 'IncomingFlowControl' for the whole client+ -- connection.+ , _outgoingFlowControl :: OutgoingFlowControl+ -- ^ Simple getter for the 'OutgoingFlowControl' for the whole client+ -- connection.+ , _payloadSplitter :: IO PayloadSplitter+ -- ^ Returns a function to split a payload.+ , _asyncs :: !Http2ClientAsyncs+ -- ^ Asynchronous operations threads.+ , _close :: ClientIO ()+ -- ^ Immediately stop processing incoming frames and closes the network+ -- connection.+ }++data InitHttp2Client = InitHttp2Client+ { _initPing :: ByteString -> ClientIO (ClientIO (FrameHeader, FramePayload))+ , _initSettings :: SettingsList -> ClientIO (ClientIO (FrameHeader, FramePayload))+ , _initGoaway :: ErrorCode -> ByteString -> ClientIO ()+ , _initStartStream :: forall a. StreamStarter a+ , _initIncomingFlowControl :: IncomingFlowControl+ , _initOutgoingFlowControl :: OutgoingFlowControl+ , _initPaylodSplitter :: IO PayloadSplitter+ , _initClose :: ClientIO ()+ -- ^ Immediately closes the connection.+ , _initStop :: ClientIO Bool+ -- ^ Stops receiving frames.+ }++{- | Set of Async threads running an Http2Client.++This asyncs are linked to the thread where the Http2Client is created.+If you modify this structure to add more Async, please also modify+'linkAsyncs' accordingly.+-}+data Http2ClientAsyncs = Http2ClientAsyncs+ { _waitSettingsAsync :: Async (Either ClientError (FrameHeader, FramePayload))+ -- ^ Async waiting for the initial settings ACK.+ , _incomingFramesAsync :: Async (Either ClientError ())+ -- ^ Async responsible for ingesting all frames, increasing the+ -- maximum-received streamID and starting the frame dispatch. See+ -- 'dispatchFrames'.+ }++{- | Links all client's asyncs to current thread using:+@ link someUnderlyingAsync @ .+-}+linkAsyncs :: Http2Client -> ClientIO ()+linkAsyncs client =+ let Http2ClientAsyncs{..} = _asyncs client+ in do+ link _waitSettingsAsync+ link _incomingFramesAsync++{- | Synonym of '_goaway'.++https://github.com/http2/http2-spec/pull/366+-}+_gtfo :: Http2Client -> ErrorCode -> ByteString -> ClientIO ()+_gtfo = _goaway++{- | Opaque proof that a client stream was initialized.++This type is only useful to force calling '_headers' in '_initStream' and+contains no information.+-}+data StreamThread = CST++-- | Record holding functions one can call while in an HTTP2 client stream.+data Http2Stream = Http2Stream+ { _headers ::+ HPACK.HeaderList ->+ (FrameFlags -> FrameFlags) ->+ ClientIO StreamThread+ -- ^ Starts the stream with HTTP headers. Flags modifier can use+ -- 'setEndStream' if no data is required passed the last block of headers.+ -- Usually, this is the only call needed to build an '_initStream'.+ , _prio :: Priority -> ClientIO ()+ -- ^ Changes the PRIORITY of this stream.+ , _rst :: ErrorCode -> ClientIO ()+ -- ^ Resets this stream with a RST frame. You should not use this stream past this call.+ , _waitEvent :: ClientIO StreamEvent+ -- ^ Waits for the next event on the stream.+ , _sendDataChunk :: (FrameFlags -> FrameFlags) -> ByteString -> ClientIO ()+ -- ^ Sends a DATA frame chunk. You can use send empty frames with only+ -- headers modifiers to close streams. This function is oblivious to framing+ -- and hence does not respect the RFC if sending large blocks. Use 'sendData'+ -- to chunk and send naively according to server\'s preferences. This function+ -- can be useful if you intend to handle the framing yourself.+ , _handlePushPromise :: StreamId -> HeaderList -> PushPromiseHandler -> ClientIO ()+ }++{- | Sends HTTP trailers.++Trailers should be the last thing sent over a stream.+-}+trailers :: Http2Stream -> HPACK.HeaderList -> (FrameFlags -> FrameFlags) -> ClientIO ()+trailers stream hdrs flagmod = void $ _headers stream hdrs flagmod++{- | Handler upon receiving a PUSH_PROMISE from the server.++The functions for 'Http2Stream' are similar to those used in ''. But callers+shall not use '_headers' to initialize the PUSH_PROMISE stream. Rather,+callers should 'waitHeaders' or '_rst' to reject the PUSH_PROMISE.++The StreamId corresponds to the parent stream as PUSH_PROMISEs are tied to a+client-initiated stream. Longer term we may move passing this handler to the+'_startStream' instead of 'newHttp2Client' (as it is for now).+-}+type PushPromiseHandler =+ StreamId -> Http2Stream -> HeaderList -> IncomingFlowControl -> OutgoingFlowControl -> ClientIO ()++{- | Starts a new stream (i.e., one HTTP request + server-pushes).++You will typically call the returned 'StreamStarter' immediately to define+what you want to do with the Http2Stream.++@ _ <- (withHttp2Stream myClient $ \stream -> StreamDefinition _ _) @++Please refer to 'StreamStarter' and 'StreamDefinition' for more.+-}+withHttp2Stream :: Http2Client -> StreamStarter a+withHttp2Stream client = _startStream client++{- | Type synonym for functions that modify flags.++Typical FlagSetter for library users are HTTP2.setEndHeader when sending+headers HTTP2.setEndStream to signal that there the client is not willing to+send more data.++We might use Endo in the future.+-}+type FlagSetter = FrameFlags -> FrameFlags++{- | Sends the HTTP2+HTTP headers of your chosing.++You must add HTTP2 pseudo-headers first, followed by your typical HTTP+headers. This function makes no verification of this+ordering/exhaustinevess.++HTTP2 pseudo-headers replace the HTTP verb + parsed url as follows:+":method" such as "GET",+":scheme" such as "https",+":path" such as "/blog/post/1234?foo=bar",+":authority" such as "haskell.org"++Note that we currently enforce the 'HTTP2.setEndHeader' but this design+choice may change in the future. Hence, we recommend you use+'HTTP2.setEndHeader' as well.+-}+headers :: Http2Stream -> HeaderList -> FlagSetter -> ClientIO StreamThread+headers = _headers++{- | Starts a new Http2Client around a frame connection.++This function is slightly safer than 'startHttp2Client' because it uses+'Control.Concurrent.Async.withAsync' instead of+'Control.Concurrent.Async.async'; plus this function calls 'linkAsyncs' to+make sure that a network error kills the controlling thread. However, this+with-pattern takes the control of the thread and can be annoying at times.++This function tries to finalize the client with a call to `_close`, a second+call to `_close` will trigger an IOException because the Handle representing+the TCP connection will be closed.+-}+runHttp2Client ::+ -- | A frame connection.+ Http2FrameConnection ->+ -- | The buffersize for the Network.HPACK encoder.+ Int ->+ -- | The buffersize for the Network.HPACK decoder.+ Int ->+ -- | Initial SETTINGS that are sent as first frame.+ SettingsList ->+ -- | Actions to run when the remote sends a GoAwayFrame+ GoAwayHandler ->+ -- | Actions to run when a control frame is not yet handled in http2-client+ -- lib (e.g., PRIORITY frames).+ FallBackFrameHandler ->+ -- | Actions to run on the client.+ (Http2Client -> ClientIO a) ->+ ClientIO a+runHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler mainHandler = do+ (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler+ withAsync incomingLoop $ \aIncoming -> do+ settsIO <- _initSettings initClient initSettings+ withAsync settsIO $ \aSettings -> do+ let client =+ Http2Client+ { _settings = _initSettings initClient+ , _ping = _initPing initClient+ , _goaway = _initGoaway initClient+ , _close =+ _initStop initClient >> _initClose initClient+ , _startStream = _initStartStream initClient+ , _incomingFlowControl = _initIncomingFlowControl initClient+ , _outgoingFlowControl = _initOutgoingFlowControl initClient+ , _payloadSplitter = _initPaylodSplitter initClient+ , _asyncs = Http2ClientAsyncs aSettings aIncoming+ }+ linkAsyncs client+ ret <- mainHandler client+ _close client+ return ret++{- | Starts a new Http2Client around a frame connection.++You may want to 'linkAsyncs' for a proper and automated cleanup of the+underlying threads.+-}+newHttp2Client ::+ -- | A frame connection.+ Http2FrameConnection ->+ -- | The buffersize for the Network.HPACK encoder.+ Int ->+ -- | The buffersize for the Network.HPACK decoder.+ Int ->+ -- | Initial SETTINGS that are sent as first frame.+ SettingsList ->+ -- | Actions to run when the remote sends a GoAwayFrame+ GoAwayHandler ->+ -- | Actions to run when a control frame is not yet handled in http2-client+ -- lib (e.g., PRIORITY frames).+ FallBackFrameHandler ->+ ClientIO Http2Client+newHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler = do+ (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler+ aIncoming <- async incomingLoop+ settsIO <- _initSettings initClient initSettings+ aSettings <- async settsIO+ return $+ Http2Client+ { _settings = _initSettings initClient+ , _ping = _initPing initClient+ , _goaway = _initGoaway initClient+ , _close =+ _initStop initClient >> _initClose initClient+ , _startStream = _initStartStream initClient+ , _incomingFlowControl = _initIncomingFlowControl initClient+ , _outgoingFlowControl = _initOutgoingFlowControl initClient+ , _payloadSplitter = _initPaylodSplitter initClient+ , _asyncs = Http2ClientAsyncs aSettings aIncoming+ }++initHttp2Client ::+ Http2FrameConnection ->+ Int ->+ Int ->+ GoAwayHandler ->+ FallBackFrameHandler ->+ ClientIO (ClientIO (), InitHttp2Client)+initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler = do+ let controlStream = makeFrameClientStream conn 0+ let ackPing = sendPingFrame controlStream HTTP2.setAck+ let ackSettings = sendSettingsFrame controlStream HTTP2.setAck []++ {- Setup for initial thread receiving server frames. -}+ dispatch <- newDispatchIO+ dispatchControl <-+ newDispatchControlIO+ encoderBufSize+ ackPing+ ackSettings+ goAwayHandler+ fallbackHandler++ let baseIncomingWindowSize = initialWindowSize . _clientSettings <$> readSettings dispatchControl+ baseOutgoingWindowSize = initialWindowSize . _serverSettings <$> readSettings dispatchControl+ _initIncomingFlowControl <- lift $ newIncomingFlowControl dispatchControl baseIncomingWindowSize (sendWindowUpdateFrame controlStream)+ windowUpdatesChan <- newChan+ _initOutgoingFlowControl <- lift $ newOutgoingFlowControl dispatchControl windowUpdatesChan baseOutgoingWindowSize++ dispatchHPACK <- newDispatchHPACKIO decoderBufSize+ (incomingLoop, endIncomingLoop) <- dispatchLoop conn dispatch dispatchControl windowUpdatesChan _initIncomingFlowControl dispatchHPACK++ {- Setup for client-initiated streams. -}+ conccurentStreams <- newIORef 0+ -- prepare client streams+ clientStreamIdMutex <- newMVar 0+ let withClientStreamId h =+ bracket+ (takeMVar clientStreamIdMutex)+ (putMVar clientStreamIdMutex . succ)+ (\k -> h (2 * k + 1)) -- Note: client StreamIds MUST be odd+ let _initStartStream getWork = do+ maxConcurrency <- fromMaybe 100 . maxConcurrentStreams . _serverSettings <$> readSettings dispatchControl+ roomNeeded <-+ atomicModifyIORef'+ conccurentStreams+ (\n -> if n < maxConcurrency then (n + 1, 0) else (n, 1 + n - maxConcurrency))+ if roomNeeded > 0+ then+ return $ Left $ TooMuchConcurrency roomNeeded+ else+ Right <$> do+ windowUpdatesChan <- newChan+ cont <- withClientStreamId $ \sid -> do+ dispatchStream <- newDispatchStreamIO sid+ initializeStream+ conn+ dispatch+ dispatchControl+ dispatchStream+ windowUpdatesChan+ getWork+ Idle+ v <- cont+ atomicModifyIORef' conccurentStreams (\n -> (n - 1, ()))+ pure v++ let _initPing dat = do+ handler <- lift $ registerPingHandler dispatchControl dat+ sendPingFrame controlStream id dat+ return $ lift $ waitPingReply handler++ let _initSettings settslist = do+ handler <- lift $ registerSetSettingsHandler dispatchControl+ sendSettingsFrame controlStream id settslist+ return $ do+ ret <- lift $ waitSetSettingsReply handler+ modifySettings+ dispatchControl+ ( \(ConnectionSettings cli srv) ->+ (ConnectionSettings (compat_updateSettings cli settslist) srv, ())+ )+ return ret+ let _initGoaway err errStr = do+ sId <- lift $ readMaxReceivedStreamIdIO dispatch+ sendGTFOFrame controlStream sId err errStr++ let _initPaylodSplitter = settingsPayloadSplitter <$> readSettings dispatchControl++ let _initStop = endIncomingLoop++ let _initClose = closeConnection conn++ return (incomingLoop, InitHttp2Client{..})++initializeStream ::+ Http2FrameConnection ->+ Dispatch ->+ DispatchControl ->+ DispatchStream ->+ Chan (FrameHeader, FramePayload) ->+ (Http2Stream -> StreamDefinition a) ->+ StreamFSMState ->+ ClientIO (ClientIO a)+initializeStream conn dispatch control stream windowUpdatesChan getWork initialState = do+ let sid = _dispatchStreamId stream+ let frameStream = makeFrameClientStream conn sid++ let events = _dispatchStreamReadEvents stream++ -- Prepare handlers.+ let _headers headersList flags = do+ splitter <- settingsPayloadSplitter <$> readSettings control+ cst <- sendHeaders frameStream (_dispatchControlHpackEncoder control) headersList splitter flags+ when (testEndStream $ flags 0) $ do+ closeLocalStream dispatch sid+ return cst+ let _waitEvent = readChan events+ let _sendDataChunk flags dat = do+ sendDataFrame frameStream flags dat+ when (testEndStream $ flags 0) $ do+ closeLocalStream dispatch sid+ let _rst = \err -> do+ sendResetFrame frameStream err+ closeReleaseStream dispatch sid+ let _prio = sendPriorityFrame frameStream+ let _handlePushPromise ppSid ppHeaders ppHandler = do+ let mkStreamActions s = StreamDefinition (return CST) (ppHandler sid s ppHeaders)+ newStream <- newDispatchStreamIO ppSid+ ppWindowsUpdatesChan <- newChan+ ppCont <-+ initializeStream+ conn+ dispatch+ control+ newStream+ ppWindowsUpdatesChan+ mkStreamActions+ ReservedRemote+ ppCont++ let streamActions = getWork $ Http2Stream{..}++ -- Register handlers for receiving frames and perform the 1st action, the+ -- stream won't be idle anymore.+ registerStream dispatch sid (StreamState windowUpdatesChan events initialState)+ _ <- _initStream streamActions++ -- Returns 2nd action.+ -- We build the flow control contexts in this action outside the exclusive+ -- lock on clientStreamIdMutex will be released.+ return $ do+ let baseIncomingWindowSize = initialWindowSize . _clientSettings <$> readSettings control+ isfc <- lift $ newIncomingFlowControl control baseIncomingWindowSize (sendWindowUpdateFrame frameStream)+ let baseOutgoingWindowSize = initialWindowSize . _serverSettings <$> readSettings control+ osfc <- lift $ newOutgoingFlowControl control windowUpdatesChan baseOutgoingWindowSize+ _handleStream streamActions isfc osfc++dispatchLoop ::+ Http2FrameConnection ->+ Dispatch ->+ DispatchControl ->+ Chan (FrameHeader, FramePayload) ->+ IncomingFlowControl ->+ DispatchHPACK ->+ ClientIO (ClientIO (), ClientIO Bool)+dispatchLoop conn d dc windowUpdatesChan inFlowControl dh = do+ let getNextFrame = next conn+ let go = delayException . forever $ do+ frame <- getNextFrame+ dispatchFramesStep frame d+ whenFrame (hasStreamId 0) frame $ \got ->+ dispatchControlFramesStep windowUpdatesChan got dc+ whenFrame (hasTypeId [FrameData]) frame $ \got ->+ creditDataFramesStep d inFlowControl got+ whenFrame (hasTypeId [FrameWindowUpdate]) frame $ \got -> do+ updateWindowsStep d got+ whenFrame (hasTypeId [FramePushPromise, FrameHeaders]) frame $ \got -> do+ let hpackLoop (FinishedWithHeaders curFh sId mkNewHdrs) = lift $ do+ newHdrs <- mkNewHdrs+ chan <- fmap _streamStateEvents <$> lookupStreamState d sId+ let msg = StreamHeadersEvent curFh newHdrs+ maybe (return ()) (flip writeChan msg) chan+ hpackLoop (FinishedWithPushPromise curFh parentSid newSid mkNewHdrs) = lift $ do+ newHdrs <- mkNewHdrs+ chan <- fmap _streamStateEvents <$> lookupStreamState d parentSid+ let msg = StreamPushPromiseEvent curFh newSid newHdrs+ maybe (return ()) (flip writeChan msg) chan+ hpackLoop (WaitContinuation act) =+ getNextFrame >>= act >>= hpackLoop+ hpackLoop (FailedHeaders curFh sId err) = lift $ do+ chan <- fmap _streamStateEvents <$> lookupStreamState d sId+ let msg = StreamErrorEvent curFh err+ maybe (return ()) (flip writeChan msg) chan+ hpackLoop (dispatchHPACKFramesStep got dh)+ whenFrame (hasTypeId [FrameRSTStream]) frame $ \got -> do+ handleRSTStep d got+ finalizeFramesStep frame d+ end <- newEmptyMVar+ let run = void $ race go (takeMVar end)+ let stop = tryPutMVar end ()+ return (run, stop)++handleRSTStep ::+ Dispatch ->+ (FrameHeader, FramePayload) ->+ ClientIO ()+handleRSTStep d (fh, payload) = do+ let sid = streamId fh+ case payload of+ (RSTStreamFrame err) -> lift $ do+ chan <- fmap _streamStateEvents <$> lookupStreamState d sid+ let msg = StreamErrorEvent fh err+ maybe (return ()) (flip writeChan msg) chan+ closeReleaseStream d sid+ _ ->+ error $ "expecting RSTFrame but got " ++ show payload++dispatchFramesStep ::+ (FrameHeader, Either FrameDecodeError FramePayload) ->+ Dispatch ->+ ClientIO ()+dispatchFramesStep (fh, _) d = do+ let sid = streamId fh+ -- Remember highest streamId.+ atomicModifyIORef' (_dispatchMaxStreamId d) (\n -> (max n sid, ()))++finalizeFramesStep ::+ (FrameHeader, Either FrameDecodeError FramePayload) ->+ Dispatch ->+ ClientIO ()+finalizeFramesStep (fh, _) d = do+ let sid = streamId fh+ -- Remote-close streams that match.+ when (testEndStream $ flags fh) $ do+ closeRemoteStream d sid++dispatchControlFramesStep ::+ Chan (FrameHeader, FramePayload) ->+ (FrameHeader, FramePayload) ->+ DispatchControl ->+ ClientIO ()+dispatchControlFramesStep windowUpdatesChan controlFrame@(fh, payload) control@(DispatchControl{..}) = do+ case payload of+ (SettingsFrame settsList)+ | not . testAck . flags $ fh -> do+ atomicModifyIORef'+ _dispatchControlConnectionSettings+ ( \(ConnectionSettings cli srv) ->+ (ConnectionSettings cli (compat_updateSettings srv settsList), ())+ )+ lift $+ maybe+ (return ())+ (_applySettings _dispatchControlHpackEncoder)+ (lookup SettingsHeaderTableSize settsList)+ _dispatchControlAckSettings+ | otherwise -> do+ handler <- lookupAndReleaseSetSettingsHandler control+ maybe (return ()) (notifySetSettingsHandler controlFrame) handler+ (PingFrame pingMsg)+ | not . testAck . flags $ fh ->+ _dispatchControlAckPing pingMsg+ | otherwise -> do+ handler <- lookupAndReleasePingHandler control pingMsg+ maybe (return ()) (notifyPingHandler controlFrame) handler+ (WindowUpdateFrame _) ->+ writeChan windowUpdatesChan controlFrame+ (GoAwayFrame lastSid errCode reason) ->+ _dispatchControlOnGoAway $ RemoteSentGoAwayFrame lastSid errCode reason+ _ ->+ _dispatchControlOnFallback controlFrame++{- | We currently need a specific step in the main loop for crediting streams+because a client user may programmatically reset and stop listening for a+stream and stop calling waitData (which credits streams).++TODO: modify the '_rst' function to wait and credit all the remaining data+that could have been sent in flight+-}+creditDataFramesStep ::+ Dispatch ->+ IncomingFlowControl ->+ (FrameHeader, FramePayload) ->+ ClientIO ()+creditDataFramesStep d flowControl (fh, payload) = do+ -- TODO: error if detect over-run. Current implementation credits+ -- everything back. Hence, over-run should never happen.+ _ <- lift $ _consumeCredit flowControl (HTTP2.payloadLength fh)+ lift $ _addCredit flowControl (HTTP2.payloadLength fh)++ -- Write to the interested streams.+ let sid = streamId fh+ case payload of+ (DataFrame dat) -> do+ chan <- fmap _streamStateEvents <$> lookupStreamState d sid+ maybe (return ()) (flip writeChan $ StreamDataEvent fh dat) chan+ _ ->+ error $ "expecting DataFrame but got " ++ show payload++updateWindowsStep ::+ Dispatch ->+ (FrameHeader, FramePayload) ->+ ClientIO ()+updateWindowsStep d got@(fh, _) = do+ let sid = HTTP2.streamId fh+ chan <- fmap _streamStateWindowUpdatesChan <$> lookupStreamState d sid+ maybe (return ()) (flip writeChan got) chan -- TODO: refer to RFC for erroring on idle/closed streams++data HPACKLoopDecision+ = ForwardHeader !StreamId+ | OpenPushPromise !StreamId !StreamId++data HPACKStepResult+ = WaitContinuation !((FrameHeader, Either FrameDecodeError FramePayload) -> ClientIO HPACKStepResult)+ | FailedHeaders !FrameHeader !StreamId ErrorCode+ | FinishedWithHeaders !FrameHeader !StreamId (IO HeaderList)+ | FinishedWithPushPromise !FrameHeader !StreamId !StreamId (IO HeaderList)++dispatchHPACKFramesStep ::+ (FrameHeader, FramePayload) ->+ DispatchHPACK ->+ HPACKStepResult+dispatchHPACKFramesStep (fh, fp) (DispatchHPACK{..}) =+ let (decision, pattern) = case fp of+ PushPromiseFrame ppSid hbf -> do+ (OpenPushPromise sid ppSid, Right hbf)+ HeadersFrame _ hbf ->+ -- TODO: handle priority+ (ForwardHeader sid, Right hbf)+ RSTStreamFrame err ->+ (ForwardHeader sid, Left err)+ _ ->+ error "wrong TypeId"+ in go fh decision pattern+ where+ sid :: StreamId+ sid = HTTP2.streamId fh++ go :: FrameHeader -> HPACKLoopDecision -> Either ErrorCode ByteString -> HPACKStepResult+ go curFh decision (Right buffer) =+ if not $ HTTP2.testEndHeader (HTTP2.flags curFh)+ then WaitContinuation $ \frame -> do+ let interrupted fh2 fp2 =+ not $ hasTypeId [FrameRSTStream, FrameContinuation] fh2 fp2+ whenFrameElse+ interrupted+ frame+ ( \_ ->+ error "invalid frame type while waiting for CONTINUATION"+ )+ ( \(lastFh, lastFp) ->+ case lastFp of+ ContinuationFrame chbf ->+ return $ go lastFh decision (Right (ByteString.append buffer chbf))+ RSTStreamFrame err ->+ return $ go lastFh decision (Left err)+ _ ->+ error "continued frame has invalid type"+ )+ else case decision of+ ForwardHeader sId ->+ FinishedWithHeaders curFh sId (decodeHeader _dispatchHPACKDynamicTable buffer)+ OpenPushPromise parentSid newSid ->+ FinishedWithPushPromise curFh parentSid newSid (decodeHeader _dispatchHPACKDynamicTable buffer)+ go curFh _ (Left err) =+ FailedHeaders curFh sid err++newIncomingFlowControl ::+ DispatchControl ->+ -- | Action to get the base window size.+ IO Int ->+ -- | Action to send the window update to the server.+ (WindowSize -> ClientIO ()) ->+ IO IncomingFlowControl+newIncomingFlowControl control getBase doSendUpdate = do+ creditAdded <- newIORef 0+ creditConsumed <- newIORef 0+ let _addCredit n = atomicModifyIORef' creditAdded (\c -> (c + n, ()))+ let _consumeCredit n = do+ conso <- atomicModifyIORef' creditConsumed (\c -> (c + n, c + n))+ base <- getBase+ extra <- readIORef creditAdded+ return $ base + extra - conso+ let _updateWindow = do+ base <- initialWindowSize . _clientSettings <$> readSettings control+ added <- readIORef creditAdded+ consumed <- readIORef creditConsumed++ let transferred = min added (HTTP2.maxWindowSize - base + consumed)+ let shouldUpdate = transferred > 0++ _addCredit (negate transferred)+ _ <- lift $ _consumeCredit (negate transferred)+ when shouldUpdate (doSendUpdate transferred)++ return shouldUpdate+ return $ IncomingFlowControl _addCredit _consumeCredit _updateWindow++newOutgoingFlowControl ::+ -- | Control dispatching reference.+ DispatchControl ->+ Chan (FrameHeader, FramePayload) ->+ -- | Action to get the instantaneous base window-size.+ IO Int ->+ IO OutgoingFlowControl+newOutgoingFlowControl control frames getBase = do+ credit <- newIORef 0+ let receive n = atomicModifyIORef' credit (\c -> (c + n, ()))+ let withdraw 0 = return 0+ withdraw n = do+ base <- lift getBase+ got <-+ atomicModifyIORef'+ credit+ ( \c ->+ if base + c >= n+ then (c - n, n)+ else (0 - base, base + c)+ )+ if got > 0+ then return got+ else do+ amount <- race (waitSettingsChange base) (waitSomeCredit frames)+ receive (either (const 0) id amount)+ withdraw n+ return $ OutgoingFlowControl receive withdraw+ where+ -- TODO: broadcast settings changes from ConnectionSettings using a better data type+ -- than IORef+busy loop. Currently the busy loop is fine because+ -- SettingsInitialWindowSize is typically set at the first frame and hence+ -- waiting one second for an update that is likely to never come is+ -- probably not an issue. There still is an opportunity risk, however, that+ -- an hasted client asks for X > initialWindowSize before the server has+ -- sent its initial SETTINGS frame.+ waitSettingsChange prev = do+ new <- initialWindowSize . _serverSettings <$> readSettings control+ if new == prev then threadDelay 1000000 >> waitSettingsChange prev else return ()+ waitSomeCredit frames = do+ got <- readChan frames+ case got of+ (_, WindowUpdateFrame amt) ->+ return amt+ _ ->+ error "got forwarded an unknown frame"++sendHeaders ::+ Http2FrameClientStream ->+ HpackEncoderContext ->+ HeaderList ->+ PayloadSplitter ->+ (FrameFlags -> FrameFlags) ->+ ClientIO StreamThread+sendHeaders s enc hdrs blockSplitter flagmod = do+ _sendFrames s (lift mkFrames)+ return CST+ where+ mkFrames = do+ headerBlockFragments <- blockSplitter <$> _encodeHeaders enc hdrs+ let framers = (HeadersFrame Nothing) : repeat ContinuationFrame+ let frames = zipWith ($) framers headerBlockFragments+ let modifiersReversed = (HTTP2.setEndHeader . flagmod) : repeat id+ let arrangedFrames = reverse $ zip modifiersReversed (reverse frames)+ return arrangedFrames++-- | A function able to split a header block into multiple fragments.+type PayloadSplitter = ByteString -> [ByteString]++-- | Split headers like so that no payload exceeds server's maxFrameSize.+settingsPayloadSplitter :: ConnectionSettings -> PayloadSplitter+settingsPayloadSplitter (ConnectionSettings _ srv) =+ fixedSizeChunks (maxFrameSize srv)++{- | Breaks a ByteString into fixed-sized chunks.++@ fixedSizeChunks 2 "hello" = ["he", "ll", "o"] @+-}+fixedSizeChunks :: Int -> ByteString -> [ByteString]+fixedSizeChunks 0 _ = error "cannot chunk by zero-length blocks"+fixedSizeChunks _ "" = []+fixedSizeChunks len bstr =+ let+ (chunk, rest) = ByteString.splitAt len bstr+ in+ chunk : fixedSizeChunks len rest++{- | Sends data, chunked according to the server's preferred chunk size.++This function does not respect HTTP2 flow-control and send chunks+sequentially. Hence, you should first ensure that you have enough+flow-control credit (with '_withdrawCredit') or risk a connection failure.+When you call _withdrawCredit keep in mind that HTTP2 has flow control at+the stream and at the connection level. If you use `http2-client` in a+multithreaded conext, you should avoid starving the connection-level+flow-control.++If you want to send bytestrings that fit in RAM, you can use+'Network.HTTP2.Client.Helpers.upload' as a function that implements+flow-control.++This function does not send frames back-to-back, that is, other frames may+get interleaved between two chunks (for instance, to give priority to other+streams, although no priority queue exists in `http2-client` so far).++Please refer to '_sendDataChunk' and '_withdrawCredit' as well.+-}+sendData :: Http2Client -> Http2Stream -> FlagSetter -> ByteString -> ClientIO ()+sendData conn stream flagmod dat = do+ splitter <- lift $ _payloadSplitter conn+ let chunks = splitter dat+ let pairs = reverse $ zip (flagmod : repeat id) (reverse chunks)+ when (null chunks) $ _sendDataChunk stream flagmod ""+ forM_ pairs $ \(flags, chunk) -> _sendDataChunk stream flags chunk++sendDataFrame ::+ Http2FrameClientStream ->+ (FrameFlags -> FrameFlags) ->+ ByteString ->+ ClientIO ()+sendDataFrame s flagmod dat = do+ sendOne s flagmod (DataFrame dat)++sendResetFrame :: Http2FrameClientStream -> ErrorCode -> ClientIO ()+sendResetFrame s err = do+ sendOne s id (RSTStreamFrame err)++sendGTFOFrame ::+ Http2FrameClientStream ->+ StreamId ->+ ErrorCode ->+ ByteString ->+ ClientIO ()+sendGTFOFrame s lastStreamId err errStr = do+ sendOne s id (GoAwayFrame lastStreamId err errStr)++rfcError :: String -> a+rfcError msg = error (msg ++ "draft-ietf-httpbis-http2-17")++sendPingFrame ::+ Http2FrameClientStream ->+ (FrameFlags -> FrameFlags) ->+ ByteString ->+ ClientIO ()+sendPingFrame s flags dat+ | _getStreamId s /= 0 =+ rfcError "PING frames are not associated with any individual stream."+ | ByteString.length dat /= 8 =+ rfcError "PING frames MUST contain 8 octets"+ | otherwise = sendOne s flags (PingFrame dat)++sendWindowUpdateFrame ::+ Http2FrameClientStream -> WindowSize -> ClientIO ()+sendWindowUpdateFrame s amount = do+ let payload = WindowUpdateFrame amount+ sendOne s id payload+ return ()++sendSettingsFrame ::+ Http2FrameClientStream ->+ (FrameFlags -> FrameFlags) ->+ SettingsList ->+ ClientIO ()+sendSettingsFrame s flags setts+ | _getStreamId s /= 0 =+ rfcError "The stream identifier for a SETTINGS frame MUST be zero (0x0)."+ | otherwise = do+ let payload = SettingsFrame setts+ sendOne s flags payload+ return ()++sendPriorityFrame :: Http2FrameClientStream -> Priority -> ClientIO ()+sendPriorityFrame s p = do+ let payload = PriorityFrame p+ sendOne s id payload+ return ()++{- | Runs an action, rethrowing exception 50ms later.++In a context where asynchronous are likely to occur this function gives a+chance to other threads to do some work before Async linking reaps them all.++In particular, servers are likely to close their TCP connection soon after+sending a GoAwayFrame and we want to give a better chance to clients to+observe the GoAwayFrame in their handlers to distinguish GoAwayFrames+followed by TCP disconnection and plain TCP resets. As a result, this+function is mostly used to delay 'dispatchFrames'. A more involved+and future design will be to inline the various loop-processes for+dispatchFrames and GoAwayHandlers in a same thread (e.g., using+pipe/conduit to retain composability).+-}+delayException :: ClientIO a -> ClientIO a+delayException act = act `catch` slowdown+ where+ slowdown :: SomeException -> ClientIO a+ slowdown e = threadDelay 50000 >> throwIO e++compat_updateSettings :: Settings -> SettingsList -> Settings+#if MIN_VERSION_http2(5,0,0)+compat_updateSettings settings kvs = foldl' update settings kvs+ where+ update def (SettingsHeaderTableSize,x) = def { headerTableSize = x }+ -- fixme: x should be 0 or 1+ update def (SettingsEnablePush,x) = def { enablePush = x > 0 }+ update def (SettingsMaxConcurrentStreams,x) = def { maxConcurrentStreams = Just x }+ update def (SettingsInitialWindowSize,x) = def { initialWindowSize = x }+ update def (SettingsMaxFrameSize,x) = def { maxFrameSize = x }+ update def (SettingsMaxHeaderListSize,x) = def { maxHeaderListSize = Just x }+ update def _ = def++#else+compat_updateSettings = HTTP2.updateSettings+#endif
src/Network/HTTP2/Client/Channels.hs view
@@ -1,44 +1,52 @@+{-# LANGUAGE CPP #-} module Network.HTTP2.Client.Channels (- FramesChan- , hasStreamId- , hasTypeId- , whenFrame- , whenFrameElse- -- re-exports- , module Control.Concurrent.Chan- ) where+ FramesChan,+ hasStreamId,+ hasTypeId,+ whenFrame,+ whenFrameElse,+ -- re-exports+ module Control.Concurrent.Chan.Lifted,+) where -import Control.Concurrent.Chan (Chan, readChan, newChan, writeChan)-import Control.Exception (Exception, throwIO)-import Network.HTTP2 (StreamId, FrameHeader, FramePayload, FrameTypeId, framePayloadToFrameTypeId, streamId)+import Control.Concurrent.Chan.Lifted (Chan, newChan, readChan, writeChan)+import Control.Exception.Lifted (Exception, throwIO)+import Network.HTTP2.Frame (FrameDecodeError, FrameHeader, FramePayload, FrameType, StreamId, framePayloadToFrameType, streamId) +import Network.HTTP2.Client.Exceptions++#if MIN_VERSION_http2(5,0,0)+#else+instance Exception FrameDecodeError+#endif+ type FramesChan e = Chan (FrameHeader, Either e FramePayload) -whenFrame- :: Exception e- => (FrameHeader -> FramePayload -> Bool)- -> (FrameHeader, Either e FramePayload)- -> ((FrameHeader, FramePayload) -> IO ())- -> IO ()+whenFrame ::+ (Exception e) =>+ (FrameHeader -> FramePayload -> Bool) ->+ (FrameHeader, Either e FramePayload) ->+ ((FrameHeader, FramePayload) -> ClientIO ()) ->+ ClientIO () whenFrame test frame handle = do whenFrameElse test frame handle (const $ pure ()) -whenFrameElse- :: Exception e- => (FrameHeader -> FramePayload -> Bool)- -> (FrameHeader, Either e FramePayload)- -> ((FrameHeader, FramePayload) -> IO a)- -> ((FrameHeader, FramePayload) -> IO a)- -> IO a+whenFrameElse ::+ (Exception e) =>+ (FrameHeader -> FramePayload -> Bool) ->+ (FrameHeader, Either e FramePayload) ->+ ((FrameHeader, FramePayload) -> ClientIO a) ->+ ((FrameHeader, FramePayload) -> ClientIO a) ->+ ClientIO a whenFrameElse test (fHead, fPayload) handleTrue handleFalse = do dat <- either throwIO pure fPayload if test fHead dat- then handleTrue (fHead, dat)- else handleFalse (fHead, dat)+ then handleTrue (fHead, dat)+ else handleFalse (fHead, dat) hasStreamId :: StreamId -> FrameHeader -> FramePayload -> Bool hasStreamId sid h _ = streamId h == sid -hasTypeId :: [FrameTypeId] -> FrameHeader -> FramePayload -> Bool-hasTypeId tids _ p = framePayloadToFrameTypeId p `elem` tids+hasTypeId :: [FrameType] -> FrameHeader -> FramePayload -> Bool+hasTypeId tids _ p = framePayloadToFrameType p `elem` tids
src/Network/HTTP2/Client/Dispatch.hs view
@@ -1,124 +1,135 @@ {-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE PackageImports #-}+ module Network.HTTP2.Client.Dispatch where -import Control.Exception (throwIO)-import Data.ByteString (ByteString)+import Control.Exception (throwIO)+import Control.Monad.Base (MonadBase, liftBase)+import Data.ByteString (ByteString) import qualified Data.ByteString.Internal as ByteString-import Foreign.Marshal.Alloc (mallocBytes, finalizerFree)-import Foreign.ForeignPtr (newForeignPtr)-import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)-import Data.IntMap (IntMap)+import Data.IORef.Lifted (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.IntMap (IntMap) import qualified Data.IntMap as IntMap-import GHC.Exception (Exception)-import Network.HPACK as HPACK+import Foreign.ForeignPtr (newForeignPtr)+import Foreign.Marshal.Alloc (finalizerFree, mallocBytes)+import GHC.Exception (Exception)+import Network.HPACK as HPACK import qualified Network.HPACK.Token as HPACK-import Network.HTTP2 as HTTP2+import Network.HTTP2.Frame as HTTP2+#if MIN_VERSION_http2(5,0,0)+import "http2" Network.HTTP2.Client (Settings, defaultSettings)+#endif -import Network.HTTP2.Client.Channels+import Network.HTTP2.Client.Channels+import Network.HTTP2.Client.Exceptions -type DispatchChan = FramesChan HTTP2Error+type DispatchChan = FramesChan FrameDecodeError -- | A fallback handler for frames.-type FallBackFrameHandler = (FrameHeader, FramePayload) -> IO ()+type FallBackFrameHandler = (FrameHeader, FramePayload) -> ClientIO () -- | Default FallBackFrameHandler that ignores frames. ignoreFallbackHandler :: FallBackFrameHandler ignoreFallbackHandler = const $ pure () -- | An exception thrown when the server sends a GoAwayFrame.-data RemoteSentGoAwayFrame = RemoteSentGoAwayFrame !StreamId !ErrorCodeId !ByteString- deriving Show+data RemoteSentGoAwayFrame = RemoteSentGoAwayFrame !StreamId !ErrorCode !ByteString+ deriving (Show)+ instance Exception RemoteSentGoAwayFrame -- | A Handler for exceptional circumstances.-type GoAwayHandler = RemoteSentGoAwayFrame -> IO ()+type GoAwayHandler = RemoteSentGoAwayFrame -> ClientIO () --- | Default GoAwayHandler throws a 'RemoteSentGoAwayFrame' in the current--- thread.------ A probably sharper handler if you want to abruptly stop any operation is to--- get the 'ThreadId' of the main client thread and using--- 'Control.Exception.Base.throwTo'.------ There's an inherent race condition when receiving a GoAway frame because the--- server will likely close the connection which will lead to TCP errors as--- well.+{- | Default GoAwayHandler throws a 'RemoteSentGoAwayFrame' in the current+thread.++A probably sharper handler if you want to abruptly stop any operation is to+get the 'ThreadId' of the main client thread and using+'Control.Exception.Base.throwTo'.++There's an inherent race condition when receiving a GoAway frame because the+server will likely close the connection which will lead to TCP errors as+well.+-} defaultGoAwayHandler :: GoAwayHandler-defaultGoAwayHandler = throwIO+defaultGoAwayHandler = lift . throwIO -data StreamFSMState =- Idle- | ReservedRemote- | Open- | HalfClosedRemote- | HalfClosedLocal- | Closed+data StreamFSMState+ = Idle+ | ReservedRemote+ | Open+ | HalfClosedRemote+ | HalfClosedLocal+ | Closed -data StreamEvent =- StreamHeadersEvent !FrameHeader !HeaderList- | StreamPushPromiseEvent !FrameHeader !StreamId !HeaderList- | StreamDataEvent !FrameHeader ByteString- | StreamErrorEvent !FrameHeader ErrorCode- deriving Show+data StreamEvent+ = StreamHeadersEvent !FrameHeader !HeaderList+ | StreamPushPromiseEvent !FrameHeader !StreamId !HeaderList+ | StreamDataEvent !FrameHeader ByteString+ | StreamErrorEvent !FrameHeader ErrorCode+ deriving (Show) -data StreamState = StreamState {- _streamStateWindowUpdatesChan :: !(Chan (FrameHeader, FramePayload))- , _streamStateEvents :: !(Chan StreamEvent)- , _streamStateFSMState :: !StreamFSMState- }+data StreamState = StreamState+ { _streamStateWindowUpdatesChan :: !(Chan (FrameHeader, FramePayload))+ , _streamStateEvents :: !(Chan StreamEvent)+ , _streamStateFSMState :: !StreamFSMState+ } -data Dispatch = Dispatch {- _dispatchMaxStreamId :: !(IORef StreamId)- , _dispatchCurrentStreams :: !(IORef (IntMap StreamState))- }+data Dispatch = Dispatch+ { _dispatchMaxStreamId :: !(IORef StreamId)+ , _dispatchCurrentStreams :: !(IORef (IntMap StreamState))+ } -newDispatchIO :: IO Dispatch+newDispatchIO :: (MonadBase IO m) => m Dispatch newDispatchIO = Dispatch <$> newIORef 0 <*> newIORef (IntMap.empty) -readMaxReceivedStreamIdIO :: Dispatch -> IO StreamId+readMaxReceivedStreamIdIO :: (MonadBase IO m) => Dispatch -> m StreamId readMaxReceivedStreamIdIO = readIORef . _dispatchMaxStreamId -registerStream :: Dispatch -> StreamId -> StreamState -> IO ()+registerStream :: (MonadBase IO m) => Dispatch -> StreamId -> StreamState -> m () registerStream d sid st = atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->- let v = (IntMap.insert sid st xs) in (v, ())+ let v = (IntMap.insert sid st xs) in (v, ()) -lookupStreamState :: Dispatch -> StreamId -> IO (Maybe StreamState)+lookupStreamState :: (MonadBase IO m) => Dispatch -> StreamId -> m (Maybe StreamState) lookupStreamState d sid = IntMap.lookup sid <$> readIORef (_dispatchCurrentStreams d) -closeLocalStream :: Dispatch -> StreamId -> IO ()+closeLocalStream :: (MonadBase IO m) => Dispatch -> StreamId -> m () closeLocalStream d sid = atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->- let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ())+ let (_, v) = IntMap.updateLookupWithKey f sid xs in (v, ()) where f :: StreamId -> StreamState -> Maybe StreamState f _ st = case _streamStateFSMState st of HalfClosedRemote -> Nothing- Closed -> Nothing- _ -> Just $ st { _streamStateFSMState = HalfClosedLocal }+ Closed -> Nothing+ _ -> Just $ st{_streamStateFSMState = HalfClosedLocal} -closeRemoteStream :: Dispatch -> StreamId -> IO ()+closeRemoteStream :: (MonadBase IO m) => Dispatch -> StreamId -> m () closeRemoteStream d sid = atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->- let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ())+ let (_, v) = IntMap.updateLookupWithKey f sid xs in (v, ()) where f :: StreamId -> StreamState -> Maybe StreamState f _ st = case _streamStateFSMState st of- HalfClosedLocal -> Nothing- Closed -> Nothing- _ -> Just $ st { _streamStateFSMState = HalfClosedRemote }+ HalfClosedLocal -> Nothing+ Closed -> Nothing+ _ -> Just $ st{_streamStateFSMState = HalfClosedRemote} -closeReleaseStream :: Dispatch -> StreamId -> IO ()+closeReleaseStream :: (MonadBase IO m) => Dispatch -> StreamId -> m () closeReleaseStream d sid = atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->- let v = (IntMap.delete sid xs) in (v, ())+ let v = (IntMap.delete sid xs) in (v, ()) -- | Couples client and server settings together.-data ConnectionSettings = ConnectionSettings {- _clientSettings :: !Settings- , _serverSettings :: !Settings- }+data ConnectionSettings = ConnectionSettings+ { _clientSettings :: !Settings+ , _serverSettings :: !Settings+ } defaultConnectionSettings :: ConnectionSettings defaultConnectionSettings =@@ -126,34 +137,37 @@ data PingHandler = PingHandler !(Chan (FrameHeader, FramePayload)) -newPingHandler :: IO PingHandler+newPingHandler :: (MonadBase IO m) => m PingHandler newPingHandler = PingHandler <$> newChan -notifyPingHandler :: (FrameHeader, FramePayload) -> PingHandler -> IO ()+notifyPingHandler :: (MonadBase IO m) => (FrameHeader, FramePayload) -> PingHandler -> m () notifyPingHandler dat (PingHandler c) = writeChan c dat -waitPingReply :: PingHandler -> IO (FrameHeader, FramePayload)+waitPingReply :: (MonadBase IO m) => PingHandler -> m (FrameHeader, FramePayload) waitPingReply (PingHandler c) = readChan c data SetSettingsHandler = SetSettingsHandler !(Chan (FrameHeader, FramePayload)) -newSetSettingsHandler :: IO SetSettingsHandler+newSetSettingsHandler :: (MonadBase IO m) => m SetSettingsHandler newSetSettingsHandler = SetSettingsHandler <$> newChan -notifySetSettingsHandler :: (FrameHeader, FramePayload) -> SetSettingsHandler -> IO ()+notifySetSettingsHandler :: (MonadBase IO m) => (FrameHeader, FramePayload) -> SetSettingsHandler -> m () notifySetSettingsHandler dat (SetSettingsHandler c) = writeChan c dat -waitSetSettingsReply :: SetSettingsHandler -> IO (FrameHeader, FramePayload)+waitSetSettingsReply :: (MonadBase IO m) => SetSettingsHandler -> m (FrameHeader, FramePayload) waitSetSettingsReply (SetSettingsHandler c) = readChan c registerPingHandler :: DispatchControl -> ByteString -> IO PingHandler registerPingHandler dc dat = do handler <- newPingHandler- atomicModifyIORef' (_dispatchControlPingHandlers dc) (\xs ->- ((dat,handler):xs, ()))+ atomicModifyIORef'+ (_dispatchControlPingHandlers dc)+ ( \xs ->+ ((dat, handler) : xs, ())+ ) return handler -lookupAndReleasePingHandler :: DispatchControl -> ByteString -> IO (Maybe PingHandler)+lookupAndReleasePingHandler :: (MonadBase IO m) => DispatchControl -> ByteString -> m (Maybe PingHandler) lookupAndReleasePingHandler dc dat = atomicModifyIORef' (_dispatchControlPingHandlers dc) f where@@ -162,95 +176,105 @@ -- storing the handlers). f xs = (filter (\x -> dat /= fst x) xs, lookup dat xs) -registerSetSettingsHandler :: DispatchControl -> IO SetSettingsHandler+registerSetSettingsHandler :: (MonadBase IO m) => DispatchControl -> m SetSettingsHandler registerSetSettingsHandler dc = do handler <- newSetSettingsHandler- atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) (\xs ->- (handler:xs, ()))+ atomicModifyIORef'+ (_dispatchControlSetSettingsHandlers dc)+ ( \xs ->+ (handler : xs, ())+ ) return handler -lookupAndReleaseSetSettingsHandler :: DispatchControl -> IO (Maybe SetSettingsHandler)+lookupAndReleaseSetSettingsHandler :: (MonadBase IO m) => DispatchControl -> m (Maybe SetSettingsHandler) lookupAndReleaseSetSettingsHandler dc = atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) f where- f [] = ([], Nothing)- f (x:xs) = (xs, Just x)+ f [] = ([], Nothing)+ f (x : xs) = (xs, Just x) -data DispatchControl = DispatchControl {- _dispatchControlConnectionSettings :: !(IORef ConnectionSettings)- , _dispatchControlHpackEncoder :: !HpackEncoderContext- , _dispatchControlAckPing :: !(ByteString -> IO ())- , _dispatchControlAckSettings :: !(IO ())- , _dispatchControlOnGoAway :: !GoAwayHandler- , _dispatchControlOnFallback :: !FallBackFrameHandler- , _dispatchControlPingHandlers :: !(IORef [(ByteString, PingHandler)])- , _dispatchControlSetSettingsHandlers :: !(IORef [SetSettingsHandler])- }+data DispatchControl = DispatchControl+ { _dispatchControlConnectionSettings :: !(IORef ConnectionSettings)+ , _dispatchControlHpackEncoder :: !HpackEncoderContext+ , _dispatchControlAckPing :: !(ByteString -> ClientIO ())+ , _dispatchControlAckSettings :: !(ClientIO ())+ , _dispatchControlOnGoAway :: !GoAwayHandler+ , _dispatchControlOnFallback :: !FallBackFrameHandler+ , _dispatchControlPingHandlers :: !(IORef [(ByteString, PingHandler)])+ , _dispatchControlSetSettingsHandlers :: !(IORef [SetSettingsHandler])+ } -newDispatchControlIO- :: Size- -> (ByteString -> IO ())- -> (IO ())- -> GoAwayHandler- -> FallBackFrameHandler- -> IO DispatchControl+newDispatchControlIO ::+ (MonadBase IO m) =>+ Size ->+ (ByteString -> ClientIO ()) ->+ (ClientIO ()) ->+ GoAwayHandler ->+ FallBackFrameHandler ->+ m DispatchControl newDispatchControlIO encoderBufSize ackPing ackSetts onGoAway onFallback =- DispatchControl <$> newIORef defaultConnectionSettings- <*> newHpackEncoderContext encoderBufSize- <*> pure ackPing- <*> pure ackSetts- <*> pure onGoAway- <*> pure onFallback- <*> newIORef []- <*> newIORef []+ DispatchControl+ <$> newIORef defaultConnectionSettings+ <*> newHpackEncoderContext encoderBufSize+ <*> pure ackPing+ <*> pure ackSetts+ <*> pure onGoAway+ <*> pure onFallback+ <*> newIORef []+ <*> newIORef [] -newHpackEncoderContext :: Size -> IO HpackEncoderContext-newHpackEncoderContext encoderBufSize = do- let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True })+newHpackEncoderContext :: (MonadBase IO m) => Size -> m HpackEncoderContext+newHpackEncoderContext encoderBufSize = liftBase $ do+ let strategy = (HPACK.defaultEncodeStrategy{HPACK.useHuffman = True}) dt <- HPACK.newDynamicTableForEncoding HPACK.defaultDynamicTableSize buf <- mallocBytes encoderBufSize ptr <- newForeignPtr finalizerFree buf- return $ HpackEncoderContext- (encoder strategy dt buf ptr)+ return $+ HpackEncoderContext+ (\hdrs -> encoder strategy dt buf ptr hdrs) (\n -> HPACK.setLimitForEncoding n dt) where encoder strategy dt buf ptr hdrs = do- let hdrs' = fmap (\(k,v) -> let !t = HPACK.toToken k in (t,v)) hdrs+ let hdrs' = fmap (\(k, v) -> let !t = HPACK.toToken k in (t, v)) hdrs remainder <- HPACK.encodeTokenHeader buf encoderBufSize strategy True dt hdrs' case remainder of- ([],len) -> pure $ ByteString.fromForeignPtr ptr 0 len- (_,_) -> throwIO HPACK.BufferOverrun+ ([], len) -> pure $ ByteString.fromForeignPtr ptr 0 len+ (_, _) -> throwIO HPACK.BufferOverrun -readSettings :: DispatchControl -> IO ConnectionSettings+readSettings :: (MonadBase IO m) => DispatchControl -> m ConnectionSettings readSettings = readIORef . _dispatchControlConnectionSettings -modifySettings :: DispatchControl -> (ConnectionSettings -> (ConnectionSettings, a)) -> IO a+modifySettings :: (MonadBase IO m) => DispatchControl -> (ConnectionSettings -> (ConnectionSettings, a)) -> m a modifySettings d = atomicModifyIORef' (_dispatchControlConnectionSettings d) -- | Helper to carry around the HPACK encoder for outgoing header blocks..-data HpackEncoderContext = HpackEncoderContext {- _encodeHeaders :: HeaderList -> IO HeaderBlockFragment- , _applySettings :: Size -> IO ()- }+data HpackEncoderContext = HpackEncoderContext+ { _encodeHeaders :: HeaderList -> IO HeaderBlockFragment+ , _applySettings :: Size -> IO ()+ } -data DispatchHPACK = DispatchHPACK {- _dispatchHPACKDynamicTable :: !DynamicTable- }+data DispatchHPACK = DispatchHPACK+ { _dispatchHPACKDynamicTable :: !DynamicTable+ } -newDispatchHPACKIO :: Size -> IO DispatchHPACK+newDispatchHPACKIO :: (MonadBase IO m) => Size -> m DispatchHPACK newDispatchHPACKIO decoderBufSize =- DispatchHPACK <$> newDecoder+ liftBase $+ DispatchHPACK <$> newDecoder where- newDecoder = newDynamicTableForDecoding- HPACK.defaultDynamicTableSize- decoderBufSize+ newDecoder =+ newDynamicTableForDecoding+ HPACK.defaultDynamicTableSize+ decoderBufSize -data DispatchStream = DispatchStream {- _dispatchStreamId :: !StreamId- , _dispatchStreamReadEvents :: !(Chan StreamEvent)- }+data DispatchStream = DispatchStream+ { _dispatchStreamId :: !StreamId+ , _dispatchStreamReadEvents :: !(Chan StreamEvent)+ } -newDispatchStreamIO :: StreamId -> IO DispatchStream+newDispatchStreamIO :: (MonadBase IO m) => StreamId -> m DispatchStream newDispatchStreamIO sid =- DispatchStream <$> pure sid- <*> newChan+ liftBase $+ DispatchStream+ <$> pure sid+ <*> newChan
+ src/Network/HTTP2/Client/Exceptions.hs view
@@ -0,0 +1,24 @@++module Network.HTTP2.Client.Exceptions (+ ClientIO+ , ClientError(..)+ , runClientIO+ , module Control.Monad.Except+ , module Control.Monad.Trans+ ) where++import Control.Monad.Trans (lift)+import Control.Exception (Exception)+import Control.Monad.Except (ExceptT, runExceptT, throwError)++type ClientIO = ExceptT ClientError IO++runClientIO :: ClientIO a -> IO (Either ClientError a)+runClientIO = runExceptT++-- | A set of errors as observed from the client.+data ClientError = EarlyEndOfStream+ -- ^ We received a TCP end-of-stream and there will be no further read-IOs possible on the connection.+ deriving (Show,Ord,Eq)+instance Exception ClientError -- For people who want to rethrow using Control.Exception+
src/Network/HTTP2/Client/FrameConnection.hs view
@@ -1,10 +1,12 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE RecordWildCards #-}-{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE RankNTypes #-} module Network.HTTP2.Client.FrameConnection ( Http2FrameConnection(..) , newHttp2FrameConnection+ , frameHttp2RawConnection -- * Interact at the Frame level. , Http2ServerStream(..) , Http2FrameClientStream(..)@@ -16,14 +18,16 @@ ) where import Control.DeepSeq (deepseq)-import Control.Exception (bracket)-import Control.Concurrent.MVar (newMVar, takeMVar, putMVar)-import Control.Monad ((>=>), void)-import Network.HTTP2 (FrameHeader(..), FrameFlags, FramePayload, HTTP2Error, encodeInfo, decodeFramePayload)-import qualified Network.HTTP2 as HTTP2+import Control.Exception.Lifted (bracket)+import Control.Concurrent.MVar.Lifted (newMVar, takeMVar, putMVar)+import Control.Monad ((>=>), void, when)+import qualified Data.ByteString as ByteString+import Network.HTTP2.Frame (FrameHeader(..), FrameFlags, FramePayload, FrameDecodeError, encodeInfo, decodeFramePayload)+import qualified Network.HTTP2.Frame as HTTP2 import Network.Socket (HostName, PortNumber) import qualified Network.TLS as TLS +import Network.HTTP2.Client.Exceptions import Network.HTTP2.Client.RawConnection data Http2FrameConnection = Http2FrameConnection {@@ -31,12 +35,12 @@ -- ^ Starts a new client stream. , _serverStream :: Http2ServerStream -- ^ Receives frames from a server.- , _closeConnection :: IO ()+ , _closeConnection :: ClientIO () -- ^ Function that will close the network connection. } -- | Closes the Http2FrameConnection abruptly.-closeConnection :: Http2FrameConnection -> IO ()+closeConnection :: Http2FrameConnection -> ClientIO () closeConnection = _closeConnection -- | Creates a client stream.@@ -46,7 +50,7 @@ makeFrameClientStream = _makeFrameClientStream data Http2FrameClientStream = Http2FrameClientStream {- _sendFrames :: IO [(FrameFlags -> FrameFlags, FramePayload)] -> IO ()+ _sendFrames :: ClientIO [(FrameFlags -> FrameFlags, FramePayload)] -> ClientIO () -- ^ Sends a frame to the server. -- The first argument is a FrameFlags modifier (e.g., to sed the -- end-of-stream flag).@@ -54,36 +58,36 @@ } -- | Sends a frame to the server.-sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> IO ()+sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> ClientIO () sendOne client f payload = _sendFrames client (pure [(f, payload)]) -- | Sends multiple back-to-back frames to the server.-sendBackToBack :: Http2FrameClientStream -> [(FrameFlags -> FrameFlags, FramePayload)] -> IO ()+sendBackToBack :: Http2FrameClientStream -> [(FrameFlags -> FrameFlags, FramePayload)] -> ClientIO () sendBackToBack client payloads = _sendFrames client (pure payloads) data Http2ServerStream = Http2ServerStream {- _nextHeaderAndFrame :: IO (FrameHeader, Either HTTP2Error FramePayload)+ _nextHeaderAndFrame :: ClientIO (FrameHeader, Either FrameDecodeError FramePayload) } -- | Waits for the next frame from the server.-next :: Http2FrameConnection -> IO (FrameHeader, Either HTTP2Error FramePayload)+next :: Http2FrameConnection -> ClientIO (FrameHeader, Either FrameDecodeError FramePayload) next = _nextHeaderAndFrame . _serverStream -- | Adds framing around a 'RawHttp2Connection'. frameHttp2RawConnection :: RawHttp2Connection- -> IO Http2FrameConnection+ -> ClientIO Http2FrameConnection frameHttp2RawConnection http2conn = do -- Prepare a local mutex, this mutex should never escape the -- function's scope. Else it might lead to bugs (e.g.,- -- https://ro-che.info/articles/2014-07-30-bracket ) - writerMutex <- newMVar () + -- https://ro-che.info/articles/2014-07-30-bracket )+ writerMutex <- newMVar () let writeProtect io = bracket (takeMVar writerMutex) (putMVar writerMutex) (const io) -- Define handlers.- let makeClientStream streamID = + let makeClientStream streamID = let putFrame modifyFF frame = let info = encodeInfo modifyFF streamID in HTTP2.encodeFrame info frame@@ -96,13 +100,16 @@ in Http2FrameClientStream putFrames streamID nextServerFrameChunk = Http2ServerStream $ do- (fTy, fh@FrameHeader{..}) <- HTTP2.decodeFrameHeader <$> _nextRaw http2conn 9+ b9 <- _nextRaw http2conn 9+ when (ByteString.length b9 /= 9) $ throwError $ EarlyEndOfStream+ let (fTy, fh@FrameHeader{..}) = HTTP2.decodeFrameHeader b9 let decoder = decodeFramePayload fTy+ buf <- _nextRaw http2conn payloadLength+ when (ByteString.length buf /= payloadLength) $ throwError $ EarlyEndOfStream -- TODO: consider splitting the iteration here to give a chance to -- _not_ decode the frame, or consider lazyness enough.- let getNextFrame = decoder fh <$> _nextRaw http2conn payloadLength- nf <- getNextFrame- return (fh, nf)+ let nf = decoder fh buf+ pure (fh, nf) gtfo = _close http2conn @@ -112,6 +119,6 @@ newHttp2FrameConnection :: HostName -> PortNumber -> Maybe TLS.ClientParams- -> IO Http2FrameConnection+ -> ClientIO Http2FrameConnection newHttp2FrameConnection host port params = do frameHttp2RawConnection =<< newRawHttp2Connection host port params
src/Network/HTTP2/Client/Helpers.hs view
@@ -21,14 +21,15 @@ ) where import Data.Time.Clock (UTCTime, getCurrentTime)-import qualified Network.HTTP2 as HTTP2+import qualified Network.HTTP2.Frame as HTTP2 import qualified Network.HPACK as HPACK import Data.ByteString (ByteString) import qualified Data.ByteString as ByteString-import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (race)+import Control.Concurrent.Lifted (threadDelay)+import Control.Concurrent.Async.Lifted (race) import Network.HTTP2.Client+import Network.HTTP2.Client.Exceptions -- | Opaque type to express an action which timed out. data TimedOut = TimedOut@@ -45,12 +46,12 @@ -- ^ timeout in microseconds -> ByteString -- ^ 8-bytes message to uniquely identify the reply- -> IO PingReply+ -> ClientIO PingReply ping conn timeout msg = do- t0 <- getCurrentTime+ t0 <- lift $ getCurrentTime waitPing <- _ping conn msg pingReply <- race (threadDelay timeout >> return TimedOut) waitPing- t1 <- getCurrentTime+ t1 <- lift $ getCurrentTime return $ (t0, t1, pingReply) -- | Result containing the unpacked headers and all frames received in on a@@ -88,7 +89,7 @@ -- ^ The corresponding HTTP stream. -> OutgoingFlowControl -- ^ The flow control for this stream.- -> IO ()+ -> ClientIO () upload "" flagmod conn _ stream _ = do sendData conn stream flagmod "" upload dat flagmod conn connectionFlowControl stream streamFlowControl = do@@ -98,7 +99,7 @@ got <- _withdrawCredit connectionFlowControl gotStream -- Recredit the stream flow control with the excedent we cannot spend on -- the connection.- _receiveCredit streamFlowControl (gotStream - got)+ lift $ _receiveCredit streamFlowControl (gotStream - got) let uploadChunks flagMod = sendData conn stream flagMod (ByteString.take got dat)@@ -114,12 +115,16 @@ -- -- This function is fine if you don't want to consume results in chunks. See -- 'fromStreamResult' to collect the complicated 'StreamResult' into a simpler--- 'StramResponse'.-waitStream :: Http2Stream+-- 'StreamResponse'.+waitStream :: Http2Client+ -- ^The connection.+ -> Http2Stream+ -- ^The stream to wait on. This stream must be part of the connection. -> IncomingFlowControl+ -- ^Incoming flow control for the __stream__. -> PushPromiseHandler- -> IO StreamResult-waitStream stream streamFlowControl ppHandler = do+ -> ClientIO StreamResult+waitStream conn stream streamFlowControl ppHandler = do ev <- _waitEvent stream case ev of StreamHeadersEvent fH hdrs@@ -130,10 +135,11 @@ return (Right hdrs, reverse dfrms, trls) StreamPushPromiseEvent _ ppSid ppHdrs -> do _handlePushPromise stream ppSid ppHdrs ppHandler- waitStream stream streamFlowControl ppHandler+ waitStream conn stream streamFlowControl ppHandler _ -> error $ "expecting StreamHeadersEvent but got " ++ show ev where+ connFlowControl = _incomingFlowControl conn waitDataFrames xs = do ev <- _waitEvent stream case ev of@@ -141,9 +147,19 @@ | HTTP2.testEndStream (HTTP2.flags fh) -> return ((Right x):xs, Nothing) | otherwise -> do- _ <- _consumeCredit streamFlowControl (HTTP2.payloadLength fh)- _addCredit streamFlowControl (HTTP2.payloadLength fh)+ let size = HTTP2.payloadLength fh+ _ <- lift $ _consumeCredit streamFlowControl size+ lift $ _addCredit streamFlowControl size _ <- _updateWindow $ streamFlowControl+ -- We also send a WINDOW_UPDATE for the connection in order+ -- to not rely on external updateWindow calls. This reduces+ -- latency and makes the function less error-prone to use.+ -- Note that the main loop (dispatchLoop) already credits+ -- the connection for received data frames, but it does not+ -- send the WINDOW_UPDATE frames.. That is why we only send+ -- those frames here, and we do not credit the connection+ -- as we do the stream in the preceding lines.+ _ <- _updateWindow $ connFlowControl waitDataFrames ((Right x):xs) StreamPushPromiseEvent _ ppSid ppHdrs -> do _handlePushPromise stream ppSid ppHdrs ppHandler
src/Network/HTTP2/Client/RawConnection.hs view
@@ -5,29 +5,35 @@ module Network.HTTP2.Client.RawConnection ( RawHttp2Connection (..) , newRawHttp2Connection+ , newRawHttp2ConnectionUnix , newRawHttp2ConnectionSocket ) where import Control.Monad (forever, when)-import Control.Concurrent.Async (Async, async, cancel, pollSTM)-import Control.Concurrent.STM (STM, atomically, retry, throwSTM)+import Control.Concurrent.Async.Lifted (Async, async, cancel, pollSTM)+import Control.Concurrent.STM (STM, atomically, check, orElse, retry, throwSTM) import Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO, readTVar, writeTVar) import Data.ByteString (ByteString) import qualified Data.ByteString as ByteString import Data.ByteString.Lazy (fromChunks) import Data.Monoid ((<>))-import qualified Network.HTTP2 as HTTP2+import qualified Network.HTTP2.Frame as HTTP2 import Network.Socket hiding (recv) import Network.Socket.ByteString import qualified Network.TLS as TLS +import Network.HTTP2.Client.Exceptions+ -- TODO: catch connection errrors data RawHttp2Connection = RawHttp2Connection {- _sendRaw :: [ByteString] -> IO ()+ _sendRaw :: [ByteString] -> ClientIO () -- ^ Function to send raw data to the server.- , _nextRaw :: Int -> IO ByteString+ , _nextRaw :: Int -> ClientIO ByteString -- ^ Function to block reading a datachunk of a given size from the server.- , _close :: IO ()+ -- An empty chunk when asking for a length larger than 0 means the underlying+ -- network session is closed. A compliant HTP2 server should have sent a+ -- GOAWAY before such an event occurs.+ , _close :: ClientIO () } -- | Initiates a RawHttp2Connection with a server.@@ -40,16 +46,34 @@ -> Maybe TLS.ClientParams -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is -- overwritten to always return ["h2", "h2-17"].- -> IO RawHttp2Connection+ -> ClientIO RawHttp2Connection newRawHttp2Connection host port mparams = do -- Connects to TCP. let hints = defaultHints { addrFlags = [AI_NUMERICSERV], addrSocketType = Stream }- addr:_ <- getAddrInfo (Just hints) (Just host) (Just $ show port)- skt <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)- setSocketOption skt NoDelay 1- connect skt (addrAddress addr)- newRawHttp2ConnectionSocket skt mparams+ rSkt <- lift $ do+ addr:_ <- getAddrInfo (Just hints) (Just host) (Just $ show port)+ skt <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+ setSocketOption skt NoDelay 1+ connect skt (addrAddress addr)+ pure skt+ newRawHttp2ConnectionSocket rSkt mparams +-- | Initiates a RawHttp2Connection with a unix domain socket.+--+-- The current code does not handle closing the connexion, yikes.+newRawHttp2ConnectionUnix :: String+ -- ^ Path to the socket.+ -> Maybe TLS.ClientParams+ -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is+ -- overwritten to always return ["h2", "h2-17"].+ -> ClientIO RawHttp2Connection+newRawHttp2ConnectionUnix path mparams = do+ rSkt <- lift $ do+ skt <- socket AF_UNIX Stream 0+ connect skt $ SockAddrUnix path+ pure skt+ newRawHttp2ConnectionSocket rSkt mparams+ -- | Initiates a RawHttp2Connection with a server over a connected socket. -- -- The current code does not handle closing the connexion, yikes.@@ -60,10 +84,10 @@ -> Maybe TLS.ClientParams -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is -- overwritten to always return ["h2", "h2-17"].- -> IO RawHttp2Connection+ -> ClientIO RawHttp2Connection newRawHttp2ConnectionSocket skt mparams = do -- Prepare structure with abstract API.- conn <- maybe (plainTextRaw skt) (tlsRaw skt) mparams+ conn <- lift $ maybe (plainTextRaw skt) (tlsRaw skt) mparams -- Initializes the HTTP2 stream. _sendRaw conn [HTTP2.connectionPreface]@@ -74,8 +98,8 @@ plainTextRaw skt = do (b,putRaw) <- startWriteWorker (sendMany skt) (a,getRaw) <- startReadWorker (recv skt)- let doClose = cancel a >> cancel b >> close skt- return $ RawHttp2Connection (atomically . putRaw) (atomically . getRaw) doClose+ let doClose = lift $ cancel a >> cancel b >> close skt+ return $ RawHttp2Connection (lift . atomically . putRaw) (lift . atomically . getRaw) doClose tlsRaw :: Socket -> TLS.ClientParams -> IO RawHttp2Connection tlsRaw skt params = do@@ -85,9 +109,9 @@ (b,putRaw) <- startWriteWorker (TLS.sendData tlsContext . fromChunks) (a,getRaw) <- startReadWorker (const $ TLS.recvData tlsContext)- let doClose = cancel a >> cancel b >> TLS.bye tlsContext >> TLS.contextClose tlsContext+ let doClose = lift $ cancel a >> cancel b >> TLS.bye tlsContext >> TLS.contextClose tlsContext - return $ RawHttp2Connection (atomically . putRaw) (atomically . getRaw) doClose+ return $ RawHttp2Connection (lift . atomically . putRaw) (lift . atomically . getRaw) doClose where modifyParams prms = prms { TLS.clientHooks = (TLS.clientHooks prms) {@@ -117,14 +141,23 @@ :: (Int -> IO ByteString) -> IO (Async (), (Int -> STM ByteString)) startReadWorker get = do+ remoteClosed <- newTVarIO False+ let onEof = atomically $ writeTVar remoteClosed True+ let emptyByteStringOnEof = readTVar remoteClosed >>= check >> pure ""+ buf <- newTVarIO ""- a <- async $ readWorkerLoop buf get- return $ (a, getRawWorker a buf)+ a <- async $ readWorkerLoop buf get onEof -readWorkerLoop :: TVar ByteString -> (Int -> IO ByteString) -> IO ()-readWorkerLoop buf next = forever $ do- dat <- next 4096- atomically $ modifyTVar' buf (\bs -> (bs <> dat))+ return $ (a, \len -> getRawWorker a buf len `orElse` emptyByteStringOnEof)++readWorkerLoop :: TVar ByteString -> (Int -> IO ByteString) -> IO () -> IO ()+readWorkerLoop buf next onEof = go+ where+ go = do+ dat <- next 4096+ if ByteString.null dat+ then onEof+ else atomically (modifyTVar' buf (\bs -> (bs <> dat))) >> go getRawWorker :: Async () -> TVar ByteString -> Int -> STM ByteString getRawWorker a buf amount = do