http2-client 0.8.0.2 → 0.9.0.0

raw patch · 9 files changed

+234/−170 lines, 9 filesdep +lifted-asyncdep +lifted-basedep +mtldep ~basePVP ok

version bump matches the API change (PVP)

Dependencies added: lifted-async, lifted-base, mtl, transformers-base

Dependency ranges changed: base

API changes (from Hackage documentation)

Files

• ChangeLog.md +4/−0
• http2-client.cabal +6/−1
• src/Network/HTTP2/Client.hs +81/−77
• src/Network/HTTP2/Client/Channels.hs +10/−8
• src/Network/HTTP2/Client/Dispatch.hs +39/−35
• src/Network/HTTP2/Client/Exceptions.hs +22/−0
• src/Network/HTTP2/Client/FrameConnection.hs +22/−16
• src/Network/HTTP2/Client/Helpers.hs +11/−10
• src/Network/HTTP2/Client/RawConnection.hs +39/−23

@@ -2,6 +2,10 @@  ## Unreleased changes +## v0.9.0.0++- introduce ClientIO as an error-carrying IO-monad for performing http2-calls+ ## v0.8.0.2  - first Changelog!

@@ -1,5 +1,5 @@ name:                http2-client-version:             0.8.0.2+version:             0.9.0.0 synopsis:            A native HTTP2 client library. description:         Please read the README.md at the homepage. homepage:            https://github.com/lucasdicioccio/http2-client@@ -16,6 +16,7 @@ library   hs-source-dirs:      src   exposed-modules:     Network.HTTP2.Client+                     , Network.HTTP2.Client.Exceptions                      , Network.HTTP2.Client.FrameConnection                      , Network.HTTP2.Client.Helpers                      , Network.HTTP2.Client.RawConnection@@ -27,10 +28,14 @@                      , containers >= 0.5 && < 1                      , deepseq >= 1.4 && < 2                      , http2 >= 1.6 && < 2+                     , lifted-async >= 0.10 && < 0.11+                     , lifted-base >= 0.2 && < 0.3+                     , mtl >= 2.2 && < 3                      , network >= 2.6 && < 3                      , stm >= 2.4 && < 3                      , time >= 1.8 && < 2                      , tls >= 1.4 && < 2+                     , transformers-base >= 0.4 && < 0.5   default-language:    Haskell2010  -- Commented-out to avoid distribution

