{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE RecordWildCards #-}

module Network.HTTP2.Client.Run where

import Control.Concurrent
import Control.Concurrent.Async
import Control.Concurrent.STM
import qualified Control.Exception as E
import qualified Data.ByteString.UTF8 as UTF8
import Data.IORef
import Data.IP (IPv6)
import Network.Control (RxFlow (..), defaultMaxData)
import Network.HTTP.Semantics.Client
import Network.HTTP.Semantics.Client.Internal
import Network.HTTP.Semantics.IO
import Network.Socket (SockAddr)
import qualified System.ThreadManager as T
import Text.Read (readMaybe)

import Imports
import Network.HTTP2.Frame
import Network.HTTP2.H2
import Network.HTTP2.H2.OutBodyIface

-- | Client configuration
data ClientConfig = ClientConfig
    { ClientConfig -> Scheme
scheme :: Scheme
    -- ^ https or http
    , ClientConfig -> Authority
authority :: Authority
    -- ^ Server name
    , ClientConfig -> Int
cacheLimit :: Int
    -- ^ The maximum number of incoming streams on the net
    , ClientConfig -> Int
connectionWindowSize :: WindowSize
    -- ^ The window size of connection.
    , ClientConfig -> Settings
settings :: Settings
    -- ^ Settings
    }
    deriving (ClientConfig -> ClientConfig -> Bool
(ClientConfig -> ClientConfig -> Bool)
-> (ClientConfig -> ClientConfig -> Bool) -> Eq ClientConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ClientConfig -> ClientConfig -> Bool
== :: ClientConfig -> ClientConfig -> Bool
$c/= :: ClientConfig -> ClientConfig -> Bool
/= :: ClientConfig -> ClientConfig -> Bool
Eq, Int -> ClientConfig -> ShowS
[ClientConfig] -> ShowS
ClientConfig -> Authority
(Int -> ClientConfig -> ShowS)
-> (ClientConfig -> Authority)
-> ([ClientConfig] -> ShowS)
-> Show ClientConfig
forall a.
(Int -> a -> ShowS) -> (a -> Authority) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ClientConfig -> ShowS
showsPrec :: Int -> ClientConfig -> ShowS
$cshow :: ClientConfig -> Authority
show :: ClientConfig -> Authority
$cshowList :: [ClientConfig] -> ShowS
showList :: [ClientConfig] -> ShowS
Show)

-- | The default client config.
--
-- The @authority@ field will be used to set the HTTP2 @:authority@
-- pseudo-header. In most cases you will want to override it to be equal to
-- @host@.
--
-- Further background on @authority@:
-- [RFC 3986](https://datatracker.ietf.org/doc/html/rfc3986#section-3.2) also
-- allows @host:port@, and most servers will accept this too. However, when
-- using TLS, many servers will expect the TLS SNI server name and the
-- @:authority@ pseudo-header to be equal, and for TLS SNI the server name
-- should not include the port. Note that HTTP2 explicitly /disallows/ using
-- @userinfo\@@ as part of the authority.
--
-- >>> defaultClientConfig
-- ClientConfig {scheme = "http", authority = "localhost", cacheLimit = 64, connectionWindowSize = 16777216, settings = Settings {headerTableSize = 4096, enablePush = True, maxConcurrentStreams = Just 64, initialWindowSize = 262144, maxFrameSize = 16384, maxHeaderListSize = Just 65536, pingRateLimit = 10, emptyFrameRateLimit = 4, settingsRateLimit = 4, rstRateLimit = 4}}
defaultClientConfig :: ClientConfig
defaultClientConfig :: ClientConfig
defaultClientConfig =
    ClientConfig
        { scheme :: Scheme
scheme = Scheme
"http"
        , authority :: Authority
authority = Authority
"localhost"
        , cacheLimit :: Int
cacheLimit = Int
64
        , connectionWindowSize :: Int
connectionWindowSize = Int
defaultMaxData
        , settings :: Settings
settings = Settings
defaultSettings
        }

-- | Running HTTP/2 client.
run :: ClientConfig -> Config -> Client a -> IO a
run :: forall a. ClientConfig -> Config -> Client a -> IO a
run cconf :: ClientConfig
cconf@ClientConfig{Int
Authority
Scheme
Settings
scheme :: ClientConfig -> Scheme
authority :: ClientConfig -> Authority
cacheLimit :: ClientConfig -> Int
connectionWindowSize :: ClientConfig -> Int
settings :: ClientConfig -> Settings
scheme :: Scheme
authority :: Authority
cacheLimit :: Int
connectionWindowSize :: Int
settings :: Settings
..} Config
conf Client a
client = do
    Context
ctx <- ClientConfig -> Config -> IO Context
setup ClientConfig
cconf Config
conf
    Config -> Context -> IO a -> IO a
forall a. Config -> Context -> IO a -> IO a
runH2 Config
conf Context
ctx (IO a -> IO a) -> IO a -> IO a
forall a b. (a -> b) -> a -> b
$ Context -> IO a
runClient Context
ctx
  where
    serverMaxStreams :: Context -> IO Int
serverMaxStreams Context
ctx = do
        Maybe Int
mx <- Settings -> Maybe Int
maxConcurrentStreams (Settings -> Maybe Int) -> IO Settings -> IO (Maybe Int)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Settings -> IO Settings
forall a. IORef a -> IO a
readIORef (Context -> IORef Settings
peerSettings Context
ctx)
        case Maybe Int
mx of
            Maybe Int
Nothing -> Int -> IO Int
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
forall a. Bounded a => a
maxBound
            Just Int
x -> Int -> IO Int
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Int
x
    possibleClientStream :: Context -> IO Int
possibleClientStream Context
ctx = do
        Int
x <- Context -> IO Int
serverMaxStreams Context
ctx
        Int
n <- OddStreamTable -> Int
oddConc (OddStreamTable -> Int) -> IO OddStreamTable -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar OddStreamTable -> IO OddStreamTable
forall a. TVar a -> IO a
readTVarIO (Context -> TVar OddStreamTable
oddStreamTable Context
ctx)
        Int -> IO Int
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Int
x Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
n)
    aux :: Context -> Aux
aux Context
ctx =
        Aux
defaultAux
            { auxPossibleClientStreams = possibleClientStream ctx
            , auxSendPing =
                sendPing
                    ctx
                    False
                    "Haskell!" -- 8 bytes
            }
    clientCore :: Context -> Request -> (Response -> IO c) -> IO c
clientCore Context
ctx Request
req Response -> IO c
processResponse = Context -> IO c -> IO c
forall {c}. Context -> IO c -> IO c
counted Context
ctx (IO c -> IO c) -> IO c -> IO c
forall a b. (a -> b) -> a -> b
$ do
        (Stream
strm, Maybe OutObj
moutobj) <- Context
-> Scheme -> Authority -> Request -> IO (Stream, Maybe OutObj)
makeStream Context
ctx Scheme
scheme Authority
authority Request
req
        case Maybe OutObj
moutobj of
            Maybe OutObj
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
            Just OutObj
outobj -> Config -> Context -> Stream -> OutObj -> Bool -> IO ()
sendRequest Config
conf Context
ctx Stream
strm OutObj
outobj Bool
False
        -- Waiting for the response is in it too: a request interrupted
        -- there -- by 'timeout', say -- left its stream open just the same.
        (Stream -> IO Response
getResponse Stream
strm IO Response -> (Response -> IO c) -> IO c
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Response -> IO c
processResponse) IO c -> IO () -> IO c
forall a b. IO a -> IO b -> IO a
`E.finally` Context -> Stream -> IO ()
doneWithStream Context
ctx Stream
strm
    -- After a GOAWAY, the connection lasts as long as a request is in
    -- 'activeRequests' ('drained').
    counted :: Context -> IO c -> IO c
counted Context
ctx =
        IO () -> IO () -> IO c -> IO c
forall a b c. IO a -> IO b -> IO c -> IO c
E.bracket_
            (STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar Int -> (Int -> Int) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (Context -> TVar Int
activeRequests Context
ctx) (Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1))
            (STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar Int -> (Int -> Int) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (Context -> TVar Int
activeRequests Context
ctx) (Int -> Int -> Int
forall a. Num a => a -> a -> a
subtract Int
1))
    runClient :: Context -> IO a
runClient Context
ctx = Client a
client (Context -> Request -> (Response -> IO r) -> IO r
forall {c}. Context -> Request -> (Response -> IO c) -> IO c
clientCore Context
ctx) (Aux -> IO a) -> Aux -> IO a
forall a b. (a -> b) -> a -> b
$ Context -> Aux
aux Context
ctx

-- | Launching a receiver and a sender.
runIO :: ClientConfig -> Config -> (ClientIO -> IO (IO a)) -> IO a
runIO :: forall a. ClientConfig -> Config -> (ClientIO -> IO (IO a)) -> IO a
runIO cconf :: ClientConfig
cconf@ClientConfig{Int
Authority
Scheme
Settings
scheme :: ClientConfig -> Scheme
authority :: ClientConfig -> Authority
cacheLimit :: ClientConfig -> Int
connectionWindowSize :: ClientConfig -> Int
settings :: ClientConfig -> Settings
scheme :: Scheme
authority :: Authority
cacheLimit :: Int
connectionWindowSize :: Int
settings :: Settings
..} conf :: Config
conf@Config{Bool
Int
Buffer
SockAddr
Manager
Int -> IO Scheme
Int -> TokenHeaderTable -> IO ()
PositionReadMaker
Scheme -> IO ()
confWriteBuffer :: Buffer
confBufferSize :: Int
confSendAll :: Scheme -> IO ()
confReadN :: Int -> IO Scheme
confPositionReadMaker :: PositionReadMaker
confTimeoutManager :: Manager
confMySockAddr :: SockAddr
confPeerSockAddr :: SockAddr
confReadNTimeout :: Bool
confOnInformational :: Int -> TokenHeaderTable -> IO ()
confOnInformational :: Config -> Int -> TokenHeaderTable -> IO ()
confReadNTimeout :: Config -> Bool
confPeerSockAddr :: Config -> SockAddr
confMySockAddr :: Config -> SockAddr
confTimeoutManager :: Config -> Manager
confPositionReadMaker :: Config -> PositionReadMaker
confReadN :: Config -> Int -> IO Scheme
confSendAll :: Config -> Scheme -> IO ()
confBufferSize :: Config -> Int
confWriteBuffer :: Config -> Buffer
..} ClientIO -> IO (IO a)
action = do
    ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
activeRequests :: Context -> TVar Int
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
resetByMe :: Context -> IORef IntSet
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
..} <- ClientConfig -> Config -> IO Context
setup ClientConfig
cconf Config
conf
    let putB :: Scheme -> IO ()
