http2-client 0.1.0.1 → 0.2.0.0

raw patch · 5 files changed

+119/−53 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README.md +10/−11
• app/Main.hs +6/−5
• http2-client.cabal +2/−2
• src/Network/HTTP2/Client.hs +59/−31
• src/Network/HTTP2/Client/Helpers.hs +42/−4

@@ -104,8 +104,13 @@ move in the "looser direction" (e.g., more concurrency) should be applied _after_ ACK-ing the SETTINGS frame. -For simplicity of implementation, the current design apply SETTINGS:-- (client prefs) before sending SETTINGS, irrespective of ACK -- we should change this+The current design apply SETTINGS:+- (client prefs) after receiving a ACK for sent SETTINGS, you get the choice to+  wait for an ACK or wait in a thread, but you must wait for an ACK to apply+  changed settings (the `_settings` function will return an IO to wait for the+  ACK and apply settings). Note that the initial SETTINGS change frame is+  waited for in a thread without library's user intervention (if you feel+  strongly against this choice, please open a bug). - (server prefs) immediately after receiving and hence before sending ACK-SETTINGS  Fortunately, changing settings mid-stream is probably a rare behavior and the@@ -126,8 +131,8 @@  A number of HTTP2 features are currently hardcoded: - PINGs are replied-to immediately (i.e., a server could hog a connection with PINGs)-- received SETTINGS are acknowledged immediately after being taken into account-- sent SETTINGS are implemented immediately before sending, and not after acking+- the initial SETTINGS frame sent to the server is waited-for in a separate+  thread, settings are applied to the connection when the server ACKs the frame - flow-control from DATA frames is decremented immediately when received (in a   separate thread) rather than when consumed from the client - similarly, flow-control re-increment every DATA received as soon as it is@@ -160,8 +165,6 @@  I think the fundamental are right but the following needs tweaking: -- onPushPromise handlers will be requested when opening new streams rather than-  when opening new connections - function to reset a stream will likely be blocking until a RST/EndStream is   received so that all DATA frames are accounted for in the flow-control system - need a way to hook custom flow-control algorithms@@ -186,11 +189,7 @@  ### Various TODO -- modify client SETTINGS only when acknowledged-  * currently make use of new client settings before the server has time to-    know them, this can lead to errors due to inconsistent view of settings-    between sender/receiver-- consider a beter frame-subscription mechanism than broadcast wake-up+- consider a better frame-subscription mechanism than broadcast wake-up   * current system of dupChan everything is prone to errors and may be costly     CPU/latency-wise if the concurrency is high - consider most performant functions for HTTP2.HPACK encoding/decoding

@@ -111,9 +111,9 @@                           , (":authority", ByteString.pack _host)                           ] <> _extraHeaders -    let onPushPromise _ stream streamFlowControl _ = void $ forkIO $ do+    let ppHandler _ stream streamFlowControl _ = void $ forkIO $ do             timePrint ("push stream started" :: String)-            waitStream stream streamFlowControl >>= timePrint+            waitStream stream streamFlowControl >>= timePrint . fromStreamResult             timePrint ("push stream ended" :: String)      let conf = [ (HTTP2.SettingsMaxFrameSize, _settingsMaxFrameSize)@@ -125,7 +125,7 @@                ]     timePrint conf -    conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf onPushPromise+    conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf     _addCredit (_incomingFlowControl conn) _initialWindowKick     _ <- forkIO $ forever $ do             updated <- _updateWindow $ _incomingFlowControl conn@@ -134,7 +134,7 @@      _ <- forkIO $ when (_interPingDelay > 0) $ forever $ do         threadDelay _interPingDelay-        (t0, t1, pingReply) <- ping _pingTimeout "pingpong" conn+        (t0, t1, pingReply) <- ping conn _pingTimeout "pingpong"         timePrint $ ("ping-reply:" :: String, pingReply, diffUTCTime t1 t0)      let go 0 idx = timePrint $ "done worker: " <> show idx@@ -142,8 +142,9 @@             _ <- (_startStream conn $ \stream ->                     let initStream = _headers stream headersPairs (HTTP2.setEndStream)                         handler streamFlowControl _ = do+                            _ <- async $ onPushPromise stream ppHandler                             timePrint $ "stream started " <> show (idx, n)-                            waitStream stream streamFlowControl >>= timePrint+                            waitStream stream streamFlowControl >>= timePrint . fromStreamResult                             timePrint $ "stream ended " <> show (idx, n)                     in StreamDefinition initStream handler)             go (n - 1) idx

@@ -1,7 +1,7 @@ name:                http2-client-version:             0.1.0.1+version:             0.2.0.0 synopsis:            A native HTTP2 client library.-description:         Please read the README.ms at the homepage.+description:         Please read the README.md at the homepage. homepage:            https://github.com/lucasdicioccio/http2-client license:             BSD3 license-file:        LICENSE