@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts   #-} {-# LANGUAGE RecordWildCards    #-} {-# LANGUAGE RankNTypes         #-} {-# LANGUAGE OverloadedStrings  #-}@@ -40,18 +41,20 @@     -- * 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 (Async, async, race, withAsync, link)-import           Control.Exception (bracket, throwIO, SomeException, catch)-import           Control.Concurrent.MVar (newEmptyMVar, newMVar, putMVar, takeMVar, tryPutMVar)-import           Control.Concurrent (threadDelay)+import           Control.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 (newIORef, atomicModifyIORef', readIORef)+import           Data.IORef.Lifted (newIORef, atomicModifyIORef', readIORef) import           Data.Maybe (fromMaybe) import           Network.HPACK as HPACK import           Network.HTTP2 as HTTP2@@ -60,6 +63,7 @@  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.@@ -78,7 +82,7 @@   -- only does accounting, the IO only does mutable changes. See '_updateWindow'.   , _consumeCredit :: WindowSize -> IO Int   -- ^ Consumes some credit and returns the credit left.-  , _updateWindow :: IO Bool+  , _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@@ -101,7 +105,7 @@ data OutgoingFlowControl = OutgoingFlowControl {     _receiveCredit  :: WindowSize -> IO ()   -- ^ Add credit (using a hidden mutable reference underneath).-  , _withdrawCredit :: WindowSize -> IO WindowSize+  , _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@@ -112,12 +116,12 @@ -- -- Please red the doc for this record fields and then see 'StreamStarter'. data StreamDefinition a = StreamDefinition {-    _initStream   :: IO StreamThread+    _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 -> IO a+  , _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@@ -139,7 +143,7 @@ -- implementation of the critical region, meanwhile, the 'StreamDefinition' -- delimits this critical region. type StreamStarter a =-     (Http2Stream -> StreamDefinition a) -> IO (Either TooMuchConcurrency 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@@ -151,18 +155,18 @@  -- | Record holding functions one can call while in an HTTP2 client session. data Http2Client = Http2Client {-    _ping             :: ByteString -> IO (IO (FrameHeader, FramePayload))+    _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 -> IO (IO (FrameHeader, FramePayload))+  , _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           :: ErrorCodeId -> ByteString -> IO ()+  , _goaway           :: ErrorCodeId -> ByteString -> ClientIO ()   -- ^ Sends a GOAWAY.   , _startStream      :: forall a. StreamStarter a   -- ^ Spawns new streams. See 'StreamStarter'.@@ -176,22 +180,22 @@   -- ^ Returns a function to split a payload.   , _asyncs         :: !Http2ClientAsyncs   -- ^ Asynchronous operations threads.-  , _close           :: IO ()+  , _close           :: ClientIO ()   -- ^ Immediately stop processing incoming frames and closes the network   -- connection.   }  data InitHttp2Client = InitHttp2Client {-    _initPing                :: ByteString -> IO (IO (FrameHeader, FramePayload))-  , _initSettings            :: SettingsList -> IO (IO (FrameHeader, FramePayload))-  , _initGoaway              :: ErrorCodeId -> ByteString -> IO ()+    _initPing                :: ByteString -> ClientIO (ClientIO (FrameHeader, FramePayload))+  , _initSettings            :: SettingsList -> ClientIO (ClientIO (FrameHeader, FramePayload))+  , _initGoaway              :: ErrorCodeId -> ByteString -> ClientIO ()   , _initStartStream         :: forall a. StreamStarter a   , _initIncomingFlowControl :: IncomingFlowControl   , _initOutgoingFlowControl :: OutgoingFlowControl   , _initPaylodSplitter      :: IO PayloadSplitter-  , _initClose               :: IO ()+  , _initClose               :: ClientIO ()   -- ^ Immediately closes the connection.-  , _initStop                :: IO Bool+  , _initStop                :: ClientIO Bool   -- ^ Stops receiving frames.   } @@ -201,9 +205,9 @@ -- If you modify this structure to add more Async, please also modify -- 'linkAsyncs' accordingly. data Http2ClientAsyncs = Http2ClientAsyncs {-    _waitSettingsAsync   :: Async (FrameHeader, FramePayload)+    _waitSettingsAsync   :: Async (Either ClientError (FrameHeader, FramePayload))   -- ^ Async waiting for the initial settings ACK.-  , _incomingFramesAsync :: Async ()+  , _incomingFramesAsync :: Async (Either ClientError ())   -- ^ Async responsible for ingesting all frames, increasing the   -- maximum-received streamID and starting the frame dispatch. See   -- 'dispatchFrames'.@@ -211,7 +215,7 @@  -- | Links all client's asyncs to current thread using: -- @ link someUnderlyingAsync @ .-linkAsyncs :: Http2Client -> IO ()+linkAsyncs :: Http2Client -> ClientIO () linkAsyncs client =     let Http2ClientAsyncs{..} = _asyncs client in do             link _waitSettingsAsync@@ -220,7 +224,7 @@ -- | Synonym of '_goaway'. -- -- https://github.com/http2/http2-spec/pull/366-_gtfo :: Http2Client -> ErrorCodeId -> ByteString -> IO ()+_gtfo :: Http2Client -> ErrorCodeId -> ByteString -> ClientIO () _gtfo = _goaway  -- | Opaque proof that a client stream was initialized.@@ -233,29 +237,29 @@ data Http2Stream = Http2Stream {     _headers      :: HPACK.HeaderList                   -> (FrameFlags -> FrameFlags)-                  -> IO StreamThread+                  -> 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 -> IO ()+  , _prio         :: Priority -> ClientIO ()   -- ^ Changes the PRIORITY of this stream.-  , _rst          :: ErrorCodeId -> IO ()+  , _rst          :: ErrorCodeId -> ClientIO ()   -- ^ Resets this stream with a RST frame. You should not use this stream past this call.-  , _waitEvent    :: IO StreamEvent+  , _waitEvent    :: ClientIO StreamEvent   -- ^ Waits for the next event on the stream.-  , _sendDataChunk     :: (FrameFlags -> FrameFlags) -> ByteString -> IO ()+  , _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 -> IO ()+  , _handlePushPromise :: StreamId -> HeaderList -> PushPromiseHandler -> ClientIO ()   }  -- | Sends HTTP trailers. -- -- Trailers should be the last thing sent over a stream.-trailers :: Http2Stream -> HPACK.HeaderList -> (FrameFlags -> FrameFlags) -> IO ()+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.@@ -268,7 +272,7 @@ -- client-initiated stream. Longer term we may move passing this handler to the -- '_startStream' instead of 'newHttp2Client' (as it is for now). type PushPromiseHandler =-    StreamId -> Http2Stream -> HeaderList -> IncomingFlowControl -> OutgoingFlowControl -> IO ()+    StreamId -> Http2Stream -> HeaderList -> IncomingFlowControl -> OutgoingFlowControl -> ClientIO ()  -- | Starts a new stream (i.e., one HTTP request + server-pushes). --@@ -305,7 +309,7 @@ -- Note that we currently enforce the 'HTTP2.setEndHeader' but this design -- choice may change in the future. Hence, we recommend you use -- 'HTTP2.setEndHeader' as well.-headers :: Http2Stream -> HeaderList -> FlagSetter -> IO StreamThread+headers :: Http2Stream -> HeaderList -> FlagSetter -> ClientIO StreamThread headers = _headers  -- | Starts a new Http2Client around a frame connection.@@ -333,9 +337,9 @@   -> FallBackFrameHandler   -- ^ Actions to run when a control frame is not yet handled in http2-client   -- lib (e.g., PRIORITY frames).-  -> (Http2Client -> IO a)+  -> (Http2Client -> ClientIO a)   -- ^ Actions to run on the client.-  -> IO a+  -> ClientIO a runHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler mainHandler = do     (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler     withAsync incomingLoop $ \aIncoming -> do@@ -376,7 +380,7 @@   -> FallBackFrameHandler   -- ^ Actions to run when a control frame is not yet handled in http2-client   -- lib (e.g., PRIORITY frames).-  -> IO Http2Client+  -> ClientIO Http2Client newHttp2Client conn encoderBufSize decoderBufSize initSettings goAwayHandler fallbackHandler = do     (incomingLoop, initClient) <- initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler     aIncoming <- async incomingLoop@@ -401,7 +405,7 @@   -> Int   -> GoAwayHandler   -> FallBackFrameHandler-  -> IO (IO (), InitHttp2Client)+  -> ClientIO (ClientIO (), InitHttp2Client) initHttp2Client conn encoderBufSize decoderBufSize goAwayHandler fallbackHandler = do     let controlStream = makeFrameClientStream conn 0     let ackPing = sendPingFrame controlStream HTTP2.setAck@@ -416,9 +420,9 @@                                             fallbackHandler      let baseWindowSize = return HTTP2.defaultInitialWindowSize-    _initIncomingFlowControl <- newIncomingFlowControl dispatchControl baseWindowSize (sendWindowUpdateFrame controlStream)+    _initIncomingFlowControl <- lift $ newIncomingFlowControl dispatchControl baseWindowSize (sendWindowUpdateFrame controlStream)     windowUpdatesChan <- newChan-    _initOutgoingFlowControl <- newOutgoingFlowControl dispatchControl windowUpdatesChan baseWindowSize+    _initOutgoingFlowControl <- lift $ newOutgoingFlowControl dispatchControl windowUpdatesChan baseWindowSize      dispatchHPACK <- newDispatchHPACKIO decoderBufSize     (incomingLoop,endIncomingLoop) <- dispatchLoop conn dispatch dispatchControl windowUpdatesChan _initIncomingFlowControl dispatchHPACK@@ -454,21 +458,21 @@                 pure v      let _initPing dat = do-            handler <- registerPingHandler dispatchControl dat+            handler <- lift $ registerPingHandler dispatchControl dat             sendPingFrame controlStream id dat-            return $ waitPingReply handler+            return $ lift $ waitPingReply handler      let _initSettings settslist = do-            handler <- registerSetSettingsHandler dispatchControl+            handler <- lift $ registerSetSettingsHandler dispatchControl             sendSettingsFrame controlStream id settslist             return $ do-                ret <- waitSetSettingsReply handler+                ret <- lift $ waitSetSettingsReply handler                 modifySettings dispatchControl                     (\(ConnectionSettings cli srv) ->                         (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ()))                 return ret     let _initGoaway err errStr = do-            sId <- readMaxReceivedStreamIdIO dispatch+            sId <- lift $ readMaxReceivedStreamIdIO dispatch             sendGTFOFrame controlStream sId err errStr      let _initPaylodSplitter = settingsPayloadSplitter <$> readSettings dispatchControl@@ -487,7 +491,7 @@   -> Chan (FrameHeader, FramePayload)   -> (Http2Stream -> StreamDefinition a)   -> StreamFSMState-  -> IO (IO a)+  -> ClientIO (ClientIO a) initializeStream conn dispatch control stream windowUpdatesChan getWork initialState = do     let sid = _dispatchStreamId stream     let frameStream = makeFrameClientStream conn sid@@ -535,9 +539,9 @@     -- lock on clientStreamIdMutex will be released.     return $ do         let baseIncomingWindowSize = initialWindowSize . _clientSettings <$> readSettings control-        isfc <- newIncomingFlowControl control baseIncomingWindowSize (sendWindowUpdateFrame frameStream)+        isfc <- lift $ newIncomingFlowControl control baseIncomingWindowSize (sendWindowUpdateFrame frameStream)         let baseOutgoingWindowSize = initialWindowSize . _serverSettings <$> readSettings control-        osfc <- newOutgoingFlowControl control windowUpdatesChan baseOutgoingWindowSize+        osfc <- lift $ newOutgoingFlowControl control windowUpdatesChan baseOutgoingWindowSize         _handleStream streamActions isfc osfc  dispatchLoop@@ -547,7 +551,7 @@   -> Chan (FrameHeader, FramePayload)   -> IncomingFlowControl   -> DispatchHPACK-  -> IO (IO (), IO Bool)+  -> ClientIO (ClientIO (), ClientIO Bool) dispatchLoop conn d dc windowUpdatesChan inFlowControl dh = do     let getNextFrame = next conn     let go = delayException . forever $ do@@ -560,19 +564,19 @@             whenFrame (hasTypeId [FrameWindowUpdate]) frame $ \got -> do                 updateWindowsStep d got             whenFrame (hasTypeId [FramePushPromise, FrameHeaders]) frame $ \got -> do-                let hpackLoop (FinishedWithHeaders curFh sId mkNewHdrs) = 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) = do+                    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)        = do+                    hpackLoop (FailedHeaders curFh sId err)        = lift $ do                         chan <- fmap _streamStateEvents <$> lookupStreamState d sId                         let msg = StreamErrorEvent curFh err                         maybe (return ()) (flip writeChan msg) chan@@ -588,11 +592,11 @@ handleRSTStep   :: Dispatch   -> (FrameHeader, FramePayload)-  -> IO ()+  -> ClientIO () handleRSTStep d (fh, payload) = do     let sid = streamId fh     case payload of-        (RSTStreamFrame err) -> do+        (RSTStreamFrame err) -> lift $ do             chan <- fmap _streamStateEvents <$> lookupStreamState d sid             let msg = StreamErrorEvent fh (HTTP2.fromErrorCodeId err)             maybe (return ()) (flip writeChan msg) chan@@ -603,7 +607,7 @@ dispatchFramesStep   :: (FrameHeader, Either HTTP2Error FramePayload)   -> Dispatch-  -> IO ()+  -> ClientIO () dispatchFramesStep (fh,_) d = do     let sid = streamId fh     -- Remember highest streamId.@@ -612,7 +616,7 @@ finalizeFramesStep   :: (FrameHeader, Either HTTP2Error FramePayload)   -> Dispatch-  -> IO ()+  -> ClientIO () finalizeFramesStep (fh,_) d = do     let sid = streamId fh     -- Remote-close streams that match.@@ -623,7 +627,7 @@   :: Chan (FrameHeader, FramePayload)   -> (FrameHeader, FramePayload)   -> DispatchControl-  -> IO ()+  -> ClientIO () dispatchControlFramesStep windowUpdatesChan controlFrame@(fh, payload) control@(DispatchControl{..}) = do     case payload of         (SettingsFrame settsList)@@ -631,7 +635,7 @@                 atomicModifyIORef' _dispatchControlConnectionSettings                                    (\(ConnectionSettings cli srv) ->                                       (ConnectionSettings cli (HTTP2.updateSettings srv settsList), ()))-                maybe (return ())+                lift $ maybe (return ())                       (_applySettings _dispatchControlHpackEncoder)                       (lookup SettingsHeaderTableSize settsList)                 _dispatchControlAckSettings@@ -662,12 +666,12 @@   :: Dispatch   -> IncomingFlowControl   -> (FrameHeader, FramePayload)-  -> IO ()+  -> ClientIO () creditDataFramesStep d flowControl (fh,payload) = do     -- TODO: error if detect over-run. Current implementation credits     -- everything back. Hence, over-run should never happen.-    _ <- _consumeCredit flowControl (HTTP2.payloadLength fh)-    _addCredit flowControl (HTTP2.payloadLength fh)+    _ <- lift $ _consumeCredit flowControl (HTTP2.payloadLength fh)+    lift $ _addCredit flowControl (HTTP2.payloadLength fh)      -- Write to the interested streams.     let sid = streamId fh@@ -681,7 +685,7 @@ updateWindowsStep   :: Dispatch   -> (FrameHeader, FramePayload)-  -> IO ()+  -> ClientIO () updateWindowsStep d got@(fh,_) = do     let sid = HTTP2.streamId fh     chan <- fmap _streamStateWindowUpdatesChan <$> lookupStreamState d sid@@ -692,7 +696,7 @@   | OpenPushPromise !StreamId !StreamId  data HPACKStepResult =-    WaitContinuation !((FrameHeader, Either HTTP2Error FramePayload) -> IO HPACKStepResult)+    WaitContinuation !((FrameHeader, Either HTTP2Error FramePayload) -> ClientIO HPACKStepResult)   | FailedHeaders !FrameHeader !StreamId ErrorCode   | FinishedWithHeaders !FrameHeader !StreamId (IO HeaderList)   | FinishedWithPushPromise !FrameHeader !StreamId !StreamId (IO HeaderList)@@ -745,7 +749,7 @@   :: DispatchControl   -> IO Int   -- ^ Action to get the base window size.-  -> (WindowSize -> IO())+  -> (WindowSize -> ClientIO ())   -- ^ Action to send the window update to the server.   -> IO IncomingFlowControl newIncomingFlowControl control getBase doSendUpdate = do@@ -766,7 +770,7 @@             let shouldUpdate = transferred > 0              _addCredit (negate transferred)-            _ <- _consumeCredit (negate transferred)+            _ <- lift $ _consumeCredit (negate transferred)             when shouldUpdate (doSendUpdate transferred)              return shouldUpdate@@ -784,7 +788,7 @@     let receive n = atomicModifyIORef' credit (\c -> (c + n, ()))     let withdraw 0 = return 0         withdraw n = do-            base <- getBase+            base <- lift getBase             got <- atomicModifyIORef' credit (\c ->                     if base + c >= n                     then (c - n, n)@@ -821,9 +825,9 @@   -> HeaderList   -> PayloadSplitter   -> (FrameFlags -> FrameFlags)-  -> IO StreamThread+  -> ClientIO StreamThread sendHeaders s enc hdrs blockSplitter flagmod = do-    _sendFrames s mkFrames+    _sendFrames s (lift mkFrames)     return CST   where     mkFrames = do@@ -873,9 +877,9 @@ -- streams, although no priority queue exists in `http2-client` so far). -- -- Please refer to '_sendDataChunk' and '_withdrawCredit' as well.-sendData :: Http2Client -> Http2Stream -> FlagSetter -> ByteString -> IO ()+sendData :: Http2Client -> Http2Stream -> FlagSetter -> ByteString -> ClientIO () sendData conn stream flagmod dat = do-    splitter <- _payloadSplitter conn+    splitter <- lift  $ _payloadSplitter conn     let chunks = splitter dat     let pairs  = reverse $ zip (flagmod : repeat id) (reverse chunks)     when (null chunks) $ _sendDataChunk stream flagmod ""@@ -883,17 +887,17 @@  sendDataFrame   :: Http2FrameClientStream-  -> (FrameFlags -> FrameFlags) -> ByteString -> IO ()+  -> (FrameFlags -> FrameFlags) -> ByteString -> ClientIO () sendDataFrame s flagmod dat = do     sendOne s flagmod (DataFrame dat) -sendResetFrame :: Http2FrameClientStream -> ErrorCodeId -> IO ()+sendResetFrame :: Http2FrameClientStream -> ErrorCodeId -> ClientIO () sendResetFrame s err = do     sendOne s id (RSTStreamFrame err)  sendGTFOFrame   :: Http2FrameClientStream-     -> StreamId -> ErrorCodeId -> ByteString -> IO ()+     -> StreamId -> ErrorCodeId -> ByteString -> ClientIO () sendGTFOFrame s lastStreamId err errStr = do     sendOne s id (GoAwayFrame lastStreamId err errStr) @@ -904,7 +908,7 @@   :: Http2FrameClientStream   -> (FrameFlags -> FrameFlags)   -> ByteString-  -> IO ()+  -> ClientIO () sendPingFrame s flags dat   | _getStreamId s /= 0        =         rfcError "PING frames are not associated with any individual stream."@@ -913,7 +917,7 @@   | otherwise                  = sendOne s flags (PingFrame dat)  sendWindowUpdateFrame-  :: Http2FrameClientStream -> WindowSize -> IO ()+  :: Http2FrameClientStream -> WindowSize -> ClientIO () sendWindowUpdateFrame s amount = do     let payload = WindowUpdateFrame amount     sendOne s id payload@@ -921,7 +925,7 @@  sendSettingsFrame   :: Http2FrameClientStream-     -> (FrameFlags -> FrameFlags) -> SettingsList -> IO ()+     -> (FrameFlags -> FrameFlags) -> SettingsList -> ClientIO () sendSettingsFrame s flags setts   | _getStreamId s /= 0        =         rfcError "The stream identifier for a SETTINGS frame MUST be zero (0x0)."@@ -930,7 +934,7 @@     sendOne s flags payload     return () -sendPriorityFrame :: Http2FrameClientStream -> Priority -> IO ()+sendPriorityFrame :: Http2FrameClientStream -> Priority -> ClientIO () sendPriorityFrame s p = do     let payload = PriorityFrame p     sendOne s id payload@@ -949,9 +953,9 @@ -- and future design will be to inline the various loop-processes for -- dispatchFrames and GoAwayHandlers in a same thread (e.g., using -- pipe/conduit to retain composability).-delayException :: IO a -> IO a+delayException :: ClientIO a -> ClientIO a delayException act = act `catch` slowdown   where-    slowdown :: SomeException -> IO a+    slowdown :: SomeException -> ClientIO a     slowdown e = threadDelay 50000 >> throwIO e 