putB Scheme
bs = ControlQ -> Control -> IO ()
enqueueControl ControlQ
controlQ (Control -> IO ()) -> Control -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe SettingsList -> [Scheme] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [Scheme
bs]
        putR :: Request -> IO (Int, Stream)
putR Request
req = do
            (Stream
strm, Maybe OutObj
moutobj) <- Context
-> Scheme -> Authority -> Request -> IO (Stream, Maybe OutObj)
makeStream Context
ctx Scheme
scheme Authority
authority Request
req
            case Maybe OutObj
moutobj of
                Maybe OutObj
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
                Just OutObj
outobj -> Config -> Context -> Stream -> OutObj -> Bool -> IO ()
sendRequest Config
conf Context
ctx Stream
strm OutObj
outobj Bool
True
            (Int, Stream) -> IO (Int, Stream)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Stream -> Int
streamNumber Stream
strm, Stream
strm)
        get :: Stream -> IO Response
get = Stream -> IO Response
getResponse
        create :: IO (Int, Stream)
create = Context -> IO (Int, Stream)
openOddStreamWait Context
ctx
    IO a
runClient <-
        ClientIO -> IO (IO a)
action (ClientIO -> IO (IO a)) -> ClientIO -> IO (IO a)
forall a b. (a -> b) -> a -> b
$ SockAddr
-> SockAddr
-> (Request -> IO (Int, Stream))
-> (Stream -> IO Response)
-> (Scheme -> IO ())
-> IO (Int, Stream)
-> ClientIO
ClientIO SockAddr
confMySockAddr SockAddr
confPeerSockAddr Request -> IO (Int, Stream)
putR Stream -> IO Response
get Scheme -> IO ()
putB IO (Int, Stream)
create
    Config -> Context -> IO a -> IO a
