diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -36,12 +36,16 @@ data Verbosity = Verbose | NonVerbose deriving Show +data UseTLS = UseTLS | PlainText + deriving Show + data QueryArgs = QueryArgs { _host :: !HostName , _port :: !PortNumber , _verb :: !Verb , _path :: !Path , _extraHeaders :: ![(ByteString, ByteString)] + , _extraTrailers :: ![(ByteString, ByteString)] , _postData :: !(Maybe PostData) , _interPingDelay :: !Int , _pingTimeout :: !Int @@ -60,6 +64,7 @@ , _decoderBufsize :: !Int , _downloadPrefix :: !FilePath , _verboseDebug :: !Verbosity + , _useTLS :: !UseTLS } deriving Show clientArgs :: Parser QueryArgs @@ -70,6 +75,7 @@ <*> verb <*> path <*> extraHeaders + <*> extraTrailers <*> postData <*> milliseconds "inter-ping-delay-ms" 0 <*> milliseconds "ping-timeout-ms" 5000 @@ -88,6 +94,7 @@ <*> decoderBufSize <*> downloadPrefix <*> verboseDebug + <*> useTLS where bstrOption = fmap ByteString.pack . strOption milliseconds what base = fmap (*1000) $ option auto (long what <> value base) @@ -98,6 +105,7 @@ port = option auto (long "port" <> value 443) verb = bstrOption (long "verb" <> value "GET") extraHeaders = many (fmap keyval $ bstrOption (short 'H')) + extraTrailers = many (fmap keyval $ bstrOption (short 'T')) postData = optional (fmap postDataForString (strOption (short 'd'))) concurrency = option auto (long "max-concurrency" <> value 100) allowPush = flag PushEnabled PushDisabled (long "disable-server-push") @@ -112,6 +120,7 @@ decoderBufSize = option auto (long "hpack-decoder-buffer-size" <> value 4096) downloadPrefix = strOption (long "push-files-prefix" <> value ":stdout-pp") verboseDebug = flag NonVerbose Verbose (long "verbose") + useTLS = flag UseTLS PlainText (long "plain-text") main :: IO () main = execParser opts >>= client @@ -131,25 +140,41 @@ -- -- Note that this implementation reads the body from files and into memory -- (better for testing concurrency, worse for large uploads). - (headersFlags, dataPostFunction) <- case _postData of - (Just (PostBytestring dataPayload)) -> - return $ (id, upload dataPayload) - (Just (PostFileContent filepath)) -> do + (headersFlags, dataPostFunction) <- case (_extraTrailers, _postData) of + ([], (Just (PostBytestring dataPayload))) -> + return $ (id, upload dataPayload HTTP2.setEndStream) + ([], (Just (PostFileContent filepath))) -> do dataPayload <- ByteString.readFile filepath - return $ (id, upload dataPayload) - Nothing -> + return $ (id, upload dataPayload HTTP2.setEndStream) + ([], Nothing) -> return $ (HTTP2.setEndStream, \_ _ _ _ -> return ()) + (_, (Just (PostBytestring dataPayload))) -> + return $ (id, (\c ofc s sofc -> do + upload dataPayload id c ofc s sofc + trailers s _extraTrailers HTTP2.setEndStream + )) + (_, (Just (PostFileContent filepath))) -> do + dataPayload <- ByteString.readFile filepath + return $ (id, (\c ofc s sofc -> do + upload dataPayload id c ofc s sofc + trailers s _extraTrailers HTTP2.setEndStream + )) + (_, Nothing) -> + return $ (id, (\_ _ s _ -> do + trailers s _extraTrailers HTTP2.setEndStream + )) let headersPairs = [ (":method", _verb) , (":scheme", "https") , (":path", _path) , (":authority", ByteString.pack _host) - ] <> _extraHeaders + ] <> _extraHeaders <> [("Trailer", ByteString.unwords $ fmap fst _extraTrailers)] + let ppHandler n idx _ stream ppHdrs streamFlowControl _ = void $ forkIO $ do let pushpath = fromMaybe "unspecified-path" (lookup ":path" ppHdrs) timePrint ("push stream started" :: String, pushpath) - ret <- fromStreamResult <$> waitStream stream streamFlowControl + ret <- fromStreamResult <$> waitStream stream streamFlowControl (ppHandler n idx) either (\e -> timePrint e) (dump PushPromiseFile pushpath n idx _downloadPrefix) ret timePrint ("push stream ended" :: String) @@ -162,14 +187,20 @@ ] timePrint conf - frameConn <- newHttp2FrameConnection _host _port (Just tlsParams) + let tlsSetting = case _useTLS of + UseTLS -> + Just tlsParams + PlainText -> + Nothing + frameConn <- newHttp2FrameConnection _host _port tlsSetting let wrappedFrameConn = frameConn { _makeFrameClientStream = \sid -> let frameClient = (_makeFrameClientStream frameConn) sid in frameClient { - _sendFrames = \xs -> do + _sendFrames = \mkFrames -> do + xs <- mkFrames print $ (">>> "::String, _getStreamId frameClient, map snd xs) - _sendFrames frameClient xs + _sendFrames frameClient (pure xs) } , _serverStream = let @@ -205,16 +236,16 @@ let go 0 idx = timePrint $ "done worker: " <> show idx go n idx = do _ <- (withHttp2Stream conn $ \stream -> - let initStream = do - _ <- async $ onPushPromise stream (ppHandler n idx) + let initStream = headers stream headersPairs headersFlags handler streamINFlowControl streamOUTFlowControl = do timePrint $ "stream started " <> show (idx, n) - _ <- async $ dataPostFunction conn - (_outgoingFlowControl conn) - stream - streamOUTFlowControl - ret <- fromStreamResult <$> waitStream stream streamINFlowControl + _ <- async $ do + dataPostFunction conn + (_outgoingFlowControl conn) + stream + streamOUTFlowControl + ret <- fromStreamResult <$> waitStream stream streamINFlowControl (ppHandler n idx) either (\e -> timePrint e) (dump MainFile _path n idx _downloadPrefix) ret timePrint $ "stream ended " <> show (idx, n) in StreamDefinition initStream handler) @@ -243,22 +274,28 @@ data DumpType = MainFile | PushPromiseFile dump :: DumpType -> Path -> Int -> Int -> FilePath -> StreamResponse -> IO () -dump MainFile _ _ _ ":none" (hdrs, _) = do +dump MainFile _ _ _ ":none" (hdrs, _, trls) = do timePrint hdrs -dump MainFile _ _ _ ":stdout" (hdrs, body) = do + timePrint trls +dump MainFile _ _ _ ":stdout" (hdrs, body, trls) = do timePrint hdrs ByteString.putStrLn body -dump PushPromiseFile _ _ _ ":stdout" (hdrs, _) = do + timePrint trls +dump PushPromiseFile _ _ _ ":stdout" (hdrs, _, trls) = do timePrint hdrs -dump MainFile _ _ _ ":stdout-pp" (hdrs, body) = do + timePrint trls +dump MainFile _ _ _ ":stdout-pp" (hdrs, body, trls) = do timePrint hdrs ByteString.putStrLn body -dump PushPromiseFile _ _ _ ":stdout-pp" (hdrs, body) = do + timePrint trls +dump PushPromiseFile _ _ _ ":stdout-pp" (hdrs, body, trls) = do timePrint hdrs ByteString.putStrLn body -dump _ querystring nquery nthread prefix (hdrs, body) = do + timePrint trls +dump _ querystring nquery nthread prefix (hdrs, body, trls) = do timePrint hdrs ByteString.writeFile filepath body + timePrint trls where filepath = mconcat [ prefix , "/" 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.5.0.0 +version: 0.6.0.0 synopsis: A native HTTP2 client library. description: Please read the README.md at the homepage. homepage: https://github.com/lucasdicioccio/http2-client 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 @@ -12,6 +12,7 @@ , newHttp2Client , withHttp2Stream , headers + , trailers , sendData -- * Starting clients , Http2Client(..) @@ -37,6 +38,7 @@ , Http2ClientAsyncs(..) , _gtfo -- * Convenience re-exports + , StreamEvent(..) , module Network.HTTP2.Client.FrameConnection , module Network.Socket , module Network.TLS @@ -46,12 +48,11 @@ import Control.Exception (bracket, throwIO, SomeException, catch) import Control.Concurrent.MVar (newMVar, takeMVar, putMVar) import Control.Concurrent (threadDelay) -import Control.Monad (forever, when, forM_) +import Control.Monad (forever, void, when, forM_) import Data.ByteString (ByteString) import qualified Data.ByteString as ByteString import Data.IORef (newIORef, atomicModifyIORef', readIORef) import Data.Maybe (fromMaybe) -import GHC.Exception (Exception) import Network.HPACK as HPACK import Network.HTTP2 as HTTP2 import Network.Socket (HostName, PortNumber) @@ -236,22 +237,23 @@ -- ^ Changes the PRIORITY of this stream. , _rst :: ErrorCodeId -> IO () -- ^ Resets this stream with a RST frame. You should not use this stream past this call. - , _waitHeaders :: IO (FrameHeader, StreamId, Either ErrorCode HeaderList) - -- ^ Waits for HTTP headers from the server. This function also passes the - -- last frame header of the PUSH-PROMISE, HEADERS, or CONTINUATION sequence of frames. - -- Waiting more than once per stream will hang as headers are sent only one time. - , _waitData :: IO (FrameHeader, Either ErrorCode ByteString) - -- ^ Waits for a DATA frame chunk. A user should testEndStream on the frame - -- header to know when the server is done with the stream. + , _waitEvent :: IO StreamEvent + -- ^ Waits for the next event on the stream. , _sendDataChunk :: (FrameFlags -> FrameFlags) -> ByteString -> IO () -- ^ Sends a DATA frame chunk. You can use send empty frames with only -- headers modifiers to close streams. This function is oblivious to framing -- and hence does not respect the RFC if sending large blocks. Use 'sendData' -- to chunk and send naively according to server\'s preferences. This function -- can be useful if you intend to handle the framing yourself. - , _waitPushPromise :: Maybe (PushPromiseHandler -> IO ()) + , _handlePushPromise :: StreamId -> HeaderList -> PushPromiseHandler -> IO () } +-- | Sends HTTP trailers. +-- +-- Trailers should be the last thing sent over a stream. +trailers :: Http2Stream -> HPACK.HeaderList -> (FrameFlags -> FrameFlags) -> IO () +trailers stream hdrs flagmod = void $ _headers stream hdrs flagmod + -- | Handler upon receiving a PUSH_PROMISE from the server. -- -- The functions for 'Http2Stream' are similar to those used in ''. But callers @@ -398,11 +400,11 @@ goAwayHandler fallbackHandler - creditFrames <- newDispatchReadChanIO dispatch - _initOutgoingFlowControl <- newOutgoingFlowControl dispatchControl 0 creditFrames _initIncomingFlowControl <- newIncomingFlowControl dispatchControl controlStream + (_initOutgoingFlowControl,windowUpdatesChan) <- newOutgoingFlowControl dispatchControl 0 + dispatchHPACK <- newDispatchHPACKIO decoderBufSize - let incomingLoop = dispatchLoop conn dispatch dispatchControl _initIncomingFlowControl dispatchHPACK + let incomingLoop = dispatchLoop conn dispatch dispatchControl windowUpdatesChan _initIncomingFlowControl dispatchHPACK {- Setup for client-initiated streams. -} conccurentStreams <- newIORef 0 @@ -421,32 +423,27 @@ return $ Left $ TooMuchConcurrency roomNeeded else Right <$> do cont <- withClientStreamId $ \sid -> do - streamHeaders <- newDispatchHPACKReadHeadersChanIO dispatchHPACK - streamFrames <- newDispatchReadChanIO dispatch - streamPP <- newDispatchHPACKReadPushPromisesChanIO dispatchHPACK + dispatchStream <- newDispatchStreamIO sid initializeStream conn + dispatch dispatchControl - streamFrames - streamHeaders - (Just streamPP) - sid + dispatchStream getWork + Idle v <- cont atomicModifyIORef' conccurentStreams (\n -> (n - 1, ())) pure v let _initPing dat = do - -- Need to dupChan before sending the query to avoid missing a fast - -- answer if the network is fast. - pingFrames <- newDispatchReadChanIO dispatch + handler <- registerPingHandler dispatchControl dat sendPingFrame controlStream id dat - return $ waitFrame (isPingReply dat) pingFrames + return $ waitPingReply handler + let _initSettings settslist = do - -- Much like _ping, we need to dupChan before sending the query. - settingsFrames <- newDispatchReadChanIO dispatch + handler <- registerSetSettingsHandler dispatchControl sendSettingsFrame controlStream id settslist return $ do - ret <- waitFrame isSettingsReply settingsFrames + ret <- waitSetSettingsReply handler modifySettings dispatchControl (\(ConnectionSettings cli srv) -> (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ())) @@ -461,56 +458,50 @@ return (incomingLoop, InitHttp2Client{..}) initializeStream - :: Exception e - => Http2FrameConnection + :: Http2FrameConnection + -> Dispatch -> DispatchControl - -> FramesChan e - -> HeadersChan - -> Maybe (PushPromisesChan e) - -- ^ Just some PushPromiseChan for new client streams. Nothing for server - -- streams. - -> StreamId + -> DispatchStream -> (Http2Stream -> StreamDefinition a) + -> StreamFSMState -> IO (IO a) -initializeStream conn control frames headersFrames mPushPromises sid getWork = do +initializeStream conn dispatch control stream getWork initialState = do + let sid = _dispatchStreamId stream let frameStream = makeFrameClientStream conn sid - -- Register interest in frames. - credits <- dupChan frames + let events = _dispatchStreamReadEvents stream + -- Builds a flow-control context. incomingStreamFlowControl <- newIncomingFlowControl control frameStream - outgoingStreamFlowControl <- newOutgoingFlowControl control sid credits + (outgoingStreamFlowControl, windowUpdatesChan) <- newOutgoingFlowControl control sid + registerStream dispatch sid (StreamState windowUpdatesChan events initialState) -- Prepare handlers. let _headers headersList flags = do splitter <- settingsPayloadSplitter <$> readSettings control - sendHeaders frameStream (_dispatchControlHpackEncoder control) headersList splitter flags - let _waitHeaders = waitHeadersWithStreamId sid headersFrames - let _waitData = do - (fh, fp) <- waitFrameWithTypeIdForStreamId sid [FrameRSTStream, FrameData] frames - case fp of - DataFrame dat -> do - _ <- _consumeCredit incomingStreamFlowControl (HTTP2.payloadLength fh) - _addCredit incomingStreamFlowControl (HTTP2.payloadLength fh) - return (fh, Right dat) - RSTStreamFrame err -> do - return (fh, Left $ HTTP2.fromErrorCodeId err) - _ -> error "waitFrameWithTypeIdForStreamId returned an unknown frame" - let _sendDataChunk = sendDataFrame frameStream - let _rst = sendResetFrame frameStream + cst <- sendHeaders frameStream (_dispatchControlHpackEncoder control) headersList splitter flags + when (testEndStream $ flags 0) $ do + closeLocalStream dispatch sid + return cst + let _waitEvent = readChan events + let _sendDataChunk flags dat = do + sendDataFrame frameStream flags dat + when (testEndStream $ flags 0) $ do + closeLocalStream dispatch sid + let _rst = \err -> do + sendResetFrame frameStream err + closeReleaseStream dispatch sid let _prio = sendPriorityFrame frameStream - let makeWaitPushPromise pushPromises ppHandler = do - (_,ppFrames,ppHeaders,ppSid,ppReadHeaders) <- waitPushPromiseWithParentStreamId sid pushPromises - let mkStreamActions stream = StreamDefinition (return CST) (ppHandler sid stream ppReadHeaders) + let _handlePushPromise ppSid ppHeaders ppHandler = do + let mkStreamActions s = StreamDefinition (return CST) (ppHandler sid s ppHeaders) + newStream <- newDispatchStreamIO ppSid ppCont <- initializeStream conn + dispatch control - ppFrames - ppHeaders - Nothing - ppSid + newStream mkStreamActions + ReservedRemote ppCont - let _waitPushPromise = fmap makeWaitPushPromise mPushPromises let streamActions = getWork $ Http2Stream{..} @@ -524,41 +515,83 @@ :: Http2FrameConnection -> Dispatch -> DispatchControl + -> Chan (FrameHeader, FramePayload) -> IncomingFlowControl -> DispatchHPACK -> IO () -dispatchLoop conn d dc inFlowControl dh = do +dispatchLoop conn d dc windowUpdatesChan inFlowControl dh = do let getNextFrame = next conn delayException . forever $ do frame <- getNextFrame dispatchFramesStep frame d whenFrame (hasStreamId 0) frame $ \got -> - dispatchControlFramesStep got dc + dispatchControlFramesStep windowUpdatesChan got dc whenFrame (hasTypeId [FrameData]) frame $ \got -> - creditDataFramesStep inFlowControl got - whenFrame (hasTypeId [FrameRSTStream, FramePushPromise, FrameHeaders]) frame $ \got -> do - let hpackLoop (FinishedWithHeaders act) = - act - hpackLoop (FinishedWithPushPromise act) = do - newDispatchReadChanIO d >>= act + creditDataFramesStep d inFlowControl got + whenFrame (hasTypeId [FrameWindowUpdate]) frame $ \got -> do + updateWindowsStep d got + whenFrame (hasTypeId [FramePushPromise, FrameHeaders]) frame $ \got -> do + let hpackLoop (FinishedWithHeaders curFh sId mkNewHdrs) = do + newHdrs <- mkNewHdrs + chan <- fmap _streamStateEvents <$> lookupStreamState d sId + let msg = StreamHeadersEvent curFh newHdrs + maybe (return ()) (flip writeChan msg) chan + hpackLoop (FinishedWithPushPromise curFh parentSid newSid mkNewHdrs) = do + newHdrs <- mkNewHdrs + chan <- fmap _streamStateEvents <$> lookupStreamState d parentSid + let msg = StreamPushPromiseEvent curFh newSid newHdrs + maybe (return ()) (flip writeChan msg) chan hpackLoop (WaitContinuation act) = getNextFrame >>= act >>= hpackLoop + hpackLoop (FailedHeaders curFh sId err) = do + chan <- fmap _streamStateEvents <$> lookupStreamState d sId + let msg = StreamErrorEvent curFh err + maybe (return ()) (flip writeChan msg) chan hpackLoop (dispatchHPACKFramesStep got dh) + whenFrame (hasTypeId [FrameRSTStream]) frame $ \got -> do + handleRSTStep d got + finalizeFramesStep frame d +handleRSTStep + :: Dispatch + -> (FrameHeader, FramePayload) + -> IO () +handleRSTStep d (fh, payload) = do + let sid = streamId fh + case payload of + (RSTStreamFrame err) -> do + chan <- fmap _streamStateEvents <$> lookupStreamState d sid + let msg = StreamErrorEvent fh (HTTP2.fromErrorCodeId err) + maybe (return ()) (flip writeChan msg) chan + closeReleaseStream d sid + _ -> + error $ "expecting RSTFrame but got " ++ show payload + dispatchFramesStep :: (FrameHeader, Either HTTP2Error FramePayload) -> Dispatch -> IO () -dispatchFramesStep frame@(fh,_) (Dispatch{..}) = do +dispatchFramesStep (fh,_) d = do + let sid = streamId fh -- Remember highest streamId. - atomicModifyIORef' _dispatchMaxStreamId (\n -> (max n (streamId fh), ())) - writeChan _dispatchWriteChan frame + atomicModifyIORef' (_dispatchMaxStreamId d) (\n -> (max n sid, ())) +finalizeFramesStep + :: (FrameHeader, Either HTTP2Error FramePayload) + -> Dispatch + -> IO () +finalizeFramesStep (fh,_) d = do + let sid = streamId fh + -- Remote-close streams that match. + when (testEndStream $ flags fh) $ do + closeRemoteStream d sid + dispatchControlFramesStep - :: (FrameHeader, FramePayload) + :: Chan (FrameHeader, FramePayload) + -> (FrameHeader, FramePayload) -> DispatchControl -> IO () -dispatchControlFramesStep controlFrame@(fh, payload) (DispatchControl{..}) = do +dispatchControlFramesStep windowUpdatesChan controlFrame@(fh, payload) control@(DispatchControl{..}) = do case payload of (SettingsFrame settsList) | not . testAck . flags $ fh -> do @@ -570,24 +603,22 @@ (lookup SettingsHeaderTableSize settsList) _dispatchControlAckSettings | otherwise -> do - ignore "TODO: settings ack should be taken into account only after reception, we should return a waitSettingsAck in the _settings function" + handler <- lookupAndReleaseSetSettingsHandler control + maybe (return ()) (notifySetSettingsHandler controlFrame) handler (PingFrame pingMsg) | not . testAck . flags $ fh -> _dispatchControlAckPing pingMsg | otherwise -> do - ignore "PingFrame replies waited for in the requestor thread" + handler <- lookupAndReleasePingHandler control pingMsg + maybe (return ()) (notifyPingHandler controlFrame) handler (WindowUpdateFrame _ ) -> - ignore "connection-wide WindowUpdateFrame waited for in OutgoingFlowControl threads" + writeChan windowUpdatesChan controlFrame (GoAwayFrame lastSid errCode reason) -> _dispatchControlOnGoAway $ RemoteSentGoAwayFrame lastSid errCode reason _ -> _dispatchControlOnFallback controlFrame - where - ignore :: String -> IO () - ignore _ = return () - -- | We currently need a specific step in the main loop for crediting streams -- because a client user may programmatically reset and stop listening for a -- stream and stop calling waitData (which credits streams). @@ -595,23 +626,43 @@ -- TODO: modify the '_rst' function to wait and credit all the remaining data -- that could have been sent in flight creditDataFramesStep - :: IncomingFlowControl + :: Dispatch + -> IncomingFlowControl -> (FrameHeader, FramePayload) -> IO () -creditDataFramesStep flowControl (fh,_) = do +creditDataFramesStep d flowControl (fh,payload) = do -- TODO: error if detect over-run. Current implementation credits -- everything back. Hence, over-run should never happen. _ <- _consumeCredit flowControl (HTTP2.payloadLength fh) _addCredit flowControl (HTTP2.payloadLength fh) + -- Write to the interested streams. + let sid = streamId fh + case payload of + (DataFrame dat) -> do + chan <- fmap _streamStateEvents <$> lookupStreamState d sid + maybe (return ()) (flip writeChan $ StreamDataEvent fh dat) chan + _ -> + error $ "expecting DataFrame but got " ++ show payload + +updateWindowsStep + :: Dispatch + -> (FrameHeader, FramePayload) + -> IO () +updateWindowsStep d got@(fh,_) = do + let sid = HTTP2.streamId fh + chan <- fmap _streamStateWindowUpdatesChan <$> lookupStreamState d sid + maybe (return ()) (flip writeChan got) chan --TODO: refer to RFC for erroring on idle/closed streams + data HPACKLoopDecision = ForwardHeader !StreamId | OpenPushPromise !StreamId !StreamId data HPACKStepResult = WaitContinuation !((FrameHeader, Either HTTP2Error FramePayload) -> IO HPACKStepResult) - | FinishedWithHeaders !(IO ()) - | FinishedWithPushPromise !(DispatchChan -> IO ()) + | FailedHeaders !FrameHeader !StreamId ErrorCode + | FinishedWithHeaders !FrameHeader !StreamId (IO HeaderList) + | FinishedWithPushPromise !FrameHeader !StreamId !StreamId (IO HeaderList) dispatchHPACKFramesStep :: (FrameHeader, FramePayload) @@ -649,23 +700,12 @@ _ -> error "continued frame has invalid type") else case decision of - ForwardHeader sId -> FinishedWithHeaders $ do - newHdrs <- decodeHeader _dispatchHPACKDynamicTable buffer - writeChan _dispatchHPACKWriteHeadersChan (curFh, sId, Right newHdrs) - OpenPushPromise parentSid newSid -> FinishedWithPushPromise $ \ppChan -> do - newHdrs <- decodeHeader _dispatchHPACKDynamicTable buffer - -- Important: We duplicate the channel here or we risk - -- losing DATA or HEADERS+CONTINUATION frames sent over - -- the main stream because even though - -- 'initializeStream' dupes frame channels, this - -- function will has no guarantee that the parent - -- stream and the push-promise handler that will - -- initiate the new stream are synchronous. - ppHeaders <- dupChan _dispatchHPACKWriteHeadersChan - writeChan _dispatchHPACKWritePushPromisesChan (parentSid, ppChan, ppHeaders, newSid, newHdrs) - - go curFh _ (Left err) = FinishedWithHeaders $ - writeChan _dispatchHPACKWriteHeadersChan (curFh, sid, (Left $ HTTP2.fromErrorCodeId err)) + ForwardHeader sId -> + FinishedWithHeaders curFh sId (decodeHeader _dispatchHPACKDynamicTable buffer) + OpenPushPromise parentSid newSid -> + FinishedWithPushPromise curFh parentSid newSid (decodeHeader _dispatchHPACKDynamicTable buffer) + go curFh _ (Left err) = + FailedHeaders curFh sid (HTTP2.fromErrorCodeId err) newIncomingFlowControl @@ -701,13 +741,12 @@ return $ IncomingFlowControl _addCredit _consumeCredit _updateWindow newOutgoingFlowControl :: - Exception e - => DispatchControl + DispatchControl -> StreamId - -> Chan (FrameHeader, Either e FramePayload) - -> IO OutgoingFlowControl -newOutgoingFlowControl control sid frames = do + -> IO (OutgoingFlowControl, Chan (FrameHeader, FramePayload)) +newOutgoingFlowControl control sid = do credit <- newIORef 0 + frames <- newChan let getBase = if sid == 0 then return HTTP2.defaultInitialWindowSize else initialWindowSize . _serverSettings <$> readSettings control @@ -722,10 +761,10 @@ if got > 0 then return got else do - amount <- race (waitSettingsChange base) waitSomeCredit + amount <- race (waitSettingsChange base) (waitSomeCredit frames) receive (either (const 0) id amount) withdraw n - return $ OutgoingFlowControl receive withdraw + return $ (OutgoingFlowControl receive withdraw, frames) where -- TODO: broadcast settings changes from ConnectionSettings using a better data type -- than IORef+busy loop. Currently the busy loop is fine because @@ -737,11 +776,13 @@ waitSettingsChange prev = do new <- initialWindowSize . _serverSettings <$> readSettings control if new == prev then threadDelay 1000000 >> waitSettingsChange prev else return () - waitSomeCredit = do - (_, fp) <- waitFrameWithTypeIdForStreamId sid [FrameWindowUpdate] frames - case fp of - WindowUpdateFrame amt -> return amt - _ -> error "waitFrameWithTypeIdForStreamId returned an unknown frame" + waitSomeCredit frames = do + got <- readChan frames + case got of + (_, WindowUpdateFrame amt) -> + return amt + _ -> + error "got forwarded an unknown frame" sendHeaders :: Http2FrameClientStream @@ -751,13 +792,16 @@ -> (FrameFlags -> FrameFlags) -> IO StreamThread sendHeaders s enc hdrs blockSplitter flagmod = do - headerBlockFragments <- blockSplitter <$> _encodeHeaders enc hdrs - let framers = (HeadersFrame Nothing) : repeat ContinuationFrame - let frames = zipWith ($) framers headerBlockFragments - let modifiersReversed = (HTTP2.setEndHeader . flagmod) : repeat id - let arrangedFrames = reverse $ zip modifiersReversed (reverse frames) - sendBackToBack s arrangedFrames + _sendFrames s mkFrames return CST + where + mkFrames = do + headerBlockFragments <- blockSplitter <$> _encodeHeaders enc hdrs + let framers = (HeadersFrame Nothing) : repeat ContinuationFrame + let frames = zipWith ($) framers headerBlockFragments + let modifiersReversed = (HTTP2.setEndHeader . flagmod) : repeat id + let arrangedFrames = reverse $ zip modifiersReversed (reverse frames) + return arrangedFrames -- | A function able to split a header block into multiple fragments. type PayloadSplitter = ByteString -> [ByteString] 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 @@ -31,9 +31,9 @@ type HeadersChan = Chan HeadersChanContent -type PushPromisesChanContent e = (StreamId, FramesChan e, HeadersChan, StreamId, HeaderList) +type PushPromisesChanContent = (StreamId, StreamId, HeaderList) -type PushPromisesChan e = Chan (PushPromisesChanContent e) +type PushPromisesChan = Chan PushPromisesChanContent waitFrameWithStreamId :: Exception e @@ -134,13 +134,13 @@ waitPushPromiseWithParentStreamId :: StreamId - -> PushPromisesChan e - -> IO (PushPromisesChanContent e) + -> PushPromisesChan + -> IO PushPromisesChanContent waitPushPromiseWithParentStreamId sid chan = loop where loop = do - tuple@(parentSid,_,_,_,_) <- readChan chan + tuple@(parentSid,_,_) <- readChan chan if parentSid == sid then return tuple else loop 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,11 +1,17 @@ - +{-# LANGUAGE BangPatterns #-} module Network.HTTP2.Client.Dispatch where import Control.Exception (throwIO) import Data.ByteString (ByteString) +import qualified Data.ByteString.Internal as ByteString +import Foreign.Marshal.Alloc (mallocBytes, finalizerFree) +import Foreign.ForeignPtr (newForeignPtr) import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) +import Data.IntMap (IntMap) +import qualified Data.IntMap as IntMap import GHC.Exception (Exception) import Network.HPACK as HPACK +import qualified Network.HPACK.Token as HPACK import Network.HTTP2 as HTTP2 import Network.HTTP2.Client.Channels @@ -40,20 +46,74 @@ defaultGoAwayHandler :: GoAwayHandler defaultGoAwayHandler = throwIO +data StreamFSMState = + Idle + | ReservedRemote + | Open + | HalfClosedRemote + | HalfClosedLocal + | Closed + +data StreamEvent = + StreamHeadersEvent !FrameHeader !HeaderList + | StreamPushPromiseEvent !FrameHeader !StreamId !HeaderList + | StreamDataEvent !FrameHeader ByteString + | StreamErrorEvent !FrameHeader ErrorCode + deriving Show + +data StreamState = StreamState { + _streamStateWindowUpdatesChan :: !(Chan (FrameHeader, FramePayload)) + , _streamStateEvents :: !(Chan StreamEvent) + , _streamStateFSMState :: !StreamFSMState + } + data Dispatch = Dispatch { - _dispatchWriteChan :: !DispatchChan - , _dispatchMaxStreamId :: !(IORef StreamId) + _dispatchMaxStreamId :: !(IORef StreamId) + , _dispatchCurrentStreams :: !(IORef (IntMap StreamState)) } newDispatchIO :: IO Dispatch -newDispatchIO = Dispatch <$> newChan <*> newIORef 0 - -newDispatchReadChanIO :: Dispatch -> IO DispatchChan -newDispatchReadChanIO = dupChan . _dispatchWriteChan +newDispatchIO = Dispatch <$> newIORef 0 <*> newIORef (IntMap.empty) readMaxReceivedStreamIdIO :: Dispatch -> IO StreamId readMaxReceivedStreamIdIO = readIORef . _dispatchMaxStreamId +registerStream :: Dispatch -> StreamId -> StreamState -> IO () +registerStream d sid st = + atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs -> + let v = (IntMap.insert sid st xs) in (v, ()) + +lookupStreamState :: Dispatch -> StreamId -> IO (Maybe StreamState) +lookupStreamState d sid = + IntMap.lookup sid <$> readIORef (_dispatchCurrentStreams d) + +closeLocalStream :: Dispatch -> StreamId -> IO () +closeLocalStream d sid = + atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs -> + let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ()) + where + f :: StreamId -> StreamState -> Maybe StreamState + f _ st = case _streamStateFSMState st of + HalfClosedRemote -> Nothing + Closed -> Nothing + _ -> Just $ st { _streamStateFSMState = HalfClosedLocal } + +closeRemoteStream :: Dispatch -> StreamId -> IO () +closeRemoteStream d sid = + atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs -> + let (_,v) = IntMap.updateLookupWithKey f sid xs in (v, ()) + where + f :: StreamId -> StreamState -> Maybe StreamState + f _ st = case _streamStateFSMState st of + HalfClosedLocal -> Nothing + Closed -> Nothing + _ -> Just $ st { _streamStateFSMState = HalfClosedRemote } + +closeReleaseStream :: Dispatch -> StreamId -> IO () +closeReleaseStream d sid = + atomicModifyIORef' (_dispatchCurrentStreams d) $ \xs -> + let v = (IntMap.delete sid xs) in (v, ()) + -- | Couples client and server settings together. data ConnectionSettings = ConnectionSettings { _clientSettings :: !Settings @@ -64,6 +124,58 @@ defaultConnectionSettings = ConnectionSettings defaultSettings defaultSettings +data PingHandler = PingHandler !(Chan (FrameHeader, FramePayload)) + +newPingHandler :: IO PingHandler +newPingHandler = PingHandler <$> newChan + +notifyPingHandler :: (FrameHeader, FramePayload) -> PingHandler -> IO () +notifyPingHandler dat (PingHandler c) = writeChan c dat + +waitPingReply :: PingHandler -> IO (FrameHeader, FramePayload) +waitPingReply (PingHandler c) = readChan c + +data SetSettingsHandler = SetSettingsHandler !(Chan (FrameHeader, FramePayload)) + +newSetSettingsHandler :: IO SetSettingsHandler +newSetSettingsHandler = SetSettingsHandler <$> newChan + +notifySetSettingsHandler :: (FrameHeader, FramePayload) -> SetSettingsHandler -> IO () +notifySetSettingsHandler dat (SetSettingsHandler c) = writeChan c dat + +waitSetSettingsReply :: SetSettingsHandler -> IO (FrameHeader, FramePayload) +waitSetSettingsReply (SetSettingsHandler c) = readChan c + +registerPingHandler :: DispatchControl -> ByteString -> IO PingHandler +registerPingHandler dc dat = do + handler <- newPingHandler + atomicModifyIORef' (_dispatchControlPingHandlers dc) (\xs -> + ((dat,handler):xs, ())) + return handler + +lookupAndReleasePingHandler :: DispatchControl -> ByteString -> IO (Maybe PingHandler) +lookupAndReleasePingHandler dc dat = + atomicModifyIORef' (_dispatchControlPingHandlers dc) f + where + -- Note: we considered doing a single pass for this folds but we expect the + -- size of handlers to be small anyway (hence, we use a List for + -- storing the handlers). + f xs = (filter (\x -> dat /= fst x) xs, lookup dat xs) + +registerSetSettingsHandler :: DispatchControl -> IO SetSettingsHandler +registerSetSettingsHandler dc = do + handler <- newSetSettingsHandler + atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) (\xs -> + (handler:xs, ())) + return handler + +lookupAndReleaseSetSettingsHandler :: DispatchControl -> IO (Maybe SetSettingsHandler) +lookupAndReleaseSetSettingsHandler dc = + atomicModifyIORef' (_dispatchControlSetSettingsHandlers dc) f + where + f [] = ([], Nothing) + f (x:xs) = (xs, Just x) + data DispatchControl = DispatchControl { _dispatchControlConnectionSettings :: !(IORef ConnectionSettings) , _dispatchControlHpackEncoder :: !HpackEncoderContext @@ -71,6 +183,8 @@ , _dispatchControlAckSettings :: !(IO ()) , _dispatchControlOnGoAway :: !GoAwayHandler , _dispatchControlOnFallback :: !FallBackFrameHandler + , _dispatchControlPingHandlers :: !(IORef [(ByteString, PingHandler)]) + , _dispatchControlSetSettingsHandlers :: !(IORef [SetSettingsHandler]) } newDispatchControlIO @@ -82,18 +196,30 @@ -> IO DispatchControl newDispatchControlIO encoderBufSize ackPing ackSetts onGoAway onFallback = DispatchControl <$> newIORef defaultConnectionSettings - <*> hpackEncoder + <*> newHpackEncoderContext encoderBufSize <*> pure ackPing <*> pure ackSetts <*> pure onGoAway <*> pure onFallback - where - hpackEncoder = do - let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True }) - dt <- HPACK.newDynamicTableForEncoding HPACK.defaultDynamicTableSize - return $ HpackEncoderContext - (HPACK.encodeHeader strategy encoderBufSize dt) + <*> newIORef [] + <*> newIORef [] + +newHpackEncoderContext :: Size -> IO HpackEncoderContext +newHpackEncoderContext encoderBufSize = 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) (\n -> HPACK.setLimitForEncoding n dt) + where + encoder strategy dt buf ptr hdrs = do + let hdrs' = fmap (\(k,v) -> let !t = HPACK.toToken k in (t,v)) hdrs + remainder <- HPACK.encodeTokenHeader buf encoderBufSize strategy True dt hdrs' + case remainder of + ([],len) -> pure $ ByteString.fromForeignPtr ptr 0 len + (_,_) -> throwIO HPACK.BufferOverrun readSettings :: DispatchControl -> IO ConnectionSettings readSettings = readIORef . _dispatchControlConnectionSettings @@ -108,23 +234,23 @@ } data DispatchHPACK = DispatchHPACK { - _dispatchHPACKWriteHeadersChan :: !HeadersChan - , _dispatchHPACKWritePushPromisesChan :: !(PushPromisesChan HTTP2Error) - , _dispatchHPACKDynamicTable :: !DynamicTable + _dispatchHPACKDynamicTable :: !DynamicTable } newDispatchHPACKIO :: Size -> IO DispatchHPACK newDispatchHPACKIO decoderBufSize = - DispatchHPACK <$> newChan <*> newChan <*> newDecoder + DispatchHPACK <$> newDecoder where newDecoder = newDynamicTableForDecoding HPACK.defaultDynamicTableSize decoderBufSize -newDispatchHPACKReadHeadersChanIO :: DispatchHPACK -> IO HeadersChan -newDispatchHPACKReadHeadersChanIO = - dupChan . _dispatchHPACKWriteHeadersChan +data DispatchStream = DispatchStream { + _dispatchStreamId :: !StreamId + , _dispatchStreamReadEvents :: !(Chan StreamEvent) + } -newDispatchHPACKReadPushPromisesChanIO :: DispatchHPACK -> IO (PushPromisesChan HTTP2Error) -newDispatchHPACKReadPushPromisesChanIO = - dupChan . _dispatchHPACKWritePushPromisesChan +newDispatchStreamIO :: StreamId -> IO DispatchStream +newDispatchStreamIO sid = + DispatchStream <$> pure sid + <*> newChan 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 @@ -45,7 +45,7 @@ makeFrameClientStream = _makeFrameClientStream data Http2FrameClientStream = Http2FrameClientStream { - _sendFrames :: [(FrameFlags -> FrameFlags, FramePayload)] -> IO () + _sendFrames :: IO [(FrameFlags -> FrameFlags, FramePayload)] -> IO () -- ^ Sends a frame to the server. -- The first argument is a FrameFlags modifier (e.g., to sed the -- end-of-stream flag). @@ -54,11 +54,11 @@ -- | Sends a frame to the server. sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> IO () -sendOne client f payload = _sendFrames client [(f, payload)] +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 = _sendFrames +sendBackToBack client payloads = _sendFrames client (pure payloads) data Http2ServerStream = Http2ServerStream { _nextHeaderAndFrame :: IO (FrameHeader, Either HTTP2Error FramePayload) @@ -91,7 +91,9 @@ let info = encodeInfo modifyFF streamID _sendRaw http2conn $ HTTP2.encodeFrame info frame - putFrames xs = writeProtect . void $ traverse (uncurry putFrame) xs + putFrames f = writeProtect . void $ do + xs <- f + traverse (uncurry putFrame) xs in Http2FrameClientStream putFrames streamID nextServerFrameChunk = Http2ServerStream $ do 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 @@ -12,7 +12,6 @@ upload , waitStream , fromStreamResult - , onPushPromise , StreamResult , StreamResponse -- * Diagnostics @@ -28,7 +27,6 @@ import qualified Data.ByteString as ByteString import Control.Concurrent (threadDelay) import Control.Concurrent.Async (race) -import Control.Monad (forever) import Network.HTTP2.Client @@ -58,10 +56,10 @@ -- | Result containing the unpacked headers and all frames received in on a -- stream. See 'StreamResponse' and 'fromStreamResult' to get a higher-level -- utility. -type StreamResult = (Either HTTP2.ErrorCode HPACK.HeaderList, [Either HTTP2.ErrorCode ByteString]) +type StreamResult = (Either HTTP2.ErrorCode HPACK.HeaderList, [Either HTTP2.ErrorCode ByteString], Maybe HPACK.HeaderList) -- | An HTTP2 response, once fully received, is made of headers and a payload. -type StreamResponse = (HPACK.HeaderList, ByteString) +type StreamResponse = (HPACK.HeaderList, ByteString, Maybe HPACK.HeaderList) -- | Uploads a whole HTTP body at a time. -- @@ -78,6 +76,8 @@ -- setEndStream and no payload. upload :: ByteString -- ^ HTTP body. + -> (HTTP2.FrameFlags -> HTTP2.FrameFlags) + -- ^ Flag modifier for the last DATA frame sent. -> Http2Client -- ^ The client. -> OutgoingFlowControl @@ -89,9 +89,9 @@ -> OutgoingFlowControl -- ^ The flow control for this stream. -> IO () -upload "" conn _ stream _ = do - sendData conn stream HTTP2.setEndStream "" -upload dat conn connectionFlowControl stream streamFlowControl = do +upload "" flagmod conn _ stream _ = do + sendData conn stream flagmod "" +upload dat flagmod conn connectionFlowControl stream streamFlowControl = do let wanted = ByteString.length dat gotStream <- _withdrawCredit streamFlowControl wanted @@ -105,46 +105,55 @@ if got == wanted then - uploadChunks HTTP2.setEndStream + uploadChunks flagmod else do uploadChunks id - upload (ByteString.drop got dat) conn connectionFlowControl stream streamFlowControl + upload (ByteString.drop got dat) flagmod conn connectionFlowControl stream streamFlowControl -- | Wait for a stream until completion. -- -- This function is fine if you don't want to consume results in chunks. See -- 'fromStreamResult' to collect the complicated 'StreamResult' into a simpler -- 'StramResponse'. -waitStream :: Http2Stream -> IncomingFlowControl -> IO StreamResult -waitStream stream streamFlowControl = do - (_,_,hdrs) <- _waitHeaders stream - dataFrames <- moredata [] - return (hdrs, reverse dataFrames) +waitStream :: Http2Stream + -> IncomingFlowControl + -> PushPromiseHandler + -> IO StreamResult +waitStream stream streamFlowControl ppHandler = do + ev <- _waitEvent stream + case ev of + StreamHeadersEvent _ hdrs -> do + (dfrms,trls) <- waitDataFrames [] + return (Right hdrs, reverse dfrms, trls) + StreamPushPromiseEvent _ ppSid ppHdrs -> do + _handlePushPromise stream ppSid ppHdrs ppHandler + waitStream stream streamFlowControl ppHandler + _ -> + error $ "expecting StreamHeadersEvent but got " ++ show ev where - moredata xs = do - (fh, x) <- _waitData stream - if HTTP2.testEndStream (HTTP2.flags fh) - then - return (x:xs) - else do - _ <- _updateWindow $ streamFlowControl - moredata (x:xs) + waitDataFrames xs = do + ev <- _waitEvent stream + case ev of + StreamDataEvent fh x + | HTTP2.testEndStream (HTTP2.flags fh) -> + return ((Right x):xs, Nothing) + | otherwise -> do + _ <- _consumeCredit streamFlowControl (HTTP2.payloadLength fh) + _addCredit streamFlowControl (HTTP2.payloadLength fh) + _ <- _updateWindow $ streamFlowControl + waitDataFrames ((Right x):xs) + StreamPushPromiseEvent _ ppSid ppHdrs -> do + _handlePushPromise stream ppSid ppHdrs ppHandler + waitDataFrames xs + StreamHeadersEvent _ hdrs -> + return (xs, Just hdrs) + _ -> + error $ "expecting StreamDataEvent but got " ++ show ev -- | Converts a StreamResult to a StramResponse, stopping at the first error -- using the `Either HTTP2.ErrorCode` monad. fromStreamResult :: StreamResult -> Either HTTP2.ErrorCode StreamResponse -fromStreamResult (headersE, chunksE) = do +fromStreamResult (headersE, chunksE, trls) = do hdrs <- headersE chunks <- sequence chunksE - return (hdrs, mconcat chunks) - --- | Sequentially wait for every push-promise with a handler. --- --- This function runs forever and you should wrap it with 'withAsync'. --- This function immediately returns if run on a stream that has no PushPromise --- handler (i.e., server-streams when already waiting for push promises). -onPushPromise :: Http2Stream -> PushPromiseHandler -> IO () -onPushPromise stream handler = - case _waitPushPromise stream of - Nothing -> pure () - Just w -> forever $ w handler + return (hdrs, mconcat chunks, trls)