diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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! 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.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 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,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 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 @@ -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 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,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 diff --git a/src/Network/HTTP2/Client/Exceptions.hs b/src/Network/HTTP2/Client/Exceptions.hs new file mode 100644 --- /dev/null +++ b/src/Network/HTTP2/Client/Exceptions.hs @@ -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 + diff --git a/src/Network/HTTP2/Client/FrameConnection.hs b/src/Network/HTTP2/Client/FrameConnection.hs --- a/src/Network/HTTP2/Client/FrameConnection.hs +++ b/src/Network/HTTP2/Client/FrameConnection.hs @@ -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 diff --git a/src/Network/HTTP2/Client/Helpers.hs b/src/Network/HTTP2/Client/Helpers.hs --- a/src/Network/HTTP2/Client/Helpers.hs +++ b/src/Network/HTTP2/Client/Helpers.hs @@ -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 diff --git a/src/Network/HTTP2/Client/RawConnection.hs b/src/Network/HTTP2/Client/RawConnection.hs --- a/src/Network/HTTP2/Client/RawConnection.hs +++ b/src/Network/HTTP2/Client/RawConnection.hs @@ -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