forall a. Config -> Context -> IO a -> IO a
runH2 Config
conf Context
ctx IO a
runClient

-- | Called once a request is done with its stream, however it ended: once
-- 'processResponse' is, or once waiting for the response was interrupted.
--
-- A response it did not read to the end left its stream open.  The server
-- went on sending the rest of the body, which nobody would read: it held
-- the stream's slot of the server's SETTINGS_MAX_CONCURRENT_STREAMS, and
-- the octets that came in were never given back to the connection window.
-- Enough such requests, or ones whose 'processResponse' threw, and new
-- requests waited for a slot, or the connection stalled.  So such a stream
-- is reset (CANCEL), and what was left of its body is given back.
--
-- A stream still idle is one whose request is still being sent: its
-- HEADERS may not have gone out yet, and a RST_STREAM ahead of them would
-- be one on an idle stream, a connection error (RFC 9113, section 5.1).
-- So that is left until the stream opens -- the request sent, or the
-- response begun -- or closes.
doneWithStream :: Context -> Stream -> IO ()
doneWithStream :: Context -> Stream -> IO ()
doneWithStream Context
ctx Stream
strm = do
    Bool
idle <- StreamState -> Bool
isIdle (StreamState -> Bool) -> IO StreamState -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar StreamState -> IO StreamState
forall a. TVar a -> IO a
readTVarIO (Stream -> TVar StreamState
streamState Stream
strm)
    if Bool
idle
        then ThreadManager -> Authority -> IO () -> IO ()
