http2-client 0.3.0.0 → 0.3.0.1

raw patch · 3 files changed

+170/−49 lines, 3 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• app/Main.hs +93/−24
• http2-client.cabal +1/−1
• src/Network/HTTP2/Client.hs +76/−24

@@ -2,12 +2,13 @@ {-# LANGUAGE RecordWildCards   #-} module Main where -import           Control.Monad (forever, when, void) import           Control.Concurrent (forkIO, threadDelay) import           Control.Concurrent.Async (async, waitAnyCancel)+import           Control.Monad (forever, when, void) import           Data.ByteString (ByteString) import qualified Data.ByteString.Char8 as ByteString import           Data.Default.Class (def)+import           Data.Maybe (fromMaybe) import           Data.Time.Clock (diffUTCTime, getCurrentTime) import qualified Network.HTTP2 as HTTP2 import qualified Network.TLS as TLS@@ -32,28 +33,33 @@ data ServerPushSwitch = PushEnabled | PushDisabled   deriving Show +data Verbosity = Verbose | NonVerbose+  deriving Show+ data QueryArgs = QueryArgs {-    _host                    :: !HostName-  , _port                    :: !PortNumber-  , _verb                    :: !Verb-  , _path                    :: !Path-  , _extraHeaders            :: ![(ByteString, ByteString)]-  , _postData                :: !(Maybe PostData)-  , _interPingDelay          :: !Int-  , _pingTimeout             :: !Int-  , _interFlowControlUpdates :: !Int-  , _settingsMaxConcurrency  :: !Int-  , _settingsAllowServerPush :: !ServerPushSwitch+    _host                       :: !HostName+  , _port                       :: !PortNumber+  , _verb                       :: !Verb+  , _path                       :: !Path+  , _extraHeaders               :: ![(ByteString, ByteString)]+  , _postData                   :: !(Maybe PostData)+  , _interPingDelay             :: !Int+  , _pingTimeout                :: !Int+  , _interFlowControlUpdates    :: !Int+  , _settingsMaxConcurrency     :: !Int+  , _settingsAllowServerPush    :: !ServerPushSwitch   , _settingsMaxFrameSize       :: !Int   , _settingsMaxHeaderBlockSize :: !Int   , _settingsInitialWindowSize  :: !Int   , _initialWindowKick          :: !Int   , _concurrentQueriesCount     :: !Int   , _numberQueries              :: !Int-  , _finalDelay                :: !Int-  , _finalMessage              :: !ByteString-  , _encoderBufsize :: !Int-  , _decoderBufsize :: !Int+  , _finalDelay                 :: !Int+  , _finalMessage               :: !ByteString+  , _encoderBufsize             :: !Int+  , _decoderBufsize             :: !Int+  , _downloadPrefix             :: !FilePath+  , _verboseDebug               :: !Verbosity   } deriving Show  clientArgs :: Parser QueryArgs@@ -80,6 +86,8 @@         <*> kthxByeMessage         <*> encoderBufSize         <*> decoderBufSize+        <*> downloadPrefix+        <*> verboseDebug   where     bstrOption = fmap ByteString.pack . strOption     milliseconds what base = fmap (*1000) $ option auto (long what <> value base)@@ -102,6 +110,8 @@     kthxByeMessage = bstrOption (long "exit-greeting" <> value "kthxbye (>;_;<)")     encoderBufSize = option auto (long "hpack-encoder-buffer-size" <> value 4096)     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")  main :: IO () main = execParser opts >>= client@@ -136,9 +146,11 @@                           , (":authority", ByteString.pack _host)                           ] <> _extraHeaders -    let ppHandler _ stream ppHdrs streamFlowControl _ = void $ forkIO $ do-            timePrint ("push stream started" :: String, ppHdrs)-            waitStream stream streamFlowControl >>= timePrint . fromStreamResult+    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+            either (\e -> timePrint e) (dump PushPromiseFile pushpath n idx _downloadPrefix) ret             timePrint ("push stream ended" :: String)      let conf = [ (HTTP2.SettingsMaxFrameSize, _settingsMaxFrameSize)@@ -150,7 +162,32 @@                ]     timePrint conf -    conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf+    frameConn <- newHttp2FrameConnection _host _port tlsParams+    let wrappedFrameConn = frameConn {+            _makeFrameClientStream = \sid ->+                let frameClient = (_makeFrameClientStream frameConn) sid+                in frameClient {+                       _sendFrames = \xs -> do+                           print $ (">>> "::String, _getStreamId frameClient, map snd xs)+                           _sendFrames frameClient xs+                   }+          , _serverStream =+              let+                currentServerStrean = _serverStream frameConn+              in+                currentServerStrean {+                  _nextHeaderAndFrame = do+                      hdrFrame@(hdr,_) <- _nextHeaderAndFrame currentServerStrean+                      print ("<<< "::String, HTTP2.streamId hdr, hdrFrame)+                      return hdrFrame+                }+          }+    conn <- case _verboseDebug of+        Verbose ->+            wrapFrameClient wrappedFrameConn _encoderBufsize _decoderBufsize conf+        NonVerbose ->+            wrapFrameClient frameConn _encoderBufsize _decoderBufsize conf+    linkHttp2Client conn     _addCredit (_incomingFlowControl conn) _initialWindowKick     _ <- forkIO $ forever $ do             updated <- _updateWindow $ _incomingFlowControl conn@@ -165,15 +202,17 @@     let go 0 idx = timePrint $ "done worker: " <> show idx         go n idx = do             _ <- (withHttp2Stream conn $ \stream ->-                    let initStream = headers stream headersPairs headersFlags+                    let initStream = do+                            _ <- async $ onPushPromise stream (ppHandler n idx)+                            headers stream headersPairs headersFlags                         handler streamINFlowControl streamOUTFlowControl = do-                            _ <- async $ onPushPromise stream ppHandler                             timePrint $ "stream started " <> show (idx, n)                             _ <- async $ dataPostFunction conn                                                           (_outgoingFlowControl conn)                                                           stream                                                           streamOUTFlowControl-                            waitStream stream streamINFlowControl >>= timePrint . fromStreamResult+                            ret <-  fromStreamResult <$> waitStream stream streamINFlowControl+                            either (\e -> timePrint e) (dump MainFile _path n idx _downloadPrefix) ret                             timePrint $ "stream ended " <> show (idx, n)                     in StreamDefinition initStream handler)             go (n - 1) idx@@ -198,7 +237,37 @@         , TLS.clientDebug                = def         } +data DumpType = MainFile | PushPromiseFile++dump :: DumpType -> Path -> Int -> Int -> FilePath -> StreamResponse -> IO ()+dump MainFile _ _ _ ":none" (hdrs, _) = do+    timePrint hdrs+dump MainFile _ _ _ ":stdout" (hdrs, body) = do+    timePrint hdrs+    ByteString.putStrLn body+dump PushPromiseFile _ _ _ ":stdout" (hdrs, _) = do+    timePrint hdrs+dump MainFile _ _ _ ":stdout-pp" (hdrs, body) = do+    timePrint hdrs+    ByteString.putStrLn body+dump PushPromiseFile _ _ _ ":stdout-pp" (hdrs, body) = do+    timePrint hdrs+    ByteString.putStrLn body+dump _ querystring nquery nthread prefix (hdrs, body) = do+    timePrint hdrs+    ByteString.writeFile filepath body+  where+    filepath = mconcat [ prefix+                       , "/" +                       , show nquery+                       , "."+                       , show nthread+                       , ByteString.unpack $ cleanPath querystring+                       ]+    cleanPath = ByteString.takeWhile (/= '?')+              . ByteString.map (\c -> if c == '/' then '_' else c)+ timePrint :: Show a => a -> IO () timePrint x = do     tst <- getCurrentTime-    print (tst, x)+    ByteString.hPutStrLn stderr $ ByteString.pack $ show (tst, x)

@@ -1,5 +1,5 @@ name:                http2-client-version:             0.3.0.0+version:             0.3.0.1 synopsis:            A native HTTP2 client library. description:         Please read the README.md at the homepage. homepage:            https://github.com/lucasdicioccio/http2-client

@@ -24,8 +24,11 @@     -- * Flow control     , IncomingFlowControl(..)     , OutgoingFlowControl(..)+    -- * Exceptions+    , linkHttp2Client     -- * Misc.     , FlagSetter+    , Http2ClientAsyncs(..)     , wrapFrameClient     , _gtfo     -- * Convenience re-exports@@ -35,11 +38,11 @@     ) where  import           Control.Exception (bracket, throw)-import           Control.Concurrent.Async (race)+import           Control.Concurrent.Async (Async, async, race, link) import           Control.Concurrent.MVar (newMVar, takeMVar, putMVar)-import           Control.Concurrent (forkIO, threadDelay)+import           Control.Concurrent (threadDelay) import           Control.Concurrent.Chan (Chan, newChan, dupChan, readChan, writeChan)-import           Control.Monad (forever, void, when, forM_)+import           Control.Monad (forever, when, forM_) import           Data.ByteString (ByteString) import qualified Data.ByteString as ByteString import           Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef)@@ -165,8 +168,45 @@   -- connection.   , _paylodSplitter :: IO PayloadSplitter   -- ^ Returns a function to split a payload.+  , _asyncs         :: !Http2ClientAsyncs+  -- ^ Asynchronous operations threads.   } +-- | Set of Async threads running an Http2Client.+--+-- If you add other asyncs here, you should also modify+-- 'linkHttp2Client'.+data Http2ClientAsyncs = Http2ClientAsyncs {+    _waitSettingsAsync   :: Async (FrameHeader, FramePayload)+  -- ^ Async waiting for the initial settings ACK.+  , _creditFlowsAsync    :: Async ()+  -- ^ Async responsible for crediting the connection incoming flow control.+  -- See 'creditDataFramesLoop'.+  , _HPACKAsync          :: Async ()+  -- ^ Async responsible for serializing every HEADERS/PUSH_PROMISE/CONTINUATION.+  -- See 'incomingHPACKFramesLoop'.+  , _controlFramesAsync  :: Async ()+  -- ^ Async responsible for handling control frames (e.g., PING).+  -- See 'incomingControlFramesLoop'.+  , _incomingFramesAsync :: Async ()+  -- ^ Async responsible for ingesting all frames, increasing the+  -- maximum-received streamID and starting the frame dispatch. See+  -- 'incomingFramesLoop'.+  }++-- | Links all asyncs threads of a client to the current thread.+--+-- See 'link'.+linkHttp2Client :: Http2Client -> IO ()+linkHttp2Client client = +    let Http2ClientAsyncs{..} = _asyncs client+    in do+        link _waitSettingsAsync+        link _creditFlowsAsync+        link _HPACKAsync+        link _controlFramesAsync+        link _incomingFramesAsync+ -- | Couples client and server settings together. data ConnectionSettings = ConnectionSettings {     _clientSettings :: !Settings@@ -329,12 +369,12 @@     -- Initial thread receiving server frames.     maxReceivedStreamId  <- newIORef 0     serverFrames <- newChan-    _ <- forkIO $ incomingFramesLoop conn serverFrames maxReceivedStreamId+    aIncoming <- async $ incomingFramesLoop conn serverFrames maxReceivedStreamId      -- Thread handling control frames.     settings  <- newIORef defaultConnectionSettings     controlFrames <- dupChan serverFrames-    _ <- forkIO $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings+    aControl <- async $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings      -- Thread handling push-promises and headers frames serializing the buffers.     serverStreamFrames <- dupChan serverFrames@@ -350,13 +390,13 @@      serverPushPromises <- newChan -    _ <- forkIO $ incomingHPACKFramesLoop serverStreamFrames-                                          serverHeaders-                                          serverPushPromises-                                          hpackDecoder+    aHPACK <- async $ incomingHPACKFramesLoop serverStreamFrames+                                              serverHeaders+                                              serverPushPromises+                                              hpackDecoder      dataFrames <- dupChan serverFrames-    _ <- forkIO $ creditDataFramesLoop _incomingFlowControl dataFrames+    aCredit <- async $ creditDataFramesLoop _incomingFlowControl dataFrames      conccurentStreams <- newIORef 0     let _startStream getWork = do@@ -403,8 +443,10 @@      let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings -    _ <- forkIO . void =<< _settings initSettings+    aSettings <- async =<< _settings initSettings +    let _asyncs = Http2ClientAsyncs aSettings aCredit aHPACK aControl aIncoming+     return $ Http2Client{..}  initializeStream@@ -539,7 +581,8 @@   | OpenPushPromise !StreamId !StreamId  type FramesChan e = Chan (FrameHeader, Either e FramePayload)-type HeadersChan = Chan (FrameHeader, StreamId, Either ErrorCode HeaderList)+type HeadersChanContent = (FrameHeader, StreamId, Either ErrorCode HeaderList)+type HeadersChan = Chan HeadersChanContent type PushPromisesChanContent e = (StreamId, FramesChan e, HeadersChan, StreamId, HeaderList) type PushPromisesChan e = Chan (PushPromisesChanContent e) @@ -606,12 +649,15 @@   -> Http2FrameClientStream   -> IO IncomingFlowControl newIncomingFlowControl settings stream = do+    let getBase = if _getStreamId stream == 0+                  then return HTTP2.defaultInitialWindowSize+                  else initialWindowSize . _clientSettings <$> readIORef settings     creditAdded <- newIORef 0     creditConsumed <- newIORef 0     let _addCredit n = atomicModifyIORef' creditAdded (\c -> (c + n, ()))     let _consumeCredit n = do             conso <- atomicModifyIORef' creditConsumed (\c -> (c + n, c + n))-            base <- initialWindowSize . _clientSettings <$> readIORef settings+            base <- getBase             extra <- readIORef creditAdded             return $ base + extra - conso     let _updateWindow = do@@ -638,10 +684,13 @@   -> IO OutgoingFlowControl newOutgoingFlowControl settings sid frames = do     credit <- newIORef 0+    let getBase = if sid == 0+                  then return HTTP2.defaultInitialWindowSize+                  else initialWindowSize . _serverSettings <$> readIORef settings     let receive n = atomicModifyIORef' credit (\c -> (c + n, ()))     let withdraw 0 = return 0         withdraw n = do-            base <- initialWindowSize . _serverSettings <$> readIORef settings+            base <- getBase             got <- atomicModifyIORef' credit (\c ->                     if base + c >= n                     then (c - n, n)@@ -789,21 +838,24 @@     return ()  waitFrameWithStreamId-  :: Exception e =>-     StreamId -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload)+  :: Exception e+  => StreamId+  -> FramesChan e+  -> IO (FrameHeader, FramePayload) waitFrameWithStreamId sid = waitFrame (\h _ -> streamId h == sid)  waitFrameWithTypeId   :: (Exception e)   => [FrameTypeId]-  -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload)+  -> FramesChan e+  -> IO (FrameHeader, FramePayload) waitFrameWithTypeId tids = waitFrame (\_ p -> HTTP2.framePayloadToFrameTypeId p `elem` tids)  waitFrameWithTypeIdForStreamId   :: (Exception e)   => StreamId   -> [FrameTypeId]-  -> Chan (FrameHeader, Either e FramePayload)+  -> FramesChan e   -> IO (FrameHeader, FramePayload) waitFrameWithTypeIdForStreamId sid tids =     waitFrame (\h p -> streamId h == sid && HTTP2.framePayloadToFrameTypeId p `elem` tids)@@ -811,7 +863,7 @@ waitFrame   :: Exception e   => (FrameHeader -> FramePayload -> Bool)-  -> Chan (FrameHeader, Either e FramePayload)+  -> FramesChan e   -> IO (FrameHeader, FramePayload) waitFrame test chan =     loop@@ -833,15 +885,15 @@  waitHeadersWithStreamId   :: StreamId-  -> Chan (FrameHeader, StreamId, t)-  -> IO (FrameHeader, StreamId, t)+  -> HeadersChan+  -> IO HeadersChanContent waitHeadersWithStreamId sid =     waitHeaders (\_ s _ -> s == sid)  waitHeaders-  :: (FrameHeader -> StreamId -> t -> Bool)-  -> Chan (FrameHeader, StreamId, t)-  -> IO (FrameHeader, StreamId, t)+  :: (FrameHeader -> StreamId -> Either ErrorCode HeaderList -> Bool)+  -> HeadersChan+  -> IO HeadersChanContent waitHeaders test chan =     loop   where