@@ -10,6 +10,7 @@     -- * Starting streams     , StreamDefinition(..)     , StreamStarter+    , TooMuchConcurrency(..)     , StreamThread     , Http2Stream(..)     -- * Sending data for POSTs.@@ -30,7 +31,7 @@ import           Control.Concurrent.MVar (newMVar, takeMVar, putMVar) import           Control.Concurrent (forkIO, threadDelay) import           Control.Concurrent.Chan (Chan, newChan, dupChan, readChan, writeChan)-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 (IORef, newIORef, atomicModifyIORef', readIORef)@@ -139,8 +140,11 @@   -- 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 ()-  -- ^ Sends a SETTINGS.+  , _settings         :: SettingsList -> IO (IO (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 ()   -- ^ Sends a GOAWAY.    , _startStream      :: forall a. StreamStarter a@@ -202,6 +206,7 @@   -- 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 :: PushPromiseHandler -> IO ()   }  -- | Handler upon receiving a PUSH_PROMISE from the server.@@ -213,8 +218,8 @@ -- The StreamId corresponds to the parent stream as PUSH_PROMISEs are tied to a -- client-initiated stream. Longer term we may move passing this handler to the -- '_startStream' instead of 'newHttp2Client' (as it is for now).-type PushPromiseHandler a =-    StreamId -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO a+type PushPromiseHandler =+    StreamId -> Http2Stream -> IncomingFlowControl -> OutgoingFlowControl -> IO ()  -- | Helper to carry around the HPACK encoder for outgoing header blocks.. data HpackEncoderContext = HpackEncoderContext {@@ -236,10 +241,8 @@                -- ^ The TLS client parameters (e.g., to allow some certificates).                -> SettingsList                -- ^ Initial SETTINGS that are sent as first frame.-               -> PushPromiseHandler a-               -- ^ Action to perform when a server sends a PUSH_PROMISE.                -> IO Http2Client-newHttp2Client host port encoderBufSize decoderBufSize tlsParams initSettings handlePPStream = do+newHttp2Client host port encoderBufSize decoderBufSize tlsParams initSettings = do     -- network connection     conn <- newHttp2FrameConnection host port tlsParams @@ -283,13 +286,12 @@     _outgoingFlowControl <- newOutgoingFlowControl settings 0 creditFrames     _incomingFlowControl <- newIncomingFlowControl settings controlStream +    serverPushPromises <- newChan+     _ <- forkIO $ incomingHPACKFramesLoop serverStreamFrames                                           serverHeaders-                                          hpackEncoder+                                          serverPushPromises                                           hpackDecoder-                                          conn-                                          settings-                                          handlePPStream      dataFrames <- dupChan serverFrames     _ <- forkIO $ creditDataFramesLoop _incomingFlowControl dataFrames@@ -309,6 +311,7 @@                                      settings                                      serverFrames                                      serverHeaders+                                     serverPushPromises                                      hpackEncoder                                      sid                                      getWork@@ -323,17 +326,22 @@             sendPingFrame controlStream id dat             return $ waitFrame (isPingReply dat) pingFrames     let _settings settslist = do-            atomicModifyIORef' settings-                       (\(ConnectionSettings cli srv) ->-                           (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ()))+            -- Much like _ping, we need to dupChan before sending the query.+            pingFrames <- dupChan serverFrames             sendSettingsFrame controlStream id settslist+            return $ do+                ret <- waitFrame isSettingsReply pingFrames+                atomicModifyIORef' settings+                    (\(ConnectionSettings cli srv) ->+                        (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ()))+                return ret     let _goaway err errStr = do             sId <- readIORef maxReceivedStreamId             sendGTFOFrame controlStream sId err errStr      let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings -    _settings initSettings+    _ <- forkIO . void =<< _settings initSettings      return $ Http2Client{..} @@ -343,17 +351,19 @@   -> IORef ConnectionSettings   -> Chan (FrameHeader, Either e FramePayload)   -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList)+  -> Chan (StreamId, StreamId)   -> HpackEncoderContext   -> StreamId   -> (Http2Stream -> StreamDefinition a)   -> IO (IO a)-initializeStream conn settings serverFrames serverHeaders hpackEncoder sid getWork = do+initializeStream conn settings serverFrames serverHeaders serverPushPromises hpackEncoder sid getWork = do     let frameStream = makeFrameClientStream conn sid      -- Register interest in frames.     frames  <- dupChan serverFrames     credits <- dupChan serverFrames     headers <- dupChan serverHeaders+    pushPromises <- dupChan serverPushPromises      -- Builds a flow-control context.     incomingStreamFlowControl <- newIncomingFlowControl settings frameStream@@ -377,6 +387,18 @@     let _sendDataChunk  = sendDataFrame frameStream     let _rst            = sendResetFrame frameStream     let _prio           = sendPriorityFrame frameStream+    let _waitPushPromise ppHandler = do+            (_,ppSid) <- waitPushPromiseWithParentStreamId sid pushPromises+            let mkStreamActions stream = StreamDefinition (return CST) (ppHandler sid stream)+            ppCont <- initializeStream conn+                                       settings+                                       serverFrames+                                       serverHeaders+                                       serverPushPromises+                                       hpackEncoder+                                       ppSid+                                       mkStreamActions+            ppCont      let streamActions = getWork $ Http2Stream{..} @@ -454,13 +476,10 @@   :: Exception e   => Chan (FrameHeader, Either e FramePayload)   -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList)-  -> HpackEncoderContext+  -> Chan (StreamId, StreamId)   -> DynamicTable-  -> Http2FrameConnection-  -> IORef ConnectionSettings-  -> PushPromiseHandler a   -> IO ()-incomingHPACKFramesLoop frames headers hpackEncoder hpackDecoder conn settings handlePPStream = forever $ do+incomingHPACKFramesLoop frames headers pushPromises hpackDecoder = forever $ do     (fh, fp) <- waitFrameWithTypeId [ FrameRSTStream                                     , FramePushPromise                                     , FrameHeaders@@ -469,15 +488,7 @@     (sId, pattern) <- case fp of             PushPromiseFrame sid hbf -> do                 let parentSid = HTTP2.streamId fh-                let mkStreamActions stream = StreamDefinition (return CST) (handlePPStream parentSid stream)-                cont <- initializeStream conn-                                         settings-                                         frames-                                         headers-                                         hpackEncoder-                                         sid-                                         mkStreamActions-                _ <- cont -- TODO: inline+                writeChan pushPromises (parentSid, sid)                 return (sid, Right hbf)             HeadersFrame _ hbf       -> -- TODO: handle priority                 return (HTTP2.streamId fh, Right hbf)@@ -712,6 +723,10 @@ isPingReply datSent _ (PingFrame datRcv) = datSent == datRcv isPingReply _       _ _                  = False +isSettingsReply :: FrameHeader -> FramePayload -> Bool+isSettingsReply fh (SettingsFrame _) = HTTP2.testAck (flags fh)+isSettingsReply _ _                  = False+ waitHeadersWithStreamId   :: StreamId   -> Chan (FrameHeader, StreamId, t)@@ -730,4 +745,17 @@         tuple@(fH, sId, headers) <- readChan chan         if test fH sId headers         then return tuple+        else loop++waitPushPromiseWithParentStreamId+  :: StreamId+  -> Chan (StreamId, StreamId)+  -> IO (StreamId, StreamId)+waitPushPromiseWithParentStreamId sid chan =+    loop+  where+    loop = do+        pair@(parentSid,_) <- readChan chan+        if parentSid == sid+        then return pair         else loop

@@ -7,21 +7,44 @@ import           Data.ByteString (ByteString) import           Control.Concurrent (threadDelay) import           Control.Concurrent.Async (race)+import           Control.Monad (forever)  import Network.HTTP2.Client -type PingReply = (UTCTime, UTCTime, Either () (HTTP2.FrameHeader, HTTP2.FramePayload))+-- | Opaque type to express an action which timed out.+data TimedOut = TimedOut+  deriving Show -ping :: Int -> ByteString -> Http2Client -> IO PingReply-ping timeout msg conn = do+-- | Result for a 'ping'.+type PingReply = (UTCTime, UTCTime, Either TimedOut (HTTP2.FrameHeader, HTTP2.FramePayload))++-- | Performs a 'ping' and waits for a reply up to a given timeout (in+-- microseconds).+ping :: Http2Client+     -- ^ client connection+     -> Int+     -- ^ timeout in microseconds+     -> ByteString+     -- ^ 8-bytes message to uniquely identify the reply+     -> IO PingReply+ping conn timeout msg = do     t0 <- getCurrentTime     waitPing <- _ping conn msg-    pingReply <- race (threadDelay timeout) waitPing+    pingReply <- race (threadDelay timeout >> return TimedOut) waitPing     t1 <- getCurrentTime     return $ (t0, t1, pingReply) +-- | 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]) +-- | An HTTP2 response, once fully received, is made of headers and a payload.+type StreamResponse = (HPACK.HeaderList, ByteString)++-- | Wait for a stream until completion.+--+-- This function is fine if you don't want to consume results in chunks. waitStream :: Http2Stream -> IncomingFlowControl -> IO StreamResult waitStream stream streamFlowControl = do     (_,_,hdrs) <- _waitHeaders stream@@ -36,3 +59,18 @@         else do             _ <- _updateWindow $ streamFlowControl             moredata (x:xs)++-- | 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+    headers <- headersE+    chunks <- sequence chunksE+    return (headers, mconcat chunks)++-- | Sequentially wait for every push-promise with a handler.+--+-- This function runs forever and you should wrap it with 'withAsync'.+onPushPromise :: Http2Stream -> PushPromiseHandler -> IO ()+onPushPromise stream handler = forever $ do+    _waitPushPromise stream handler