@@ -6,21 +6,23 @@  , whenFrame  , whenFrameElse  -- re-exports- , module Control.Concurrent.Chan+ , module Control.Concurrent.Chan.Lifted  ) where -import           Control.Concurrent.Chan (Chan, readChan, newChan, writeChan)-import           Control.Exception (Exception, throwIO)+import           Control.Concurrent.Chan.Lifted (Chan, readChan, newChan, writeChan)+import           Control.Exception.Lifted (Exception, throwIO) import           Network.HTTP2 (StreamId, FrameHeader, FramePayload, FrameTypeId, framePayloadToFrameTypeId, streamId) +import           Network.HTTP2.Client.Exceptions+ type FramesChan e = Chan (FrameHeader, Either e FramePayload)  whenFrame   :: Exception e   => (FrameHeader -> FramePayload -> Bool)   -> (FrameHeader, Either e FramePayload)-  -> ((FrameHeader, FramePayload) -> IO ())-  -> IO ()+  -> ((FrameHeader, FramePayload) -> ClientIO ())+  -> ClientIO () whenFrame test frame handle = do     whenFrameElse test frame handle (const $ pure ()) @@ -28,9 +30,9 @@   :: Exception e   => (FrameHeader -> FramePayload -> Bool)   -> (FrameHeader, Either e FramePayload)-  -> ((FrameHeader, FramePayload) -> IO a)-  -> ((FrameHeader, FramePayload) -> IO a)-  -> IO a+  -> ((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

@@ -1,12 +1,14 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE BangPatterns #-} module Network.HTTP2.Client.Dispatch where  import           Control.Exception (throwIO)+import           Control.Monad.Base (MonadBase, liftBase) import           Data.ByteString (ByteString) import qualified Data.ByteString.Internal as ByteString import           Foreign.Marshal.Alloc (mallocBytes, finalizerFree) import           Foreign.ForeignPtr (newForeignPtr)-import           Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import           Data.IORef.Lifted (IORef, atomicModifyIORef', newIORef, readIORef) import           Data.IntMap (IntMap) import qualified Data.IntMap as IntMap import           GHC.Exception (Exception)@@ -15,11 +17,12 @@ import           Network.HTTP2 as HTTP2  import           Network.HTTP2.Client.Channels+import           Network.HTTP2.Client.Exceptions  type DispatchChan = FramesChan HTTP2Error  -- | A fallback handler for frames.-type FallBackFrameHandler = (FrameHeader, FramePayload) -> IO ()+type FallBackFrameHandler = (FrameHeader, FramePayload) -> ClientIO ()  -- | Default FallBackFrameHandler that ignores frames. ignoreFallbackHandler :: FallBackFrameHandler@@ -31,7 +34,7 @@ instance Exception RemoteSentGoAwayFrame  -- | A Handler for exceptional circumstances.-type GoAwayHandler = RemoteSentGoAwayFrame -> IO ()+type GoAwayHandler = RemoteSentGoAwayFrame -> ClientIO ()  -- | Default GoAwayHandler throws a 'RemoteSentGoAwayFrame' in the current -- thread.@@ -44,7 +47,7 @@ -- server will likely close the connection which will lead to TCP errors as -- well. defaultGoAwayHandler :: GoAwayHandler-defaultGoAwayHandler = throwIO+defaultGoAwayHandler = lift . throwIO  data StreamFSMState =     Idle@@ -72,22 +75,22 @@   , _dispatchCurrentStreams :: !(IORef (IntMap StreamState))   } -newDispatchIO :: IO Dispatch+newDispatchIO :: MonadBase IO m => m Dispatch newDispatchIO = Dispatch <$> newIORef 0 <*> newIORef (IntMap.empty) -readMaxReceivedStreamIdIO :: Dispatch -> IO StreamId+readMaxReceivedStreamIdIO :: MonadBase IO m => Dispatch -> m StreamId readMaxReceivedStreamIdIO = readIORef . _dispatchMaxStreamId -registerStream :: Dispatch -> StreamId -> StreamState -> IO ()+registerStream :: MonadBase IO m => Dispatch -> StreamId -> StreamState -> m () registerStream d sid st =     atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->       let v = (IntMap.insert sid st xs) in (v, ()) -lookupStreamState :: Dispatch -> StreamId -> IO (Maybe StreamState)+lookupStreamState :: MonadBase IO m => Dispatch -> StreamId -> m (Maybe StreamState) lookupStreamState d sid =     IntMap.lookup sid <$> readIORef (_dispatchCurrentStreams d) -closeLocalStream :: Dispatch -> StreamId -> IO ()+closeLocalStream :: MonadBase IO m => Dispatch -> StreamId -> m () closeLocalStream d sid =     atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->       let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ())@@ -98,7 +101,7 @@         Closed           -> Nothing         _ -> Just $ st { _streamStateFSMState = HalfClosedLocal } -closeRemoteStream :: Dispatch -> StreamId -> IO ()+closeRemoteStream :: MonadBase IO m => Dispatch -> StreamId -> m () closeRemoteStream d sid =     atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->       let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ())@@ -109,7 +112,7 @@         Closed           -> Nothing         _ -> Just $ st { _streamStateFSMState = HalfClosedRemote } -closeReleaseStream :: Dispatch -> StreamId -> IO ()+closeReleaseStream :: MonadBase IO m => Dispatch -> StreamId -> m () closeReleaseStream d sid =     atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs ->       let v = (IntMap.delete sid xs) in (v, ())@@ -126,24 +129,24 @@  data PingHandler = PingHandler !(Chan (FrameHeader, FramePayload)) -newPingHandler :: IO PingHandler+newPingHandler :: MonadBase IO m => m PingHandler newPingHandler = PingHandler <$> newChan -notifyPingHandler :: (FrameHeader, FramePayload) -> PingHandler -> IO ()+notifyPingHandler :: MonadBase IO m => (FrameHeader, FramePayload) -> PingHandler -> m () notifyPingHandler dat (PingHandler c) = writeChan c dat -waitPingReply :: PingHandler -> IO (FrameHeader, FramePayload)+waitPingReply :: MonadBase IO m => PingHandler -> m (FrameHeader, FramePayload) waitPingReply (PingHandler c) = readChan c  data SetSettingsHandler = SetSettingsHandler !(Chan (FrameHeader, FramePayload)) -newSetSettingsHandler :: IO SetSettingsHandler+newSetSettingsHandler :: MonadBase IO m => m SetSettingsHandler newSetSettingsHandler = SetSettingsHandler <$> newChan -notifySetSettingsHandler :: (FrameHeader, FramePayload) -> SetSettingsHandler -> IO ()+notifySetSettingsHandler :: MonadBase IO m => (FrameHeader, FramePayload) -> SetSettingsHandler -> m () notifySetSettingsHandler dat (SetSettingsHandler c) = writeChan c dat -waitSetSettingsReply :: SetSettingsHandler -> IO (FrameHeader, FramePayload)+waitSetSettingsReply :: MonadBase IO m => SetSettingsHandler -> m (FrameHeader, FramePayload) waitSetSettingsReply (SetSettingsHandler c) = readChan c  registerPingHandler :: DispatchControl -> ByteString -> IO PingHandler@@ -153,7 +156,7 @@         ((dat,handler):xs, ()))     return handler -lookupAndReleasePingHandler :: DispatchControl -> ByteString -> IO (Maybe PingHandler)+lookupAndReleasePingHandler :: MonadBase IO m => DispatchControl -> ByteString -> m (Maybe PingHandler) lookupAndReleasePingHandler dc dat =     atomicModifyIORef' (_dispatchControlPingHandlers dc) f   where@@ -162,14 +165,14 @@     -- storing the handlers).     f xs = (filter (\x -> dat /= fst x) xs, lookup dat xs) -registerSetSettingsHandler :: DispatchControl -> IO SetSettingsHandler+registerSetSettingsHandler :: MonadBase IO m => DispatchControl -> m SetSettingsHandler registerSetSettingsHandler dc = do     handler <- newSetSettingsHandler     atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) (\xs ->         (handler:xs, ()))     return handler -lookupAndReleaseSetSettingsHandler :: DispatchControl -> IO (Maybe SetSettingsHandler)+lookupAndReleaseSetSettingsHandler :: MonadBase IO m => DispatchControl -> m (Maybe SetSettingsHandler) lookupAndReleaseSetSettingsHandler dc =     atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) f   where@@ -179,8 +182,8 @@ data DispatchControl = DispatchControl {     _dispatchControlConnectionSettings  :: !(IORef ConnectionSettings)   , _dispatchControlHpackEncoder        :: !HpackEncoderContext-  , _dispatchControlAckPing             :: !(ByteString -> IO ())-  , _dispatchControlAckSettings         :: !(IO ())+  , _dispatchControlAckPing             :: !(ByteString -> ClientIO ())+  , _dispatchControlAckSettings         :: !(ClientIO ())   , _dispatchControlOnGoAway            :: !GoAwayHandler   , _dispatchControlOnFallback          :: !FallBackFrameHandler   , _dispatchControlPingHandlers        :: !(IORef [(ByteString, PingHandler)])@@ -188,12 +191,13 @@   }  newDispatchControlIO-  :: Size-  -> (ByteString -> IO ())-  -> (IO ())+  :: MonadBase IO m+  => Size+  -> (ByteString -> ClientIO ())+  -> (ClientIO ())   -> GoAwayHandler   -> FallBackFrameHandler-  -> IO DispatchControl+  -> m DispatchControl newDispatchControlIO encoderBufSize ackPing ackSetts onGoAway onFallback =     DispatchControl <$> newIORef defaultConnectionSettings                     <*> newHpackEncoderContext encoderBufSize@@ -204,14 +208,14 @@                     <*> newIORef []                     <*> newIORef [] -newHpackEncoderContext :: Size -> IO HpackEncoderContext-newHpackEncoderContext encoderBufSize = do+newHpackEncoderContext :: MonadBase IO m => Size -> m HpackEncoderContext+newHpackEncoderContext encoderBufSize = liftBase $ do     let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True })     dt <- HPACK.newDynamicTableForEncoding HPACK.defaultDynamicTableSize     buf <- mallocBytes encoderBufSize     ptr <- newForeignPtr finalizerFree buf     return $ HpackEncoderContext-            (encoder strategy dt buf ptr)+            (\hdrs -> encoder strategy dt buf ptr hdrs)             (\n -> HPACK.setLimitForEncoding n dt)   where     encoder strategy dt buf ptr hdrs = do@@ -221,10 +225,10 @@             ([],len) -> pure $ ByteString.fromForeignPtr ptr 0 len             (_,_)  -> throwIO HPACK.BufferOverrun -readSettings :: DispatchControl -> IO ConnectionSettings+readSettings :: MonadBase IO m => DispatchControl -> m ConnectionSettings readSettings = readIORef . _dispatchControlConnectionSettings -modifySettings :: DispatchControl -> (ConnectionSettings -> (ConnectionSettings, a)) -> IO a+modifySettings :: MonadBase IO m => DispatchControl -> (ConnectionSettings -> (ConnectionSettings, a)) -> m a modifySettings d = atomicModifyIORef' (_dispatchControlConnectionSettings d)  -- | Helper to carry around the HPACK encoder for outgoing header blocks..@@ -237,8 +241,8 @@     _dispatchHPACKDynamicTable          :: !DynamicTable   } -newDispatchHPACKIO :: Size -> IO DispatchHPACK-newDispatchHPACKIO decoderBufSize =+newDispatchHPACKIO :: MonadBase IO m => Size -> m DispatchHPACK+newDispatchHPACKIO decoderBufSize = liftBase $     DispatchHPACK <$> newDecoder   where     newDecoder = newDynamicTableForDecoding@@ -250,7 +254,7 @@   , _dispatchStreamReadEvents :: !(Chan StreamEvent)   } -newDispatchStreamIO :: StreamId -> IO DispatchStream-newDispatchStreamIO sid =+newDispatchStreamIO :: MonadBase IO m => StreamId -> m DispatchStream+newDispatchStreamIO sid = liftBase $     DispatchStream <$> pure sid                    <*> newChan

@@ -0,0 +1,22 @@++module Network.HTTP2.Client.Exceptions (+    ClientIO+  , ClientError(..)+  , runClientIO+  , module Control.Monad.Except+  ) where++import           Control.Exception (Exception)+import           Control.Monad.Except (ExceptT, runExceptT, throwError, lift)++type ClientIO = ExceptT ClientError IO++runClientIO :: ClientIO a -> IO (Either ClientError a)+runClientIO = runExceptT++-- | A set of errors as observed from the client.+data ClientError = EarlyEndOfStream+  -- ^ We received a TCP end-of-stream and there will be no further read-IOs possible on the connection.+  deriving (Show,Ord,Eq)+instance Exception ClientError -- For people who want to rethrow using Control.Exception+

@@ -1,3 +1,4 @@+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards  #-} {-# LANGUAGE RankNTypes  #-}@@ -16,14 +17,16 @@     ) where  import           Control.DeepSeq (deepseq)-import           Control.Exception (bracket)-import           Control.Concurrent.MVar (newMVar, takeMVar, putMVar)-import           Control.Monad ((>=>), void)+import           Control.Exception.Lifted (bracket)+import           Control.Concurrent.MVar.Lifted (newMVar, takeMVar, putMVar)+import           Control.Monad ((>=>), void, when)+import qualified Data.ByteString as ByteString import           Network.HTTP2 (FrameHeader(..), FrameFlags, FramePayload, HTTP2Error, encodeInfo, decodeFramePayload) import qualified Network.HTTP2 as HTTP2 import           Network.Socket (HostName, PortNumber) import qualified Network.TLS as TLS +import           Network.HTTP2.Client.Exceptions import           Network.HTTP2.Client.RawConnection  data Http2FrameConnection = Http2FrameConnection {@@ -31,12 +34,12 @@   -- ^ Starts a new client stream.   , _serverStream     :: Http2ServerStream   -- ^ Receives frames from a server.-  , _closeConnection  :: IO ()+  , _closeConnection  :: ClientIO ()   -- ^ Function that will close the network connection.   }  -- | Closes the Http2FrameConnection abruptly.-closeConnection :: Http2FrameConnection -> IO ()+closeConnection :: Http2FrameConnection -> ClientIO () closeConnection = _closeConnection  -- | Creates a client stream.@@ -46,7 +49,7 @@ makeFrameClientStream = _makeFrameClientStream  data Http2FrameClientStream = Http2FrameClientStream {-    _sendFrames :: IO [(FrameFlags -> FrameFlags, FramePayload)] -> IO ()+    _sendFrames :: ClientIO [(FrameFlags -> FrameFlags, FramePayload)] -> ClientIO ()   -- ^ Sends a frame to the server.   -- The first argument is a FrameFlags modifier (e.g., to sed the   -- end-of-stream flag).@@ -54,25 +57,25 @@   }  -- | Sends a frame to the server.-sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> IO ()+sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> ClientIO () sendOne client f payload = _sendFrames client (pure [(f, payload)])  -- | Sends multiple back-to-back frames to the server.-sendBackToBack :: Http2FrameClientStream -> [(FrameFlags -> FrameFlags, FramePayload)] -> IO ()+sendBackToBack :: Http2FrameClientStream -> [(FrameFlags -> FrameFlags, FramePayload)] -> ClientIO () sendBackToBack client payloads = _sendFrames client (pure payloads)  data Http2ServerStream = Http2ServerStream {-    _nextHeaderAndFrame :: IO (FrameHeader, Either HTTP2Error FramePayload)+    _nextHeaderAndFrame :: ClientIO (FrameHeader, Either HTTP2Error FramePayload)   }  -- | Waits for the next frame from the server.-next :: Http2FrameConnection -> IO (FrameHeader, Either HTTP2Error FramePayload)+next :: Http2FrameConnection -> ClientIO (FrameHeader, Either HTTP2Error FramePayload) next = _nextHeaderAndFrame . _serverStream  -- | Adds framing around a 'RawHttp2Connection'. frameHttp2RawConnection   :: RawHttp2Connection-  -> IO Http2FrameConnection+  -> ClientIO Http2FrameConnection frameHttp2RawConnection http2conn = do     -- Prepare a local mutex, this mutex should never escape the     -- function's scope. Else it might lead to bugs (e.g.,@@ -96,13 +99,16 @@              in Http2FrameClientStream putFrames streamID          nextServerFrameChunk = Http2ServerStream $ do-            (fTy, fh@FrameHeader{..}) <- HTTP2.decodeFrameHeader <$> _nextRaw http2conn 9+            b9 <- _nextRaw http2conn 9+            when (ByteString.length b9 /= 9) $ throwError $ EarlyEndOfStream+            let (fTy, fh@FrameHeader{..}) = HTTP2.decodeFrameHeader b9             let decoder = decodeFramePayload fTy+            buf <- _nextRaw http2conn payloadLength+            when (ByteString.length buf /= payloadLength) $ throwError $ EarlyEndOfStream             -- TODO: consider splitting the iteration here to give a chance to             -- _not_ decode the frame, or consider lazyness enough.-            let getNextFrame = decoder fh <$> _nextRaw http2conn payloadLength-            nf <- getNextFrame-            return (fh, nf)+            let nf = decoder fh buf+            pure (fh, nf)          gtfo = _close http2conn @@ -112,6 +118,6 @@ newHttp2FrameConnection :: HostName                         -> PortNumber                         -> Maybe TLS.ClientParams-                        -> IO Http2FrameConnection+                        -> ClientIO Http2FrameConnection newHttp2FrameConnection host port params = do     frameHttp2RawConnection =<< newRawHttp2Connection host port params

@@ -25,10 +25,11 @@ import qualified Network.HPACK as HPACK import           Data.ByteString (ByteString) import qualified Data.ByteString as ByteString-import           Control.Concurrent (threadDelay)-import           Control.Concurrent.Async (race)+import           Control.Concurrent.Lifted (threadDelay)+import           Control.Concurrent.Async.Lifted (race)  import Network.HTTP2.Client+import Network.HTTP2.Client.Exceptions  -- | Opaque type to express an action which timed out. data TimedOut = TimedOut@@ -45,12 +46,12 @@      -- ^ timeout in microseconds      -> ByteString      -- ^ 8-bytes message to uniquely identify the reply-     -> IO PingReply+     -> ClientIO PingReply ping conn timeout msg = do-    t0 <- getCurrentTime+    t0 <- lift $ getCurrentTime     waitPing <- _ping conn msg     pingReply <- race (threadDelay timeout >> return TimedOut) waitPing-    t1 <- getCurrentTime+    t1 <- lift $ getCurrentTime     return $ (t0, t1, pingReply)  -- | Result containing the unpacked headers and all frames received in on a@@ -88,7 +89,7 @@        -- ^ The corresponding HTTP stream.        -> OutgoingFlowControl        -- ^ The flow control for this stream.-       -> IO ()+       -> ClientIO () upload "" flagmod conn _ stream _ = do     sendData conn stream flagmod "" upload dat flagmod conn connectionFlowControl stream streamFlowControl = do@@ -98,7 +99,7 @@     got       <- _withdrawCredit connectionFlowControl gotStream     -- Recredit the stream flow control with the excedent we cannot spend on     -- the connection.-    _receiveCredit streamFlowControl (gotStream - got)+    lift $ _receiveCredit streamFlowControl (gotStream - got)      let uploadChunks flagMod =             sendData conn stream flagMod (ByteString.take got dat)@@ -118,7 +119,7 @@ waitStream :: Http2Stream            -> IncomingFlowControl            -> PushPromiseHandler-           -> IO StreamResult+           -> ClientIO StreamResult waitStream stream streamFlowControl ppHandler = do     ev <- _waitEvent stream     case ev of@@ -141,8 +142,8 @@                 | HTTP2.testEndStream (HTTP2.flags fh) ->                     return ((Right x):xs, Nothing)                 | otherwise                            -> do-                    _ <- _consumeCredit streamFlowControl (HTTP2.payloadLength fh)-                    _addCredit streamFlowControl (HTTP2.payloadLength fh)+                    _ <- lift $ _consumeCredit streamFlowControl (HTTP2.payloadLength fh)+                    lift $ _addCredit streamFlowControl (HTTP2.payloadLength fh)                     _ <- _updateWindow $ streamFlowControl                     waitDataFrames ((Right x):xs)             StreamPushPromiseEvent _ ppSid ppHdrs -> do

@@ -9,8 +9,8 @@     ) where  import           Control.Monad (forever, when)-import           Control.Concurrent.Async (Async, async, cancel, pollSTM)-import           Control.Concurrent.STM (STM, atomically, retry, throwSTM)+import           Control.Concurrent.Async.Lifted (Async, async, cancel, pollSTM)+import           Control.Concurrent.STM (STM, atomically, check, orElse, retry, throwSTM) import           Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO, readTVar, writeTVar) import           Data.ByteString (ByteString) import qualified Data.ByteString as ByteString@@ -21,13 +21,18 @@ import           Network.Socket.ByteString import qualified Network.TLS as TLS +import           Network.HTTP2.Client.Exceptions+ -- TODO: catch connection errrors data RawHttp2Connection = RawHttp2Connection {-    _sendRaw :: [ByteString] -> IO ()+    _sendRaw :: [ByteString] -> ClientIO ()   -- ^ Function to send raw data to the server.-  , _nextRaw :: Int -> IO ByteString+  , _nextRaw :: Int -> ClientIO ByteString   -- ^ Function to block reading a datachunk of a given size from the server.-  , _close   :: IO ()+  -- An empty chunk when asking for a length larger than 0 means the underlying+  -- network session is closed. A compliant HTP2 server should have sent a+  -- GOAWAY before such an event occurs.+  , _close   :: ClientIO ()   }  -- | Initiates a RawHttp2Connection with a server.@@ -40,15 +45,17 @@                       -> Maybe TLS.ClientParams                       -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is                       -- overwritten to always return ["h2", "h2-17"].-                      -> IO RawHttp2Connection+                      -> ClientIO RawHttp2Connection newRawHttp2Connection host port mparams = do     -- Connects to TCP.     let hints = defaultHints { addrFlags = [AI_NUMERICSERV], addrSocketType = Stream }-    addr:_ <- getAddrInfo (Just hints) (Just host) (Just $ show port)-    skt <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)-    setSocketOption skt NoDelay 1-    connect skt (addrAddress addr)-    newRawHttp2ConnectionSocket skt mparams+    rSkt <- lift $ do+        addr:_ <- getAddrInfo (Just hints) (Just host) (Just $ show port)+        skt <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+        setSocketOption skt NoDelay 1+        connect skt (addrAddress addr)+        pure skt+    newRawHttp2ConnectionSocket rSkt mparams  -- | Initiates a RawHttp2Connection with a server over a connected socket. --@@ -60,10 +67,10 @@   -> Maybe TLS.ClientParams   -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is   -- overwritten to always return ["h2", "h2-17"].-  -> IO RawHttp2Connection+  -> ClientIO RawHttp2Connection newRawHttp2ConnectionSocket skt mparams = do     -- Prepare structure with abstract API.-    conn <- maybe (plainTextRaw skt) (tlsRaw skt) mparams+    conn <- lift $ maybe (plainTextRaw skt) (tlsRaw skt) mparams      -- Initializes the HTTP2 stream.     _sendRaw conn [HTTP2.connectionPreface]@@ -74,8 +81,8 @@ plainTextRaw skt = do     (b,putRaw) <- startWriteWorker (sendMany skt)     (a,getRaw) <- startReadWorker (recv skt)-    let doClose = cancel a >> cancel b >> close skt-    return $ RawHttp2Connection (atomically . putRaw) (atomically . getRaw) doClose+    let doClose = lift $ cancel a >> cancel b >> close skt+    return $ RawHttp2Connection (lift . atomically . putRaw) (lift . atomically . getRaw) doClose  tlsRaw :: Socket -> TLS.ClientParams -> IO RawHttp2Connection tlsRaw skt params = do@@ -85,9 +92,9 @@      (b,putRaw) <- startWriteWorker (TLS.sendData tlsContext . fromChunks)     (a,getRaw) <- startReadWorker (const $ TLS.recvData tlsContext)-    let doClose       = cancel a >> cancel b >> TLS.bye tlsContext >> TLS.contextClose tlsContext+    let doClose       = lift $ cancel a >> cancel b >> TLS.bye tlsContext >> TLS.contextClose tlsContext -    return $ RawHttp2Connection (atomically . putRaw) (atomically . getRaw) doClose+    return $ RawHttp2Connection (lift . atomically . putRaw) (lift . atomically . getRaw) doClose   where     modifyParams prms = prms {         TLS.clientHooks = (TLS.clientHooks prms) {@@ -117,14 +124,23 @@   :: (Int -> IO ByteString)   -> IO (Async (), (Int -> STM ByteString)) startReadWorker get = do+    remoteClosed <- newTVarIO False+    let onEof = atomically $ writeTVar remoteClosed True+    let emptyByteStringOnEof = readTVar remoteClosed >>= check >> pure ""+     buf <- newTVarIO ""-    a <- async $ readWorkerLoop buf get-    return $ (a, getRawWorker a buf)+    a <- async $ readWorkerLoop buf get onEof -readWorkerLoop :: TVar ByteString -> (Int -> IO ByteString) -> IO ()-readWorkerLoop buf next = forever $ do-    dat <- next 4096-    atomically $ modifyTVar' buf (\bs -> (bs <> dat))+    return $ (a, \len -> getRawWorker a buf len `orElse` emptyByteStringOnEof)++readWorkerLoop :: TVar ByteString -> (Int -> IO ByteString) -> IO () -> IO ()+readWorkerLoop buf next onEof = go+  where+    go = do+        dat <- next 4096+        if ByteString.null dat+        then onEof+        else atomically (modifyTVar' buf (\bs -> (bs <> dat))) >> go  getRawWorker :: Async () -> TVar ByteString -> Int -> STM ByteString getRawWorker a buf amount = do