T.forkManaged (Context -> ThreadManager
threadManager Context
ctx) Authority
label (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
            STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar (Stream -> TVar StreamState
streamState Stream
strm) STM StreamState -> (StreamState -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
check (Bool -> STM ()) -> (StreamState -> Bool) -> StreamState -> STM ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Bool -> Bool
not (Bool -> Bool) -> (StreamState -> Bool) -> StreamState -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StreamState -> Bool
isIdle
            Context -> Stream -> IO ()
doneWithOpenStream Context
ctx Stream
strm
        else Context -> Stream -> IO ()
doneWithOpenStream Context
ctx Stream
strm
  where
    label :: Authority
label = Authority
"H2 done with stream " Authority -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> Authority
forall a. Show a => a -> Authority
show (Stream -> Int
streamNumber Stream
strm)

doneWithOpenStream :: Context -> Stream -> IO ()
doneWithOpenStream :: Context -> Stream -> IO ()
doneWithOpenStream Context
ctx Stream
strm = do
    Bool
cancelled <-
        Context -> Stream -> ClosedCode -> IO Bool
closeIfReceiving Context
ctx Stream
strm (ClosedCode -> IO Bool) -> ClosedCode -> IO Bool
forall a b. (a -> b) -> a -> b
$ SomeException -> ClosedCode
ResetByMe (SomeException -> ClosedCode) -> SomeException -> ClosedCode
forall a b. (a -> b) -> a -> b
$ CancelledStream -> SomeException
forall e. Exception e => e -> SomeException
E.toException CancelledStream
CancelledStream
    if Bool
cancelled
        then do
            ControlQ -> Control -> IO ()
enqueueControl (Context -> ControlQ
controlQ Context
ctx) (Control -> IO ()) -> Control -> IO ()
forall a b. (a -> b) -> a -> b
$
                Maybe SettingsList -> [Scheme] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [ErrorCode -> Int -> Scheme
resetFrame ErrorCode
Cancel (Int -> Scheme) -> Int -> Scheme
forall a b. (a -> b) -> a -> b
$ Stream -> Int
streamNumber Stream
strm]
            Context -> Stream -> IO ()
giveBackUnread Context
ctx Stream
strm
        else Context -> Stream -> IO ()
adjustRxWindow Context
ctx Stream
strm

getResponse :: Stream -> IO Response
getResponse :: Stream -> IO Response
getResponse Stream
strm = do
    Either SomeException InpObj
mRsp <- MVar (Either SomeException InpObj)
-> IO (Either SomeException InpObj)
forall a. MVar a -> IO a
takeMVar (MVar (Either SomeException InpObj)
 -> IO (Either SomeException InpObj))
-> MVar (Either SomeException InpObj)
-> IO (Either SomeException InpObj)
forall a b. (a -> b) -> a -> b
$ Stream -> MVar (Either SomeException InpObj)
streamInput Stream
strm
    case Either SomeException InpObj
mRsp of
        Left SomeException
err -> SomeException -> IO Response
forall e a. Exception e => e -> IO a
E.throwIO SomeException
err
        Right InpObj
rsp -> Response -> IO Response
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Response -> IO Response) -> Response -> IO Response
forall a b. (a -> b) -> a -> b
$ InpObj -> Response
Response InpObj
rsp

setup :: ClientConfig -> Config -> IO Context
setup :: ClientConfig -> Config -> IO Context
setup ClientConfig{Int
Authority
Scheme
Settings
scheme :: ClientConfig -> Scheme
authority :: ClientConfig -> Authority
cacheLimit :: ClientConfig -> Int
connectionWindowSize :: ClientConfig -> Int
settings :: ClientConfig -> Settings
scheme :: Scheme
authority :: Authority
cacheLimit :: Int
connectionWindowSize :: Int
settings :: Settings
..} conf :: Config
conf@Config{Bool
Int
Buffer
SockAddr
Manager
Int -> IO Scheme
Int -> TokenHeaderTable -> IO ()
PositionReadMaker
Scheme -> IO ()
confOnInformational :: Config -> Int -> TokenHeaderTable -> IO ()
confReadNTimeout :: Config -> Bool
confPeerSockAddr :: Config -> SockAddr
confMySockAddr :: Config -> SockAddr
confTimeoutManager :: Config -> Manager
confPositionReadMaker :: Config -> PositionReadMaker
confReadN :: Config -> Int -> IO Scheme
confSendAll :: Config -> Scheme -> IO ()
confBufferSize :: Config -> Int
confWriteBuffer :: Config -> Buffer
confWriteBuffer :: Buffer
confBufferSize :: Int
confSendAll :: Scheme -> IO ()
confReadN :: Int -> IO Scheme
confPositionReadMaker :: PositionReadMaker
confTimeoutManager :: Manager
confMySockAddr :: SockAddr
confPeerSockAddr :: SockAddr
confReadNTimeout :: Bool
confOnInformational :: Int -> TokenHeaderTable -> IO ()
..} = do
    let clientInfo :: RoleInfo
clientInfo = Scheme -> Authority -> RoleInfo
newClientInfo Scheme
scheme Authority
authority
    Context
ctx <-
        RoleInfo
-> Config
-> Int
-> Int
-> Settings
-> Manager
-> Maybe (STM Bool)
-> IO Context
newContext
            RoleInfo
clientInfo
            Config
conf
            Int
cacheLimit
            Int
connectionWindowSize
            Settings
settings
            Manager
confTimeoutManager
            Maybe (STM Bool)
forall a. Maybe a
Nothing
    Context -> IO ()
exchangeSettings Context
ctx
    Context -> IO Context
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Context
ctx

runH2 :: Config -> Context -> IO a -> IO a
runH2 :: forall a. Config -> Context -> IO a -> IO a
runH2 Config
conf Context
ctx IO a
runClient = do
    ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a
forall a.
ThreadManager -> IO a -> (Maybe SomeException -> IO ()) -> IO a
stopAfterAndWait ThreadManager
mgr (IO a -> IO (Either SomeException a)
forall e a. Exception e => IO a -> IO (Either e a)
E.try IO a
runAll IO (Either SomeException a)
-> (Either SomeException a -> IO a) -> IO a
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Config -> Context -> Either SomeException a -> IO a
forall a. Config -> Context -> Either SomeException a -> IO a
closureClient Config
conf Context
ctx) ((Maybe SomeException -> IO ()) -> IO a)
-> (Maybe SomeException -> IO ()) -> IO a
forall a b. (a -> b) -> a -> b
$ \Maybe SomeException
res ->
        TVar OddStreamTable
-> TVar EvenStreamTable -> Maybe SomeException -> IO ()
closeAllStreams (Context -> TVar OddStreamTable
oddStreamTable Context
ctx) (Context -> TVar EvenStreamTable
evenStreamTable Context
ctx) Maybe SomeException
res
  where
    mgr :: ThreadManager
mgr = Context -> ThreadManager
threadManager Context
ctx
    runReceiver :: IO SomeException
runReceiver = Context -> Config -> IO SomeException
frameReceiver Context
ctx Config
conf
    runSender :: IO SomeException
runSender = Context -> Config -> IO SomeException
frameSender Context
ctx Config
conf
    runClientReceiver :: IO a
runClientReceiver = do
        Authority -> IO ()
labelMe Authority
"H2 ClientReceiver"
        IO SomeException -> (Async SomeException -> IO a) -> IO a
forall a b. IO a -> (Async a -> IO b) -> IO b
withAsync IO SomeException
runReceiver ((Async SomeException -> IO a) -> IO a)
-> (Async SomeException -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async SomeException
ar ->
            IO a -> (Async a -> IO a) -> IO a
forall a b. IO a -> (Async a -> IO b) -> IO b
withAsync IO a
runClient ((Async a -> IO a) -> IO a) -> (Async a -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async a
ac -> do
                Either SomeException a
er <- Async SomeException -> Async a -> IO (Either SomeException a)
forall a b. Async a -> Async b -> IO (Either a b)
waitEither Async SomeException
ar Async a
ac
                case Either SomeException a
er of
                    Right a
r -> a -> IO a
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return a
r
                    Left SomeException
err -> do
                        Bool
goingaway <- Maybe Int -> Bool
forall a. Maybe a -> Bool
isJust (Maybe Int -> Bool) -> IO (Maybe Int) -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar (Maybe Int) -> IO (Maybe Int)
forall a. TVar a -> IO a
readTVarIO (Context -> TVar (Maybe Int)
peerGoAway Context
ctx)
                        case SomeException -> Maybe HTTP2Error
forall e. Exception e => SomeException -> Maybe e
E.fromException SomeException
err of
                            -- The connection has run its course after the
                            -- server's GOAWAY.  The client function is let
                            -- finish, rather than killed with it: every
                            -- request it makes from here is refused, and
                            -- one still waiting is failed now.
                            Just HTTP2Error
ConnectionIsClosed
                                | Bool
goingaway -> do
                                    TVar OddStreamTable
-> TVar EvenStreamTable -> Maybe SomeException -> IO ()
closeAllStreams
                                        (Context -> TVar OddStreamTable
oddStreamTable Context
ctx)
                                        (Context -> TVar EvenStreamTable
evenStreamTable Context
ctx)
                                        (SomeException -> Maybe SomeException
forall a. a -> Maybe a
Just SomeException
err)
                                    Async a -> IO a
forall a. Async a -> IO a
wait Async a
ac
                            Maybe HTTP2Error
_ -> SomeException -> IO a
forall e a. Exception e => e -> IO a
E.throwIO SomeException
err

    -- When 'runClientReceiver' terminates, it is important we give the sender
    -- a chance to terminate cleanly also (it's possible the client terminated
    -- but there are still some messages in the queue to be sent).
    --
    -- If the client terminated successfully, we ignore any other errors in the
    -- sender (indeed, any exception here might simply be that the background
    -- threads were cancelled /because/ the client terminated).
    --
    -- If the sender terminates first, it failed, and no request can go out
    -- any more: the client is stopped and the sender's error reported, rather
    -- than the client left waiting on a connection nothing sends on.
    runAll :: IO a
runAll =
        IO SomeException -> (Async SomeException -> IO a) -> IO a
forall a b. IO a -> (Async a -> IO b) -> IO b
withAsync IO SomeException
runSender ((Async SomeException -> IO a) -> IO a)
-> (Async SomeException -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async SomeException
as ->
            IO a -> (Async a -> IO a) -> IO a
forall a b. IO a -> (Async a -> IO b) -> IO b
withAsync IO a
runClientReceiver ((Async a -> IO a) -> IO a) -> (Async a -> IO a) -> IO a
forall a b. (a -> b) -> a -> b
$ \Async a
ac -> do
                Either SomeException a
r <- Async SomeException -> Async a -> IO (Either SomeException a)
forall a b. Async a -> Async b -> IO (Either a b)
waitEither Async SomeException
as Async a
ac
                case Either SomeException a
r of
                    Right a
x -> Async SomeException -> IO SomeException
forall a. Async a -> IO a
wait Async SomeException
as IO SomeException -> IO a -> IO a
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> a -> IO a
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return a
x
                    Left SomeException
e -> do
                        -- The sender also finishes, normally, as soon as the
                        -- receiver is done and the queues are empty, and may
                        -- get there before the client side is seen to.  Only
                        -- with the receiver still running did it fail.
                        Maybe SomeException
done <- TVar (Maybe SomeException) -> IO (Maybe SomeException)
forall a. TVar a -> IO a
readTVarIO (TVar (Maybe SomeException) -> IO (Maybe SomeException))
-> TVar (Maybe SomeException) -> IO (Maybe SomeException)
forall a b. (a -> b) -> a -> b
$ Context -> TVar (Maybe SomeException)
receiverDone Context
ctx
                        case Maybe SomeException
done of
                            Just SomeException
_ -> Async a -> IO a
forall a. Async a -> IO a
wait Async a
ac
                            Maybe SomeException
Nothing -> SomeException -> IO a
forall e a. Exception e => e -> IO a
E.throwIO SomeException
e

makeStream
    :: Context
    -> Scheme
    -> Authority
    -> Request
    -> IO (Stream, Maybe OutObj)
makeStream :: Context
-> Scheme -> Authority -> Request -> IO (Stream, Maybe OutObj)
makeStream ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} Scheme
scheme Authority
auth (Request OutObj
req) = do
    -- Checking push promises
    let hdr0 :: [Header]
hdr0 = OutObj -> [Header]
outObjHeaders OutObj
req
        method :: Scheme
method = Scheme -> Maybe Scheme -> Scheme
forall a. a -> Maybe a -> a
fromMaybe (Authority -> Scheme
forall a. HasCallStack => Authority -> a
error Authority
"makeStream:method") (Maybe Scheme -> Scheme) -> Maybe Scheme -> Scheme
forall a b. (a -> b) -> a -> b
$ HeaderName -> [Header] -> Maybe Scheme
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup HeaderName
":method" [Header]
hdr0
        path :: Scheme
path = Scheme -> Maybe Scheme -> Scheme
forall a. a -> Maybe a -> a
fromMaybe (Authority -> Scheme
forall a. HasCallStack => Authority -> a
error Authority
"makeStream:path") (Maybe Scheme -> Scheme) -> Maybe Scheme -> Scheme
forall a b. (a -> b) -> a -> b
$ HeaderName -> [Header] -> Maybe Scheme
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup HeaderName
":path" [Header]
hdr0
    Maybe Stream
mstrm0 <- TVar EvenStreamTable -> Scheme -> Scheme -> IO (Maybe Stream)
lookupEvenCache TVar EvenStreamTable
evenStreamTable Scheme
method Scheme
path
    case Maybe Stream
mstrm0 of
        Just Stream
strm0 -> do
            TVar EvenStreamTable -> Scheme -> Scheme -> IO ()
deleteEvenCache TVar EvenStreamTable
evenStreamTable Scheme
method Scheme
path
            (Stream, Maybe OutObj) -> IO (Stream, Maybe OutObj)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Stream
strm0, Maybe OutObj
forall a. Maybe a
Nothing)
        Maybe Stream
Nothing -> do
            -- Arch/Sender is originally implemented for servers where
            -- the ordering of responses can be out-of-order.
            -- But for clients, the ordering must be maintained.
            -- To implement this, 'outputQStreamID' is used.
            let isIPv6 :: Bool
isIPv6 = Maybe IPv6 -> Bool
forall a. Maybe a -> Bool
isJust (Authority -> Maybe IPv6
forall a. Read a => Authority -> Maybe a
readMaybe Authority
auth :: Maybe IPv6)
                auth' :: Scheme
auth'
                    | Bool
isIPv6 = Scheme
"[" Scheme -> Scheme -> Scheme
forall a. Semigroup a => a -> a -> a
<> Authority -> Scheme
UTF8.fromString Authority
auth Scheme -> Scheme -> Scheme
forall a. Semigroup a => a -> a -> a
<> Scheme
"]"
                    | Bool
otherwise = Authority -> Scheme
UTF8.fromString Authority
auth
            let hdr1, hdr2 :: [Header]
                hdr1 :: [Header]
hdr1
                    | Scheme
scheme Scheme -> Scheme -> Bool
forall a. Eq a => a -> a -> Bool
/= Scheme
"" = (HeaderName
":scheme", Scheme
scheme) Header -> [Header] -> [Header]
forall a. a -> [a] -> [a]
: [Header]
hdr0
                    | Bool
otherwise = [Header]
hdr0
                hdr2 :: [Header]
hdr2
                    | Authority
auth Authority -> Authority -> Bool
forall a. Eq a => a -> a -> Bool
/= Authority
"" = (HeaderName
":authority", Scheme
auth') Header -> [Header] -> [Header]
forall a. a -> [a] -> [a]
: [Header]
hdr1
                    | Bool
otherwise = [Header]
hdr1
                req' :: OutObj
req' = OutObj
req{outObjHeaders = hdr2}
            -- FLOW CONTROL: SETTINGS_MAX_CONCURRENT_STREAMS: send: respecting peer's limit
            (Int
_sid, Stream
newstrm) <- Context -> IO (Int, Stream)
openOddStreamWait Context
ctx
            IORef (Maybe Scheme) -> Maybe Scheme -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef (Stream -> IORef (Maybe Scheme)
streamRequestMethod Stream
newstrm) (Maybe Scheme -> IO ()) -> Maybe Scheme -> IO ()
forall a b. (a -> b) -> a -> b
$ Scheme -> Maybe Scheme
forall a. a -> Maybe a
Just Scheme
method
            (Stream, Maybe OutObj) -> IO (Stream, Maybe OutObj)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Stream
newstrm, OutObj -> Maybe OutObj
forall a. a -> Maybe a
Just OutObj
req')

sendRequest :: Config -> Context -> Stream -> OutObj -> Bool -> IO ()
sendRequest :: Config -> Context -> Stream -> OutObj -> Bool -> IO ()
sendRequest Config{Bool
Int
Buffer
SockAddr
Manager
Int -> IO Scheme
Int -> TokenHeaderTable -> IO ()
PositionReadMaker
Scheme -> IO ()
confOnInformational :: Config -> Int -> TokenHeaderTable -> IO ()
confReadNTimeout :: Config -> Bool
confPeerSockAddr :: Config -> SockAddr
confMySockAddr :: Config -> SockAddr
confTimeoutManager :: Config -> Manager
confPositionReadMaker :: Config -> PositionReadMaker
confReadN :: Config -> Int -> IO Scheme
confSendAll :: Config -> Scheme -> IO ()
confBufferSize :: Config -> Int
confWriteBuffer :: Config -> Buffer
confWriteBuffer :: Buffer
confBufferSize :: Int
confSendAll :: Scheme -> IO ()
confReadN :: Int -> IO Scheme
confPositionReadMaker :: PositionReadMaker
confTimeoutManager :: Manager
confMySockAddr :: SockAddr
confPeerSockAddr :: SockAddr
confReadNTimeout :: Bool
confOnInformational :: Int -> TokenHeaderTable -> IO ()
..} ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} Stream
strm OutObj{[Header]
OutBody
TrailersMaker
outObjHeaders :: OutObj -> [Header]
outObjHeaders :: [Header]
outObjBody :: OutBody
outObjTrailers :: TrailersMaker
outObjTrailers :: OutObj -> TrailersMaker
outObjBody :: OutObj -> OutBody
..} Bool
io = do
    let sid :: Int
sid = Stream -> Int
streamNumber Stream
strm
    (Maybe DynaNext
mnext, Maybe (TBQueue StreamingChunk)
mtbq) <- (IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO () -> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a b. IO a -> IO b -> IO a
`E.onException` Int -> IO ()
abandon Int
sid) (IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
 -> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk)))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a b. (a -> b) -> a -> b
