http2-client 0.5.0.0 → 0.6.0.0

raw patch · 7 files changed

+431/−213 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• app/Main.hs +61/−24
• http2-client.cabal +1/−1
• src/Network/HTTP2/Client.hs +164/−120
• src/Network/HTTP2/Client/Channels.hs +5/−5
• src/Network/HTTP2/Client/Dispatch.hs +150/−24
• src/Network/HTTP2/Client/FrameConnection.hs +6/−4
• src/Network/HTTP2/Client/Helpers.hs +44/−35

@@ -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                        , "/" 

@@ -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

@@ -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]

@@ -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

@@ -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

@@ -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

@@ -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)