diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright Lucas DiCioccio (c) 2017 + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above + copyright notice, this list of conditions and the following + disclaimer in the documentation and/or other materials provided + with the distribution. + + * Neither the name of Author name here nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,200 @@ +# http2-client + +An native-Haskell HTTP2 client library based on `http2` and `tls` packages. + +## General design + +HTTP2 is a heavy protocol. HTTP2 features pipelining, query and responses +interleaving, server-pushes, pings, stateful compression and flow-control, +priorization etc. This library aims at exposing these features so that library +users can integrate `http2-client` in a variety of applications. In short, we'd +like to expose as many HTTP2 features as possible. Hence, the `http2-client` +programming interface can feel low-level for users with expectations to get an +API as simple as in HTTP1.x. + +Exposing most HTTP2 primitives as a drawback: the library allows a client to +behave abnormally with-respect to the HTTP2 spec. That said, we try to prevent +notoriously-difficult errors such as concurrency bugs by coercing users with +the programming API following Haskell's philosophy to factor-out errors at +compile time. For instance, a client can send DATA frames on a stream after +closing it with a RST (easy to spot). However, a multi-threaded client will not +be able to interleave DATA frames with HEADERS and their CONTINUATIONs and the +locking required to achieve this invariant is hidden (hard to implement). + +Following this philosophy, we prefer to offer a somewhat low-level API in +`Network.HTTP2.Client` and higher-level APIs (with a different performance +trade-off) in `Network.HTTP2.Client.Helpers`. For instance, +`Network.HTTP2.Client.Helpers.waitStream` will consume a whole stream in memory +before returning whereas `Network.HTTP2.Client` users will have to take chunks +one at a time. We look forward to the linear arrows extension for improving the +library design. + +## Usage + +First, make sure you are somewhat familiar with HTTP and HTTP2 standards by +reading RFCs or Wikipedia pages. If you use the library, feel free to shoot me +an e-mail (cf. commits) or a tweet @lucasdicioccio . + +### Installation + +This package is a standard `Stack` project, please also refer to Stack's +documentation if you have trouble installing or using this package. + +### Opening a stream + +First, you open a (TLS-protected) connection to a server and configure the +initial SETTINGS to advertise. Then you can open and consume streams. Opening +streams takes a stream-definition and expresses two sequential parts. First, +sending the HTTP headers, which reserves an increasing stream-ID with the +server. Second, you consume a stream by sending DATA chunk or receiving DATA +chunks. One thing that can prevent concurrency is if you have too many opened +streams for the server. The `http2-client` library tracks server's max +concurrency preference and will prevent you from opening too many streams. + +### Sending chunked data + +Sent data must be chunked according to server's preferences. A function named +`sendData` performs the chunking but this chunking could have some suboptimal +overhead if you want to repeatedly call sendData with a buffer size that is not +a multiple of the server's preferred chunk size. + +### Flow control + +HTTP2 mandates a flow-control system that cannot be disabled. DATA chunks +consume credit from the flow-control system. The standard defines a +flow-control context per stream plus one global per-connection. + +** Received DATA flow control ** In order to keep receiving data you need to +periodically transfer credit to the server. One transfers credit to server by +calling `_updateWindow`, which transfers locally-accumulated credit (you +accumulate credit with `addCredit`). The current implementation already follows +a "zero-sum" credit where received DATA is immediately consumed and +re-credited. That is, if you only keep calling `_updateWindow` at some +frequency the stream will progress. You can also `_addCredit` to permit +receiving more DATA on a stream/connection (e.g., if you want to implement +something like TCP slow-start). + +** Sent DATA flow control ** A server following the HTTP2 specification +strictly will kick you for sending too much data. The `http2-client` library +allows you to be more aggressive than the server allows and you have to care +for your streams. We provide an incoming flow-control context that will allow +you to call `_withdrawCredit` to wait until some credit is available. At the +time of this writing, the `sendData` function does not call `_withdrawCredit` +and we provide no equivalent. Note that the chunking and flow-control +mechanisms have interesting interactions in HTTP2 in a multi-threaded context. +Pay attention to always take credit in the per-stream flow-control context +before taking it from the global per-connection flow-control context. +Otherwise, you risk starving the global per-connection flow-control with no +guarantee that you'll be allowed to send a DATA frame. + +## Settings changes + +The HTTP2 RFC acknowledges the inherent race conditions that may occur when +changing SETTINGS. The `http2-client` library should be rather permissive and +accept rather than reject frames caught violating inconsistent settings once +client settings are made stricter. Conversely, the `http2-client` library tries +to enforce server-SETTINGS strictly before ACKnowledging the setting changes. +This configuration can lead to problems if the server send more-permissive +SETTINGS (e.g., allowing a large default window size -> which recredits all +streams) but if the server applies this change locally only after receiving the +client ACK. One way to be double-sure the `http2-client` library is always +strict would be to apply settings changes in two steps: settings that move in +the "stricter direction" (e.g., fewer concurrency, smaller initial window) +should be applied _before_ ACK-ing the SETTING frame. Meanwhile settings that +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 +- (server prefs) immediately after receiving and hence before sending ACK-SETTINGS + +Fortunately, changing settings mid-stream is probably a rare behavior and the +default SETTINGS are large enough to avoid creating fatal errors before +sending/receiving the initial SETTINGS frames. + +## Help and examples + +We currently provide a command-line example client: `http2-client-exe` which I +use as a test client and you could use to test various flow-control parameters. +Read `app/Main.hs` and `src/Network/HTTP2/Client/Helpers.hs` to look at how a +user application is implemented. + +The Haddocks should have plenty implementation details, so please have a look. +Otherwise, you can ask help by creating an Issue on the bug-tracker. + +## Things that are hardcoded + +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 +- 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 + received + +## Contributing + +Contributions are welcome. As I start integrating external contribution I plan +to follow the following procedure: +- stop pushing directly into master +- develop any patch in a new branch, branched from master +- merge requests target master + +Please pay attention to the following: +- avoid introducing external dependencies, especially if dependencies are not in stackage +- avoid reformatting-only merge requests +- please verify that you can `stack clean` and `stack build --pedantic` + +General mindset to have during code-reviews: +- be kind +- be patient +- surpass egos and bring data if there is a disagreement + +## Bugtracker + +Most of the following points have their own issues on the issue tracker at +GitHub: https://github.com/lucasdicioccio/http2-client/issues . + +### Things that will likely change the API + +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 + +### Support of the HTTP2 standard + +The current implementation follows the HTTP2 standard except for the following: +- does not handle `PRIORITY` +- does not expose padding +- does not handle `SETTINGS_MAX_HEADER_LIST_SIZE` + * it's unclear to me whether this limitation is applied per frame or in total + * the accounting is done before compression with 32 extra bytes per header +- does not implement most of the checks that should trigger protocol errors + * unwanted frames on idle/closed streams + * increasing IDs only + * invalid settings + * invalid window in flow-control + * invalid frame sizes + * data-consumed out of flow-control limits + * authority of push promise https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-10.1 + * ... + +### 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 + * 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 + * currently using naive function 'Network.HPACK.encodeHeader' +- consider using a typeclass or a way to change the frame client +- we need a story for IO exceptions +- we need a story for instrumentation diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/app/Main.hs b/app/Main.hs new file mode 100644 --- /dev/null +++ b/app/Main.hs @@ -0,0 +1,174 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +module Main where + +import Control.Monad (forever, when, void) +import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent.Async (async, waitAnyCancel) +import Data.ByteString (ByteString) +import qualified Data.ByteString.Char8 as ByteString +import Data.Default.Class (def) +import Data.Time.Clock (diffUTCTime, getCurrentTime) +import qualified Network.HTTP2 as HTTP2 +import qualified Network.TLS as TLS +import qualified Network.TLS.Extra.Cipher as TLS +import Options.Applicative +import Data.Monoid ((<>)) +import System.IO + +import Network.HTTP2.Client +import Network.HTTP2.Client.Helpers + +type Path = ByteString +type Verb = ByteString + +data ServerPushSwitch = PushEnabled | PushDisabled + deriving Show + +data QueryArgs = QueryArgs { + _host :: !HostName + , _port :: !PortNumber + , _verb :: !Verb + , _path :: !Path + , _extraHeaders :: ![(ByteString, ByteString)] + , _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 + } deriving Show + +clientArgs :: Parser QueryArgs +clientArgs = + QueryArgs + <$> host + <*> port + <*> verb + <*> path + <*> extraHeaders + <*> milliseconds "inter-ping-delay-ms" 0 + <*> milliseconds "ping-timeout-ms" 5000 + <*> milliseconds "inter-flow-control-updates-ms" 1000 + <*> concurrency + <*> allowPush + <*> frameBytes + <*> headersBytes + <*> initialWindowBytes + <*> initialWindowKick + <*> numConcurrentThreads + <*> numQueriesPerThread + <*> milliseconds "delay-before-quitting-ms" 0 + <*> kthxByeMessage + <*> encoderBufSize + <*> decoderBufSize + where + bstrOption = fmap ByteString.pack . strOption + milliseconds what base = fmap (*1000) $ option auto (long what <> value base) + keyval kv = let (k,v1) = ByteString.break (== ':') kv in (k, ByteString.drop 1 v1) + + host = strOption (long "host" <> value "127.0.0.1") + path = bstrOption (long "path" <> value "/") + port = option auto (long "port" <> value 443) + verb = bstrOption (long "verb" <> value "GET") + extraHeaders = many (fmap keyval $ bstrOption (short 'H')) + concurrency = option auto (long "max-concurrency" <> value 100) + allowPush = flag PushEnabled PushDisabled (long "disable-server-push") + frameBytes = option auto (long "max-frame-size" <> value 1048576) + headersBytes = option auto (long "max-headers-list-size" <> value 1048576) + initialWindowBytes = option auto (long "initial-window-size" <> value 10485760) + initialWindowKick = option auto (long "initial-window-kick" <> value 0) + numConcurrentThreads = option auto (long "num-concurrent-threads" <> value 1) + numQueriesPerThread = option auto (long "num-queries-per-thread" <> value 1) + 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) + +main :: IO () +main = execParser opts >>= client + where + opts = info (helper <*> clientArgs) (mconcat [ + fullDesc + , header "http2-client-exe: a CLI HTTP2 client written in Haskell" + ]) + +client :: QueryArgs -> IO () +client QueryArgs{..} = do + hSetBuffering stdout LineBuffering + + let headersPairs = [ (":method", _verb) + , (":scheme", "https") + , (":path", _path) + , (":authority", ByteString.pack _host) + ] <> _extraHeaders + + let onPushPromise _ stream streamFlowControl _ = void $ forkIO $ do + timePrint ("push stream started" :: String) + waitStream stream streamFlowControl >>= timePrint + timePrint ("push stream ended" :: String) + + let conf = [ (HTTP2.SettingsMaxFrameSize, _settingsMaxFrameSize) + , (HTTP2.SettingsMaxConcurrentStreams, _settingsMaxConcurrency) + , (HTTP2.SettingsMaxHeaderBlockSize, _settingsMaxHeaderBlockSize) + , (HTTP2.SettingsInitialWindowSize, _settingsInitialWindowSize) + , (HTTP2.SettingsEnablePush, case _settingsAllowServerPush of + PushEnabled -> 1 ; PushDisabled -> 0) + ] + timePrint conf + + conn <- newHttp2Client _host _port _encoderBufsize _decoderBufsize tlsParams conf onPushPromise + _addCredit (_incomingFlowControl conn) _initialWindowKick + _ <- forkIO $ forever $ do + updated <- _updateWindow $ _incomingFlowControl conn + when updated $ timePrint ("sending flow-control update" :: String) + threadDelay _interFlowControlUpdates + + _ <- forkIO $ when (_interPingDelay > 0) $ forever $ do + threadDelay _interPingDelay + (t0, t1, pingReply) <- ping _pingTimeout "pingpong" conn + timePrint $ ("ping-reply:" :: String, pingReply, diffUTCTime t1 t0) + + let go 0 idx = timePrint $ "done worker: " <> show idx + go n idx = do + _ <- (_startStream conn $ \stream -> + let initStream = _headers stream headersPairs (HTTP2.setEndStream) + handler streamFlowControl _ = do + timePrint $ "stream started " <> show (idx, n) + waitStream stream streamFlowControl >>= timePrint + timePrint $ "stream ended " <> show (idx, n) + in StreamDefinition initStream handler) + go (n - 1) idx + + _ <- waitAnyCancel =<< traverse (async . go _numberQueries) [1 .. _concurrentQueriesCount] + + when (_finalDelay > 0) (threadDelay _finalDelay) + + _gtfo conn HTTP2.NoError _finalMessage + + return () + where + tlsParams = TLS.ClientParams { + TLS.clientWantSessionResume = Nothing + , TLS.clientUseMaxFragmentLength = Nothing + , TLS.clientServerIdentification = ("127.0.0.1", "") + , TLS.clientUseServerNameIndication = True + , TLS.clientShared = def + , TLS.clientHooks = def { TLS.onServerCertificate = \_ _ _ _ -> return [] + } + , TLS.clientSupported = def { TLS.supportedCiphers = TLS.ciphersuite_default } + , TLS.clientDebug = def + } + +timePrint :: Show a => a -> IO () +timePrint x = do + tst <- getCurrentTime + print (tst, x) diff --git a/http2-client.cabal b/http2-client.cabal new file mode 100644 --- /dev/null +++ b/http2-client.cabal @@ -0,0 +1,59 @@ +name: http2-client +version: 0.1.0.0 +synopsis: A native HTTP2 client library. +description: Please read the README.ms at the homepage. +homepage: https://github.com/lucasdicioccio/http2-client +license: BSD3 +license-file: LICENSE +author: Lucas DiCioccio +maintainer: lucas@dicioccio.fr +copyright: 2017 Lucas DiCioccio +category: Web +build-type: Simple +extra-source-files: README.md +cabal-version: >=1.10 + +library + hs-source-dirs: src + exposed-modules: Network.HTTP2.Client + , Network.HTTP2.Client.FrameConnection + , Network.HTTP2.Client.Helpers + , Network.HTTP2.Client.RawConnection + build-depends: base >= 4.7 && < 5 + , async + , bytestring + , connection + , containers + , http2 + , network + , time + , tls + default-language: Haskell2010 + +executable http2-client-exe + hs-source-dirs: app + main-is: Main.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: base + , async + , bytestring + , data-default-class + , http2 + , http2-client + , optparse-applicative + , time + , tls + default-language: Haskell2010 + +test-suite http2-client-test + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Spec.hs + build-depends: base + , http2-client + ghc-options: -threaded -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/lucasdicioccio/http2-client diff --git a/src/Network/HTTP2/Client.hs b/src/Network/HTTP2/Client.hs new file mode 100644 --- /dev/null +++ b/src/Network/HTTP2/Client.hs @@ -0,0 +1,726 @@ +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE OverloadedStrings #-} + +module Network.HTTP2.Client ( + Http2Client(..) + , newHttp2Client + , PushPromiseHandler + , Http2Stream(..) + , sendData + , StreamThread + , _gtfo + , StreamDefinition(..) + , IncomingFlowControl(..) + , OutgoingFlowControl(..) + , module Network.HTTP2.Client.FrameConnection + , module Network.Socket + , module Network.TLS + ) where + +import Control.Exception (bracket, throw) +import Control.Concurrent.Async (race) +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 Data.ByteString (ByteString) +import qualified Data.ByteString as ByteString +import Data.IORef (IORef, newIORef, atomicModifyIORef', readIORef) +import Data.Maybe (fromMaybe) +import Data.Monoid ((<>)) +import GHC.Exception (Exception) +import Network.HPACK as HPACK +import Network.HTTP2 as HTTP2 +import Network.Socket (HostName, PortNumber) +import Network.TLS (ClientParams) + +import Network.HTTP2.Client.FrameConnection + +-- | Offers credit-based flow-control. +-- +-- Any mutable changes are atomic and hence work as intended in a multithreaded +-- setup. +-- +-- The design of the flow-control mechanism is subject to changes. One +-- important thing to keep in mind with current implementation is that both the +-- connection and streams are credited with '_addCredit' as soon as DATA frames +-- arrive, hence no-need to account for the DATA frames (but you can account +-- for delay-bandwidth product for instance). +data IncomingFlowControl = IncomingFlowControl { + _addCredit :: WindowSize -> IO () + -- ^ Add credit (using a hidden mutable reference underneath). This function + -- only does accounting, the IO only does mutable changes. See '_updateWindow'. + , _consumeCredit :: WindowSize -> IO Int + -- ^ Consumes some credit and returns the credit left. + , _updateWindow :: IO Bool + -- ^ Sends a WINDOW_UPDATE frame crediting it with the whole amount credited + -- since the last _updateWindow call. The boolean tells whether an update was + -- actually sent or not. A reason for not sending an update is if there is no + -- credit in the flow-control system. + } + +-- | Receives credit-based flow-control or block. +-- +-- There is no way to observe the total amount of credit and receive/withdraw +-- are atomic hence this object is thread-safe. However we plan to propose an +-- STM-based API to allow withdrawing atomically from both the connection and a +-- per-stream 'OutgoingFlowControl' objects at a same time. Without such +-- atomicity one must ensure consumers do not exhaust the connection credit +-- before taking the per-stream credit (else they might prevent others sending +-- data without taking any). +-- +-- Longer term we plan to hide outgoing-flow-control increment/decrement +-- altogether because exception between withdrawing credit and sending DATA +-- could mean lost credit (and hence hanging streams). +data OutgoingFlowControl = OutgoingFlowControl { + _receiveCredit :: WindowSize -> IO () + -- ^ Add credit (using a hidden mutable reference underneath). + , _withdrawCredit :: WindowSize -> IO WindowSize + -- ^ Wait until we can take credit from stash. The returned value correspond + -- to the amount that could be withdrawn, which is min(current, wanted). A + -- caller should withdraw credit to send DATA chunks and put back any unused + -- credit with _receiveCredit. + } + +-- | Defines a client stream. +-- +-- Please red the doc for this record fields and then see 'StreamStarter'. +data StreamDefinition a = StreamDefinition { + _initStream :: IO StreamThread + -- ^ Function to initialize a new client stream. This function runs in a + -- exclusive-access section of the code and may prevent other threads to + -- initialize new streams. Hence, you should ensure this IO does not wait for + -- long periods of time. + , _handleStream :: IncomingFlowControl -> OutgoingFlowControl -> IO a + -- ^ Function to operate with the stream. IncomingFlowControl currently is + -- credited on your behalf as soon as a DATA frame arrives (and before you + -- handle it with '_waitData'). However we do not send WINDOW_UPDATE with + -- '_updateWindow'. This design may change in the future to give more leeway + -- to library users. + } + +-- | Type alias for callback-based functions starting new streams. +-- +-- The callback a user must provide takes an 'Http2Stream' and returns a +-- 'StreamDefinition'. This construction may seem wrong because a 'StreamDefinition' +-- contains an initialization and a handler functions. The explanation for this +-- twistedness is as follows: in HTTP2 stream-ids must be monotonically +-- increasing, if we want to support multi-threaded clients we need to +-- serialize access to a critical region of the code when clients send +-- HEADERS+CONTINUATIONs frames. +-- +-- Passing the 'Http2Stream' object as part of the callback avoids leaking the +-- implementation of the critical region, meanwhile, the 'StreamDefinition' +-- delimits this critical region. +type StreamStarter a = + (Http2Stream -> StreamDefinition a) -> IO (Either TooMuchConcurrency a) + +-- | Whether or not the client library believes the server will reject the new +-- stream. The Int content corresponds to the number of streams that should end +-- before accepting more streams. A reason this number can be more than zero is +-- that servers can change (and hence reduce) the advertised number of allowed +-- 'maxConcurrentStreams' at any time. +newtype TooMuchConcurrency = TooMuchConcurrency { _getStreamRoomNeeded :: Int } + deriving Show + +-- | Record holding functions one can call while in an HTTP2 client session. +data Http2Client = Http2Client { + _ping :: ByteString -> IO (IO (FrameHeader, FramePayload)) + -- ^ Send a PING, the payload size must be exactly eight bytes. + -- Returns an IO to wait for a ping reply. No timeout is provided. Only the + -- first call to this IO will return if a reply is received. Hence we + -- recommend wrapping this IO in an Async (e.g., with @race (threadDelay + -- timeout)@.) + , _settings :: SettingsList -> IO () + -- ^ Sends a SETTINGS. + , _goaway :: ErrorCodeId -> ByteString -> IO () + -- ^ Sends a GOAWAY. + , _startStream :: forall a. StreamStarter a + -- ^ Spawns new streams. See 'StreamStarter'. + , _incomingFlowControl :: IncomingFlowControl + -- ^ Simple getter for the 'IncomingFlowControl' for the whole client + -- connection. + , _outgoingFlowControl :: OutgoingFlowControl + -- ^ Simple getter for the 'OutgoingFlowControl' for the whole client + -- connection. + , _paylodSplitter :: IO PayloadSplitter + -- ^ Returns a function to split a payload. + } + +-- | Couples client and server settings together. +data ConnectionSettings = ConnectionSettings { + _clientSettings :: !Settings + , _serverSettings :: !Settings + } + +defaultConnectionSettings :: ConnectionSettings +defaultConnectionSettings = + ConnectionSettings defaultSettings defaultSettings + +-- | Synonym of '_goaway'. +-- +-- https://github.com/http2/http2-spec/pull/366 +_gtfo :: Http2Client -> ErrorCodeId -> ByteString -> IO () +_gtfo = _goaway + +-- | Opaque proof that a client stream was initialized. +-- +-- This type is only useful to force calling '_headers' in '_initStream' and +-- contains no information. +data StreamThread = CST + +-- | Record holding functions one can call while in an HTTP2 client stream. +data Http2Stream = Http2Stream { + _headers :: HPACK.HeaderList + -> (FrameFlags -> FrameFlags) + -> IO StreamThread + -- ^ Starts the stream with HTTP headers. Flags modifier can use + -- 'setEndStream' if no data is required passed the last block of headers. + -- Usually, this is the only call needed to build an '_initStream'. + , _prio :: Priority -> IO () + -- ^ 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. + , _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. + } + +-- | Handler upon receiving a PUSH_PROMISE from the server. +-- +-- The functions for 'Http2Stream' are similar to those used in ''. But callers +-- shall not use '_headers' to initialize the PUSH_PROMISE stream. Rather, +-- callers should 'waitHeaders' or '_rst' to reject the PUSH_PROMISE. +-- +-- 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 + +-- | Helper to carry around the HPACK encoder for outgoing header blocks.. +data HpackEncoderContext = HpackEncoderContext { + _encodeHeaders :: HeaderList -> IO HeaderBlockFragment + , _applySettings :: Int -> IO () + } + +-- | Starts a new Http2Client with a remote Host/Port. TLS ClientParams are +-- mandatory because we only support TLS-protected streams for now. +newHttp2Client :: HostName + -- ^ Host to connect to. + -> PortNumber + -- ^ Port number to connect to (usually 443 on the web). + -> Int + -- ^ The buffersize for the Network.HPACK encoder. + -> Int + -- ^ The buffersize for the Network.HPACK decoder. + -> ClientParams + -- ^ 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 + -- network connection + conn <- newHttp2FrameConnection host port tlsParams + + -- prepare hpack contexts + hpackEncoder <- do + let strategy = (HPACK.defaultEncodeStrategy { HPACK.useHuffman = True }) + dt <- HPACK.newDynamicTableForEncoding HPACK.defaultDynamicTableSize + let _encodeHeaders = HPACK.encodeHeader strategy encoderBufSize dt + let _applySettings n = HPACK.setLimitForEncoding n dt + return HpackEncoderContext{..} + + -- prepare client streams + clientStreamIdMutex <- newMVar 0 + let withClientStreamId h = bracket (takeMVar clientStreamIdMutex) + (putMVar clientStreamIdMutex . succ) + (\k -> h (2 * k + 1)) -- Note: client StreamIds MUST be odd + + let controlStream = makeFrameClientStream conn 0 + let ackPing = sendPingFrame controlStream HTTP2.setAck + let ackSettings = sendSettingsFrame controlStream HTTP2.setAck [] + + -- Initial thread receiving server frames. + maxReceivedStreamId <- newIORef 0 + serverFrames <- newChan + _ <- forkIO $ incomingFramesLoop conn serverFrames maxReceivedStreamId + + -- Thread handling control frames. + settings <- newIORef defaultConnectionSettings + controlFrames <- dupChan serverFrames + _ <- forkIO $ incomingControlFramesLoop controlFrames settings hpackEncoder ackPing ackSettings + + -- Thread handling push-promises and headers frames serializing the buffers. + serverStreamFrames <- dupChan serverFrames + serverHeaders <- newChan + hpackDecoder <- do + dt <- newDynamicTableForDecoding HPACK.defaultDynamicTableSize decoderBufSize + return dt + + creditFrames <- dupChan serverFrames + + _outgoingFlowControl <- newOutgoingFlowControl settings 0 creditFrames + _incomingFlowControl <- newIncomingFlowControl settings controlStream + + _ <- forkIO $ incomingHPACKFramesLoop serverStreamFrames + serverHeaders + hpackEncoder + hpackDecoder + conn + settings + handlePPStream + + dataFrames <- dupChan serverFrames + _ <- forkIO $ creditDataFramesLoop _incomingFlowControl dataFrames + + conccurentStreams <- newIORef 0 + let _startStream getWork = do + maxConcurrency <- fromMaybe 100 . maxConcurrentStreams . _serverSettings + <$> readIORef settings + roomNeeded <- atomicModifyIORef' conccurentStreams + (\n -> if n < maxConcurrency then (n + 1, 0) else (n, 1 + n - maxConcurrency)) + if roomNeeded > 0 + then + return $ Left $ TooMuchConcurrency roomNeeded + else do + cont <- withClientStreamId $ \sid -> do + initializeStream conn + settings + serverFrames + serverHeaders + hpackEncoder + sid + getWork + v <- cont + atomicModifyIORef' conccurentStreams (\n -> (n - 1, ())) + return $ Right v + + let _ping dat = do + -- Need to dupChan before sending the query to avoid missing a fast + -- answer if the network is fast. + pingFrames <- dupChan serverFrames + sendPingFrame controlStream id dat + return $ waitFrame (isPingReply dat) pingFrames + let _settings settslist = do + atomicModifyIORef' settings + (\(ConnectionSettings cli srv) -> + (ConnectionSettings (HTTP2.updateSettings cli settslist) srv, ())) + sendSettingsFrame controlStream id settslist + let _goaway err errStr = do + sId <- readIORef maxReceivedStreamId + sendGTFOFrame controlStream sId err errStr + + let _paylodSplitter = settingsPayloadSplitter <$> readIORef settings + + _settings initSettings + + return $ Http2Client{..} + +initializeStream + :: Exception e + => Http2FrameConnection + -> IORef ConnectionSettings + -> Chan (FrameHeader, Either e FramePayload) + -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList) + -> HpackEncoderContext + -> StreamId + -> (Http2Stream -> StreamDefinition a) + -> IO (IO a) +initializeStream conn settings serverFrames serverHeaders hpackEncoder sid getWork = do + let frameStream = makeFrameClientStream conn sid + + -- Register interest in frames. + frames <- dupChan serverFrames + credits <- dupChan serverFrames + headers <- dupChan serverHeaders + + -- Builds a flow-control context. + incomingStreamFlowControl <- newIncomingFlowControl settings frameStream + outgoingStreamFlowControl <- newOutgoingFlowControl settings sid credits + + -- Prepare handlers. + let _headers headersList flags = do + splitter <- settingsPayloadSplitter <$> readIORef settings + sendHeaders frameStream hpackEncoder headersList splitter flags + let _waitHeaders = waitHeadersWithStreamId sid headers + 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 + let _prio = sendPriorityFrame frameStream + + let streamActions = getWork $ Http2Stream{..} + + -- Perform the 1st action, the stream won't be idle anymore. + _ <- _initStream streamActions + + -- Returns 2nd action. + return $ _handleStream streamActions incomingStreamFlowControl outgoingStreamFlowControl + +incomingFramesLoop + :: Http2FrameConnection + -> Chan (FrameHeader, Either HTTP2Error FramePayload) + -> IORef StreamId + -> IO () +incomingFramesLoop conn frames maxReceivedStreamId = forever $ do + frame@(fh, _) <- next conn + -- Remember highest streamId. + atomicModifyIORef' maxReceivedStreamId (\n -> (max n (streamId fh), ())) + writeChan frames frame + +incomingControlFramesLoop + :: Exception e + => Chan (FrameHeader, Either e FramePayload) + -> IORef ConnectionSettings + -> HpackEncoderContext + -> (ByteString -> IO ()) + -> IO () + -> IO () +incomingControlFramesLoop frames settings hpackEncoder ackPing ackSettings = forever $ do + controlFrame@(fh, payload) <- waitFrameWithStreamId 0 frames + case payload of + (SettingsFrame settsList) + | not . testAck . flags $ fh -> do + atomicModifyIORef' settings + (\(ConnectionSettings cli srv) -> + (ConnectionSettings cli (HTTP2.updateSettings srv settsList), ())) + maybe (return ()) + (_applySettings hpackEncoder) + (lookup SettingsHeaderTableSize settsList) + ackSettings + | otherwise -> do + ignore "TODO: settings ack should be taken into account only after reception, we should return a waitSettingsAck in the _settings function" + (PingFrame pingMsg) + | not . testAck . flags $ fh -> + ackPing pingMsg + | otherwise -> do + ignore "PingFrame replies waited for in the requestor thread" + (WindowUpdateFrame _ ) -> + ignore "connection-wide WindowUpdateFrame waited for in OutgoingFlowControl threads" + _ -> putStrLn ("UNHANDLED frame: " <> show controlFrame) + + where + ignore :: String -> IO () + ignore _ = return () + +-- | We currently need a specific loop for crediting streams because a client +-- user may programmatically reset and stop listening for a stream and stop +-- calling waitData (which credits streams). +-- +-- TODO: modify the '_rst' function to wait and credit all the remaining data +-- that could have been sent in flight +creditDataFramesLoop + :: Exception e + => IncomingFlowControl + -> Chan (FrameHeader, Either e FramePayload) + -> IO () +creditDataFramesLoop flowControl frames = forever $ do + (fh,_) <- waitFrameWithTypeId [FrameData] frames + -- TODO: error if detect over-run, current implementation credits + -- everything back so that should never happen + _ <- _consumeCredit flowControl (HTTP2.payloadLength fh) + _addCredit flowControl (HTTP2.payloadLength fh) + +incomingHPACKFramesLoop + :: Exception e + => Chan (FrameHeader, Either e FramePayload) + -> Chan (FrameHeader, StreamId, Either ErrorCode HeaderList) + -> HpackEncoderContext + -> DynamicTable + -> Http2FrameConnection + -> IORef ConnectionSettings + -> PushPromiseHandler a + -> IO () +incomingHPACKFramesLoop frames headers hpackEncoder hpackDecoder conn settings handlePPStream = forever $ do + (fh, fp) <- waitFrameWithTypeId [ FrameRSTStream + , FramePushPromise + , FrameHeaders + ] + frames + (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 + return (sid, Right hbf) + HeadersFrame _ hbf -> -- TODO: handle priority + return (HTTP2.streamId fh, Right hbf) + RSTStreamFrame err -> + return (HTTP2.streamId fh, Left err) + _ -> error "wrong TypeId" + + let go curFh (Right buffer) = + if not $ HTTP2.testEndHeader (HTTP2.flags curFh) + then do + (lastFh, lastFp) <- waitFrameWithTypeId [ FrameRSTStream + , FrameContinuation + ] + frames + case lastFp of + ContinuationFrame chbf -> + go lastFh (Right (ByteString.append buffer chbf)) + RSTStreamFrame err -> + go lastFh (Left err) + _ -> error "waitFrameWithTypeIdForStreamId returned an unknown frame" + else do + hdrs <- decodeHeader hpackDecoder buffer + writeChan headers (curFh, sId, Right hdrs) + + go curFh (Left err) = + writeChan headers (curFh, sId, (Left $ HTTP2.fromErrorCodeId err)) + + go fh pattern + +newIncomingFlowControl + :: IORef ConnectionSettings + -> Http2FrameClientStream + -> IO IncomingFlowControl +newIncomingFlowControl settings stream = do + 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 + extra <- readIORef creditAdded + return $ base + extra - conso + let _updateWindow = do + base <- initialWindowSize . _clientSettings <$> readIORef settings + added <- readIORef creditAdded + consumed <- readIORef creditConsumed + + let transferred = min added (HTTP2.maxWindowSize - base + consumed) + let shouldUpdate = transferred > 0 + + _addCredit (negate transferred) + _ <- _consumeCredit (negate transferred) + + when shouldUpdate (sendWindowUpdateFrame stream transferred) + + return shouldUpdate + return $ IncomingFlowControl _addCredit _consumeCredit _updateWindow + +newOutgoingFlowControl :: + Exception e + => IORef ConnectionSettings + -> StreamId + -> Chan (FrameHeader, Either e FramePayload) + -> IO OutgoingFlowControl +newOutgoingFlowControl settings sid frames = do + credit <- newIORef 0 + let receive n = atomicModifyIORef' credit (\c -> (c + n, ())) + let withdraw n = do + base <- initialWindowSize . _serverSettings <$> readIORef settings + got <- atomicModifyIORef' credit (\c -> + if base + c >= n + then (c - n, n) + else (0 - base, base + c)) + if got > 0 + then return got + else do + amount <- race (waitSettingsChange base) waitSomeCredit + receive (either (const 0) id amount) + withdraw n + return $ OutgoingFlowControl receive withdraw + where + -- TODO: broadcast settings changes from ConnectionSettings using a better data type + -- than IORef+busy loop. Currently the busy loop is fine because + -- SettingsInitialWindowSize is typically set at the first frame and hence + -- waiting one second for an update that is likely to never come is + -- probably not an issue. There still is an opportunity risk, however, that + -- an hasted client asks for X > initialWindowSize before the server has + -- sent its initial SETTINGS frame. + waitSettingsChange prev = do + new <- initialWindowSize . _serverSettings <$> readIORef settings + 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" + +sendHeaders + :: Http2FrameClientStream + -> HpackEncoderContext + -> HeaderList + -> PayloadSplitter + -> (FrameFlags -> FrameFlags) + -> IO StreamThread +sendHeaders s enc headers blockSplitter flagmod = do + headerBlockFragments <- blockSplitter <$> _encodeHeaders enc headers + 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 + return CST + +-- | A function able to split a header block into multiple fragments. +type PayloadSplitter = ByteString -> [ByteString] + +-- | Split headers like so that no payload exceeds server's maxFrameSize. +settingsPayloadSplitter :: ConnectionSettings -> PayloadSplitter +settingsPayloadSplitter (ConnectionSettings _ srv) = + fixedSizeChunks (maxFrameSize srv) + +-- | Breaks a ByteString into fixed-sized chunks. +-- +-- @ fixedSizeChunks 2 "hello" = ["he", "ll", "o"] @ +fixedSizeChunks :: Int -> ByteString -> [ByteString] +fixedSizeChunks 0 _ = error "cannot chunk by zero-length blocks" +fixedSizeChunks _ "" = [] +fixedSizeChunks len bstr = + let + (chunk, rest) = ByteString.splitAt len bstr + in + chunk : fixedSizeChunks len rest + +sendData :: Http2Client -> Http2Stream -> (FrameFlags -> FrameFlags) -> ByteString -> IO () +sendData conn stream flagmod dat = do + splitter <- _paylodSplitter conn + let chunks = splitter dat + let pairs = reverse $ zip (flagmod : repeat id) (reverse chunks) + forM_ pairs $ \(flags, chunk) -> _sendDataChunk stream flags chunk + +sendDataFrame + :: Http2FrameClientStream + -> (FrameFlags -> FrameFlags) -> ByteString -> IO () +sendDataFrame s flagmod dat = do + sendOne s flagmod (DataFrame dat) + +sendResetFrame :: Http2FrameClientStream -> ErrorCodeId -> IO () +sendResetFrame s err = do + sendOne s id (RSTStreamFrame err) + +sendGTFOFrame + :: Http2FrameClientStream + -> StreamId -> ErrorCodeId -> ByteString -> IO () +sendGTFOFrame s lastStreamId err errStr = do + sendOne s id (GoAwayFrame lastStreamId err errStr) + +rfcError :: String -> a +rfcError msg = error (msg ++ "draft-ietf-httpbis-http2-17") + +sendPingFrame + :: Http2FrameClientStream + -> (FrameFlags -> FrameFlags) + -> ByteString + -> IO () +sendPingFrame s flags dat + | _getStreamId s /= 0 = + rfcError "PING frames are not associated with any individual stream." + | ByteString.length dat /= 8 = + rfcError "PING frames MUST contain 8 octets" + | otherwise = sendOne s flags (PingFrame dat) + +sendWindowUpdateFrame + :: Http2FrameClientStream -> WindowSize -> IO () +sendWindowUpdateFrame s amount = do + let payload = WindowUpdateFrame amount + sendOne s id payload + return () + +sendSettingsFrame + :: Http2FrameClientStream + -> (FrameFlags -> FrameFlags) -> SettingsList -> IO () +sendSettingsFrame s flags setts + | _getStreamId s /= 0 = + rfcError "The stream identifier for a SETTINGS frame MUST be zero (0x0)." + | otherwise = do + let payload = SettingsFrame setts + sendOne s flags payload + return () + +sendPriorityFrame :: Http2FrameClientStream -> Priority -> IO () +sendPriorityFrame s p = do + let payload = PriorityFrame p + sendOne s id payload + return () + +waitFrameWithStreamId + :: Exception e => + StreamId -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload) +waitFrameWithStreamId sid = waitFrame (\h _ -> streamId h == sid) + +waitFrameWithTypeId + :: (Exception e) + => [FrameTypeId] + -> Chan (FrameHeader, Either e FramePayload) -> IO (FrameHeader, FramePayload) +waitFrameWithTypeId tids = waitFrame (\_ p -> HTTP2.framePayloadToFrameTypeId p `elem` tids) + +waitFrameWithTypeIdForStreamId + :: (Exception e) + => StreamId + -> [FrameTypeId] + -> Chan (FrameHeader, Either e FramePayload) + -> IO (FrameHeader, FramePayload) +waitFrameWithTypeIdForStreamId sid tids = + waitFrame (\h p -> streamId h == sid && HTTP2.framePayloadToFrameTypeId p `elem` tids) + +waitFrame + :: Exception e + => (FrameHeader -> FramePayload -> Bool) + -> Chan (FrameHeader, Either e FramePayload) + -> IO (FrameHeader, FramePayload) +waitFrame test chan = + loop + where + loop = do + (fHead, fPayload) <- readChan chan + let dat = either throw id fPayload + if test fHead dat + then return (fHead, dat) + else loop + +isPingReply :: ByteString -> FrameHeader -> FramePayload -> Bool +isPingReply datSent _ (PingFrame datRcv) = datSent == datRcv +isPingReply _ _ _ = False + +waitHeadersWithStreamId + :: StreamId + -> Chan (FrameHeader, StreamId, t) + -> IO (FrameHeader, StreamId, t) +waitHeadersWithStreamId sid = + waitHeaders (\_ s _ -> s == sid) + +waitHeaders + :: (FrameHeader -> StreamId -> t -> Bool) + -> Chan (FrameHeader, StreamId, t) + -> IO (FrameHeader, StreamId, t) +waitHeaders test chan = + loop + where + loop = do + tuple@(fH, sId, headers) <- readChan chan + if test fH sId headers + then return tuple + else loop diff --git a/src/Network/HTTP2/Client/FrameConnection.hs b/src/Network/HTTP2/Client/FrameConnection.hs new file mode 100644 --- /dev/null +++ b/src/Network/HTTP2/Client/FrameConnection.hs @@ -0,0 +1,107 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RankNTypes #-} + +module Network.HTTP2.Client.FrameConnection ( + Http2FrameConnection + , newHttp2FrameConnection + -- * Interact at the Frame level. + , Http2FrameClientStream(..) + , makeFrameClientStream + , sendOne + , sendBackToBack + , next + , closeConnection + ) where + +import Control.Exception (bracket) +import Control.Concurrent.MVar (newMVar, takeMVar, putMVar) +import Control.Monad (void) +import Network.HTTP2 (FrameHeader(..), FrameFlags, FramePayload, HTTP2Error, encodeInfo, decodeFramePayload) +import qualified Network.HTTP2 as HTTP2 +import Network.Socket (HostName, PortNumber) +import qualified Network.TLS as TLS + +import Network.HTTP2.Client.RawConnection + +data Http2FrameConnection = Http2FrameConnection { + _makeFrameClientStream :: HTTP2.StreamId -> Http2FrameClientStream + -- ^ Starts a new client stream. + , _serverStream :: Http2ServerStream + -- ^ Receives frames from a server. + , _closeConnection :: IO () + -- ^ Function that will close the network connection. + } + +-- | Closes the Http2FrameConnection abruptly. +closeConnection :: Http2FrameConnection -> IO () +closeConnection = _closeConnection + +-- | Creates a client stream. +makeFrameClientStream :: Http2FrameConnection + -> HTTP2.StreamId + -> Http2FrameClientStream +makeFrameClientStream = _makeFrameClientStream + +data Http2FrameClientStream = Http2FrameClientStream { + _sendFrames :: [(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). + , _getStreamId :: HTTP2.StreamId -- TODO: hide me + } + +-- | Sends a frame to the server. +sendOne :: Http2FrameClientStream -> (FrameFlags -> FrameFlags) -> FramePayload -> IO () +sendOne client f payload = _sendFrames client [(f, payload)] + +-- | Sends multiple back-to-back frames to the server. +sendBackToBack :: Http2FrameClientStream -> [(FrameFlags -> FrameFlags, FramePayload)] -> IO () +sendBackToBack = _sendFrames + +data Http2ServerStream = Http2ServerStream { + _nextHeaderAndFrame :: IO (FrameHeader, Either HTTP2Error FramePayload) + } + +-- | Waits for the next frame from the server. +next :: Http2FrameConnection -> IO (FrameHeader, Either HTTP2Error FramePayload) +next = _nextHeaderAndFrame . _serverStream + +-- | Creates a new 'Http2FrameConnection' to a given host for a frame-to-frame communication. +newHttp2FrameConnection :: HostName + -> PortNumber + -> TLS.ClientParams + -> IO Http2FrameConnection +newHttp2FrameConnection host port params = do + -- Spawns an HTTP2 connection. + http2conn <- newRawHttp2Connection host port params + + -- Prepare a local mutex, this mutex should never escape the + -- function's scope. Else it might lead to bugs (e.g., + -- https://ro-che.info/articles/2014-07-30-bracket ) + writerMutex <- newMVar () + + let writeProtect io = + bracket (takeMVar writerMutex) (putMVar writerMutex) (const io) + + -- Define handlers. + let makeClientStream streamID = + let putFrame modifyFF frame = do + let info = encodeInfo modifyFF streamID + _sendRaw http2conn $ + HTTP2.encodeFrame info frame + putFrames xs = writeProtect . void $ traverse (uncurry putFrame) xs + in Http2FrameClientStream putFrames streamID + + nextServerFrameChunk = Http2ServerStream $ do + (fTy, fh@FrameHeader{..}) <- HTTP2.decodeFrameHeader <$> _nextRaw http2conn 9 + let decoder = decodeFramePayload fTy + -- TODO: consider splitting the iteration here to give a chance to + -- _not_ decode the frame, or consider lazyness enough. + let getNextFrame = decoder fh <$> _nextRaw http2conn payloadLength + nf <- getNextFrame + return (fh, nf) + + gtfo = _close http2conn + + return $ Http2FrameConnection makeClientStream nextServerFrameChunk gtfo diff --git a/src/Network/HTTP2/Client/Helpers.hs b/src/Network/HTTP2/Client/Helpers.hs new file mode 100644 --- /dev/null +++ b/src/Network/HTTP2/Client/Helpers.hs @@ -0,0 +1,38 @@ + +module Network.HTTP2.Client.Helpers where + +import Data.Time.Clock (UTCTime, getCurrentTime) +import qualified Network.HTTP2 as HTTP2 +import qualified Network.HPACK as HPACK +import Data.ByteString (ByteString) +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (race) + +import Network.HTTP2.Client + +type PingReply = (UTCTime, UTCTime, Either () (HTTP2.FrameHeader, HTTP2.FramePayload)) + +ping :: Int -> ByteString -> Http2Client -> IO PingReply +ping timeout msg conn = do + t0 <- getCurrentTime + waitPing <- _ping conn msg + pingReply <- race (threadDelay timeout) waitPing + t1 <- getCurrentTime + return $ (t0, t1, pingReply) + +type StreamResult = (Either HTTP2.ErrorCode HPACK.HeaderList, [Either HTTP2.ErrorCode ByteString]) + +waitStream :: Http2Stream -> IncomingFlowControl -> IO StreamResult +waitStream stream streamFlowControl = do + (_,_,hdrs) <- _waitHeaders stream + dataFrames <- moredata [] + return (hdrs, reverse dataFrames) + 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) diff --git a/src/Network/HTTP2/Client/RawConnection.hs b/src/Network/HTTP2/Client/RawConnection.hs new file mode 100644 --- /dev/null +++ b/src/Network/HTTP2/Client/RawConnection.hs @@ -0,0 +1,56 @@ +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE RankNTypes #-} + +module Network.HTTP2.Client.RawConnection ( + RawHttp2Connection (..) + , newRawHttp2Connection + ) where + +import Data.ByteString (ByteString) +import Network.Connection (connectTo, initConnectionContext, ConnectionParams(..), TLSSettings(..), connectionPut, connectionGetExact, connectionClose) +import qualified Network.HTTP2 as HTTP2 +import Network.Socket (HostName, PortNumber) +import qualified Network.TLS as TLS + + +-- TODO: catch connection errrors +data RawHttp2Connection = RawHttp2Connection { + _sendRaw :: ByteString -> IO () + -- ^ Function to send raw data to the server. + , _nextRaw :: Int -> IO ByteString + -- ^ Function to block reading a datachunk of a given size from the server. + , _close :: IO () + } + +-- | Initiates a RawHttp2Connection with a server. +-- +-- The current code does not handle closing the connexion, yikes. +newRawHttp2Connection :: HostName + -- ^ Server's hostname. + -> PortNumber + -- ^ Server's port to connect to. + -> TLS.ClientParams + -- ^ TLS parameters. The 'TLS.onSuggestALPN' hook is + -- overwritten to always return ["h2", "h2-17"]. + -> IO RawHttp2Connection +newRawHttp2Connection host port params = do + -- Connects to SSL. + ctx <- initConnectionContext + conn <- connectTo ctx connParams + + -- Define raw byte-stream handlers. + let putRaw dat = connectionPut conn dat + let getRaw amount = connectionGetExact conn amount + let doClose = connectionClose conn + + -- Initializes the HTTP2 stream. + putRaw HTTP2.connectionPreface + + return $ RawHttp2Connection putRaw getRaw doClose + where + overwrittenALPNHook = (TLS.clientHooks params) { + TLS.onSuggestALPN = return $ Just [ "h2", "h2-17" ] + } + modifiedParams = params { TLS.clientHooks = overwrittenALPNHook } + connParams = ConnectionParams host port (Just . TLSSettings $ modifiedParams) Nothing diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,2 @@ +main :: IO () +main = putStrLn "Test suite not yet implemented"