$ case OutBody
outObjBody of
        OutBody
OutBodyNone -> (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe DynaNext
forall a. Maybe a
Nothing, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyFile (FileSpec Authority
path FileOffset
fileoff FileOffset
bytecount) -> do
            (PositionRead
pread, Sentinel
sentinel) <- PositionReadMaker
confPositionReadMaker Authority
path
            let next :: DynaNext
next = PositionRead -> FileOffset -> FileOffset -> Sentinel -> DynaNext
fillFileBodyGetNext PositionRead
pread FileOffset
fileoff FileOffset
bytecount Sentinel
sentinel
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyBuilder Builder
builder -> do
            let next :: DynaNext
next = Builder -> DynaNext
fillBuilderBodyGetNext Builder
builder
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing)
        OutBodyStreaming (Builder -> IO ()) -> IO () -> IO ()
strmbdy -> do
            TBQueue StreamingChunk
q <- Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming Context
ctx Stream
strm ((OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk))
-> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
forall a b. (a -> b) -> a -> b
$ \OutBodyIface
iface ->
                OutBodyIface -> forall x. IO x -> IO x
outBodyUnmask OutBodyIface
iface (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ (Builder -> IO ()) -> IO () -> IO ()
strmbdy (OutBodyIface -> Builder -> IO ()
outBodyPush OutBodyIface
iface) (OutBodyIface -> IO ()
outBodyFlush OutBodyIface
iface)
            let next :: DynaNext
next = TBQueue StreamingChunk -> DynaNext
nextForStreaming TBQueue StreamingChunk
q
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, TBQueue StreamingChunk -> Maybe (TBQueue StreamingChunk)
forall a. a -> Maybe a
Just TBQueue StreamingChunk
q)
        OutBodyStreamingIface OutBodyIface -> IO ()
