diff --git a/http2-client.cabal b/http2-client.cabal --- a/http2-client.cabal +++ b/http2-client.cabal @@ -1,5 +1,5 @@ name: http2-client -version: 0.10.0.1 +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 @@ -27,14 +27,14 @@ , bytestring >= 0.11 && < 0.13 , containers >= 0.5 && < 0.8 , deepseq >= 1.4 && < 1.6 - , http2 >= 4.1 && < 5 + , 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 >= 2 && < 2.0.3 + , tls >= 1.8.0 && < 2.0.3 , transformers-base >= 0.4 && < 0.5 default-language: Haskell2010 diff --git a/src/Network/HTTP2/Client.hs b/src/Network/HTTP2/Client.hs --- a/src/Network/HTTP2/Client.hs +++ b/src/Network/HTTP2/Client.hs @@ -1,962 +1,1046 @@ -{-# LANGUAGE FlexibleContexts #-} -{-# 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.HTTP2.Client.Exceptions - , module Network.Socket - , module Network.TLS - ) where - -import Control.Concurrent.Async.Lifted (Async, async, race, withAsync, link) -import Control.Exception.Lifted (bracket, throwIO, SomeException, catch) -import Control.Concurrent.MVar.Lifted (newEmptyMVar, newMVar, putMVar, takeMVar, tryPutMVar) -import Control.Concurrent.Lifted (threadDelay) -import Control.Monad (forever, void, when, forM_) -import Control.Monad.IO.Class (liftIO) -import Data.ByteString (ByteString) -import qualified Data.ByteString as ByteString -import Data.IORef.Lifted (newIORef, atomicModifyIORef', 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 - --- | 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 - :: 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 -> ClientIO a) - -- ^ Actions to run on the client. - -> 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 - :: 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). - -> 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 (HTTP2.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 (HTTP2.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 - -> IO Int - -- ^ Action to get the base window size. - -> (WindowSize -> ClientIO ()) - -- ^ 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) - _ <- lift $ _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 <- 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 - +{-# 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 diff --git a/src/Network/HTTP2/Client/Channels.hs b/src/Network/HTTP2/Client/Channels.hs --- a/src/Network/HTTP2/Client/Channels.hs +++ b/src/Network/HTTP2/Client/Channels.hs @@ -1,45 +1,49 @@ +{-# LANGUAGE CPP #-} module Network.HTTP2.Client.Channels ( - FramesChan - , hasStreamId - , hasTypeId - , whenFrame - , whenFrameElse - -- re-exports - , module Control.Concurrent.Chan.Lifted - ) where + FramesChan, + hasStreamId, + hasTypeId, + whenFrame, + whenFrameElse, + -- re-exports + module Control.Concurrent.Chan.Lifted, +) where -import Control.Concurrent.Chan.Lifted (Chan, readChan, newChan, writeChan) -import Control.Exception.Lifted (Exception, throwIO) -import Network.HTTP2.Frame (StreamId, FrameHeader, FramePayload, FrameType, FrameDecodeError, framePayloadToFrameType, 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 +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) -> ClientIO ()) - -> ClientIO () +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) -> ClientIO a) - -> ((FrameHeader, FramePayload) -> ClientIO a) - -> ClientIO 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 diff --git a/src/Network/HTTP2/Client/Dispatch.hs b/src/Network/HTTP2/Client/Dispatch.hs --- a/src/Network/HTTP2/Client/Dispatch.hs +++ b/src/Network/HTTP2/Client/Dispatch.hs @@ -1,23 +1,29 @@ -{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE PackageImports #-} + module Network.HTTP2.Client.Dispatch where -import Control.Exception (throwIO) -import Control.Monad.Base (MonadBase, liftBase) -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.Lifted (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.Frame 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.Exceptions +import Network.HTTP2.Client.Channels +import Network.HTTP2.Client.Exceptions type DispatchChan = FramesChan FrameDecodeError @@ -30,98 +36,100 @@ -- | An exception thrown when the server sends a GoAwayFrame. data RemoteSentGoAwayFrame = RemoteSentGoAwayFrame !StreamId !ErrorCode !ByteString - deriving Show + deriving (Show) + instance Exception RemoteSentGoAwayFrame -- | A Handler for exceptional circumstances. 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 = 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 :: MonadBase IO m => m Dispatch +newDispatchIO :: (MonadBase IO m) => m Dispatch newDispatchIO = Dispatch <$> newIORef 0 <*> newIORef (IntMap.empty) -readMaxReceivedStreamIdIO :: MonadBase IO m => Dispatch -> m StreamId +readMaxReceivedStreamIdIO :: (MonadBase IO m) => Dispatch -> m StreamId readMaxReceivedStreamIdIO = readIORef . _dispatchMaxStreamId -registerStream :: MonadBase IO m => Dispatch -> StreamId -> StreamState -> m () +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 :: MonadBase IO m => Dispatch -> StreamId -> m (Maybe StreamState) +lookupStreamState :: (MonadBase IO m) => Dispatch -> StreamId -> m (Maybe StreamState) lookupStreamState d sid = IntMap.lookup sid <$> readIORef (_dispatchCurrentStreams d) -closeLocalStream :: MonadBase IO m => Dispatch -> StreamId -> m () +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 :: MonadBase IO m => Dispatch -> StreamId -> m () +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 :: MonadBase IO m => Dispatch -> StreamId -> m () +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 = @@ -129,34 +137,37 @@ data PingHandler = PingHandler !(Chan (FrameHeader, FramePayload)) -newPingHandler :: MonadBase IO m => m PingHandler +newPingHandler :: (MonadBase IO m) => m PingHandler newPingHandler = PingHandler <$> newChan -notifyPingHandler :: MonadBase IO m => (FrameHeader, FramePayload) -> PingHandler -> m () +notifyPingHandler :: (MonadBase IO m) => (FrameHeader, FramePayload) -> PingHandler -> m () notifyPingHandler dat (PingHandler c) = writeChan c dat -waitPingReply :: MonadBase IO m => PingHandler -> m (FrameHeader, FramePayload) +waitPingReply :: (MonadBase IO m) => PingHandler -> m (FrameHeader, FramePayload) waitPingReply (PingHandler c) = readChan c data SetSettingsHandler = SetSettingsHandler !(Chan (FrameHeader, FramePayload)) -newSetSettingsHandler :: MonadBase IO m => m SetSettingsHandler +newSetSettingsHandler :: (MonadBase IO m) => m SetSettingsHandler newSetSettingsHandler = SetSettingsHandler <$> newChan -notifySetSettingsHandler :: MonadBase IO m => (FrameHeader, FramePayload) -> SetSettingsHandler -> m () +notifySetSettingsHandler :: (MonadBase IO m) => (FrameHeader, FramePayload) -> SetSettingsHandler -> m () notifySetSettingsHandler dat (SetSettingsHandler c) = writeChan c dat -waitSetSettingsReply :: MonadBase IO m => SetSettingsHandler -> m (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 :: MonadBase IO m => DispatchControl -> ByteString -> m (Maybe PingHandler) +lookupAndReleasePingHandler :: (MonadBase IO m) => DispatchControl -> ByteString -> m (Maybe PingHandler) lookupAndReleasePingHandler dc dat = atomicModifyIORef' (_dispatchControlPingHandlers dc) f where @@ -165,96 +176,105 @@ -- storing the handlers). f xs = (filter (\x -> dat /= fst x) xs, lookup dat xs) -registerSetSettingsHandler :: MonadBase IO m => DispatchControl -> m 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 :: MonadBase IO m => DispatchControl -> m (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 -> ClientIO ()) - , _dispatchControlAckSettings :: !(ClientIO ()) - , _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 - :: MonadBase IO m - => Size - -> (ByteString -> ClientIO ()) - -> (ClientIO ()) - -> GoAwayHandler - -> FallBackFrameHandler - -> m 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 :: MonadBase IO m => Size -> m HpackEncoderContext +newHpackEncoderContext :: (MonadBase IO m) => Size -> m HpackEncoderContext newHpackEncoderContext encoderBufSize = liftBase $ do - let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True }) + let strategy = (HPACK.defaultEncodeStrategy{HPACK.useHuffman = True}) dt <- HPACK.newDynamicTableForEncoding HPACK.defaultDynamicTableSize buf <- mallocBytes encoderBufSize ptr <- newForeignPtr finalizerFree buf - return $ HpackEncoderContext + 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 :: MonadBase IO m => DispatchControl -> m ConnectionSettings +readSettings :: (MonadBase IO m) => DispatchControl -> m ConnectionSettings readSettings = readIORef . _dispatchControlConnectionSettings -modifySettings :: MonadBase IO m => DispatchControl -> (ConnectionSettings -> (ConnectionSettings, a)) -> m 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 :: MonadBase IO m => Size -> m DispatchHPACK -newDispatchHPACKIO decoderBufSize = liftBase $ - DispatchHPACK <$> newDecoder +newDispatchHPACKIO :: (MonadBase IO m) => Size -> m DispatchHPACK +newDispatchHPACKIO decoderBufSize = + 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 :: MonadBase IO m => StreamId -> m DispatchStream -newDispatchStreamIO sid = liftBase $ - DispatchStream <$> pure sid - <*> newChan +newDispatchStreamIO :: (MonadBase IO m) => StreamId -> m DispatchStream +newDispatchStreamIO sid = + liftBase $ + DispatchStream + <$> pure sid + <*> newChan