strmbdy -> do
            TBQueue StreamingChunk
q <- Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming Context
ctx Stream
strm OutBodyIface -> IO ()
strmbdy
            let next :: DynaNext
next = TBQueue StreamingChunk -> DynaNext
nextForStreaming TBQueue StreamingChunk
q
            (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
-> IO (Maybe DynaNext, Maybe (TBQueue StreamingChunk))
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (DynaNext -> Maybe DynaNext
forall a. a -> Maybe a
Just DynaNext
next, TBQueue StreamingChunk -> Maybe (TBQueue StreamingChunk)
forall a. a -> Maybe a
Just TBQueue StreamingChunk
q)
    let ot :: OutputType
ot = [Header] -> Maybe DynaNext -> TrailersMaker -> OutputType
OHeader [Header]
outObjHeaders Maybe DynaNext
mnext TrailersMaker
outObjTrailers
    if Bool
io
        then do
            let out :: Output
out = Context
-> Stream -> Maybe (TBQueue StreamingChunk) -> OutputType -> Output
makeOutputIO Context
ctx Stream
strm Maybe (TBQueue StreamingChunk)
mtbq OutputType
ot
            Int -> Output -> IO ()
pushOutput Int
sid Output
out IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`E.onException` Int -> IO ()
abandon Int
sid
        else do
            (IO Sync
pop, Output
out) <- Stream -> OutputType -> IO (IO Sync, Output)
makeOutput Stream
strm OutputType
ot
            Int -> Output -> IO ()
pushOutput Int
sid Output
out IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`E.onException` Int -> IO ()
abandon Int
sid
            LoopCheck
lc <- Stream
-> Maybe (TBQueue StreamingChunk) -> Maybe Handle -> IO LoopCheck
newLoopCheck Stream
strm Maybe (TBQueue StreamingChunk)
mtbq Maybe Handle
forall a. Maybe a
Nothing
            ThreadManager -> Authority -> IO () -> IO ()
T.forkManaged ThreadManager
threadManager Authority
label (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Context -> IO Sync -> LoopCheck -> IO ()
syncWithSender' Context
ctx IO Sync
pop LoopCheck
lc
  where
    label :: Authority
label = Authority
"H2 request sender for stream " Authority -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> Authority
forall a. Show a => a -> Authority
show (Stream -> Int
streamNumber Stream
strm)
    pushOutput :: Int -> Output -> IO ()
pushOutput Int
sid Output
out = STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
        Int
sidOK <- TVar Int -> STM Int
forall a. TVar a -> STM a
readTVar TVar Int
outputQStreamID
        Bool -> STM ()
check (Int
sidOK Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
sid)
        TVar Int -> Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar Int
outputQStreamID (Int
sid Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
2)
        TQueue Output -> Output -> STM ()
enqueueOutputSTM TQueue Output
outputQ Output
out
    -- The request failed before it was queued -- the file of a
    -- 'requestFile' could not be opened, say, or the thread was killed while
    -- waiting for its turn.  Its stream id was taken but nothing went out on
    -- it, and requests go out in stream id order: 'pushOutput' waits for
    -- 'outputQStreamID' to reach its own id.  Left as it was, that turn never
    -- came, so every later request waited for ever, and the stream held its
    -- concurrency slot.  So the stream is taken out of the table, and a
    -- thread passes its turn on once it arrives; the id goes unused, which a
    -- later, higher one closes implicitly (RFC 9113, section 5.1.1).
    abandon :: Int -> IO ()
abandon Int
sid = do
        Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
strm ClosedCode
Killed
        ThreadManager -> Authority -> IO () -> IO ()
T.forkManaged ThreadManager
threadManager (Authority
"H2 skipping stream " Authority -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> Authority
forall a. Show a => a -> Authority
show Int
sid) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
            STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                Int
sidOK <- TVar Int -> STM Int
forall a. TVar a -> STM a
readTVar TVar Int
outputQStreamID
                Bool -> STM ()
check (Int
sidOK Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
sid)
                TVar Int -> Int -> STM ()
forall a. TVar a -> a -> STM ()
writeTVar TVar Int
outputQStreamID (Int
sid Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
2)

sendStreaming
    :: Context
    -> Stream
    -> (OutBodyIface -> IO ())
    -> IO (TBQueue StreamingChunk)
sendStreaming :: Context
-> Stream -> (OutBodyIface -> IO ()) -> IO (TBQueue StreamingChunk)
sendStreaming ctx :: Context
ctx@Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} Stream
strm OutBodyIface -> IO ()
strmbdy = do
    TBQueue StreamingChunk
tbq <- Natural -> IO (TBQueue StreamingChunk)
forall a. Natural -> IO (TBQueue a)
newTBQueueIO Natural
10 -- fixme: hard coding: 10
    ThreadManager
-> Authority -> ((forall x. IO x -> IO x) -> IO ()) -> IO ()
T.forkManagedUnmask ThreadManager
threadManager Authority
label (((forall x. IO x -> IO x) -> IO ()) -> IO ())
-> ((forall x. IO x -> IO x) -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \forall x. IO x -> IO x
unmask ->
        Context
-> Stream
-> TBQueue StreamingChunk
-> (forall x. IO x -> IO x)
-> (OutBodyIface -> IO ())
-> IO ()
forall r.
Context
-> Stream
-> TBQueue StreamingChunk
-> (forall x. IO x -> IO x)
-> (OutBodyIface -> IO r)
-> IO r
withOutBodyIface Context
ctx Stream
strm TBQueue StreamingChunk
tbq IO a -> IO a
forall x. IO x -> IO x
unmask OutBodyIface -> IO ()
strmbdy
    TBQueue StreamingChunk -> IO (TBQueue StreamingChunk)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return TBQueue StreamingChunk
tbq
  where
    label :: Authority
label = Authority
"H2 request streaming sender for stream " Authority -> ShowS
forall a. [a] -> [a] -> [a]
++ Int -> Authority
forall a. Show a => a -> Authority
show (Stream -> Int
streamNumber Stream
strm)

exchangeSettings :: Context -> IO ()
exchangeSettings :: Context -> IO ()
exchangeSettings Context{TVar Int
TVar (Maybe Int)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef Int
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
Int -> TokenHeaderTable -> IO ()
peerSettings :: Context -> IORef Settings
oddStreamTable :: Context -> TVar OddStreamTable
activeRequests :: Context -> TVar Int
resetByMe :: Context -> IORef IntSet
peerGoAway :: Context -> TVar (Maybe Int)
informationalCallback :: Context -> Int -> TokenHeaderTable -> IO ()
appDone :: Context -> STM Bool
receiverDone :: Context -> TVar (Maybe SomeException)
timeoutManager :: Context -> Manager
threadManager :: Context -> ThreadManager
peerSockAddr :: Context -> SockAddr
mySockAddr :: Context -> SockAddr
rstRate :: Context -> Rate
emptyFrameRate :: Context -> Rate
settingsRate :: Context -> Rate
pingRate :: Context -> Rate
rxFlow :: Context -> IORef RxFlow
txFlow :: Context -> TVar TxFlow
decodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable :: Context -> DynamicTable
controlQ :: Context -> ControlQ
outputQStreamID :: Context -> TVar Int
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef Int
peerLastStreamId :: Context -> IORef Int
peerStreamId :: Context -> IORef Int
myStreamId :: Context -> TVar Int
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar Int
peerStreamId :: IORef Int
peerLastStreamId :: IORef Int
outputBufferLimit :: IORef Int
outputQ :: TQueue Output
outputQStreamID :: TVar Int
controlQ :: ControlQ
encodeDynamicTable :: DynamicTable
decodeDynamicTable :: DynamicTable
txFlow :: TVar TxFlow
rxFlow :: IORef RxFlow
pingRate :: Rate
settingsRate :: Rate
emptyFrameRate :: Rate
rstRate :: Rate
mySockAddr :: SockAddr
peerSockAddr :: SockAddr
threadManager :: ThreadManager
timeoutManager :: Manager
receiverDone :: TVar (Maybe SomeException)
appDone :: STM Bool
informationalCallback :: Int -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe Int)
activeRequests :: TVar Int
resetByMe :: IORef IntSet
..} = do
    Int
connRxWS <- RxFlow -> Int
rxfBufSize (RxFlow -> Int) -> IO RxFlow -> IO Int
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef RxFlow -> IO RxFlow
forall a. IORef a -> IO a
readIORef IORef RxFlow
rxFlow
    let frames :: [Scheme]
frames = Settings -> Int -> [Scheme]
makeNegotiationFrames Settings
mySettings Int
connRxWS
        setframe :: Control
setframe = Maybe SettingsList -> [Scheme] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing (Scheme
connectionPreface Scheme -> [Scheme] -> [Scheme]
forall a. a -> [a] -> [a]
: [Scheme]
frames)
    IORef Bool -> Bool -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef Bool
myFirstSettings Bool
True
    ControlQ -> Control -> IO ()
enqueueControl ControlQ
controlQ Control
setframe

data ClientIO = ClientIO
    { ClientIO -> SockAddr
cioMySockAddr :: SockAddr
    , ClientIO -> SockAddr
cioPeerSockAddr :: SockAddr
    , ClientIO -> Request -> IO (Int, Stream)
cioWriteRequest :: Request -> IO (StreamId, Stream)
    , ClientIO -> Stream -> IO Response
cioReadResponse :: Stream -> IO Response
    , ClientIO -> Scheme -> IO ()
cioWriteBytes :: ByteString -> IO ()
    , ClientIO -> IO (Int, Stream)
cioCreateStream :: IO (StreamId, Stream)
    }