{-# LANGUAGE MultiWayIf #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE RecordWildCards #-}

module Network.HTTP2.H2.Sync (
    LoopCheck (..),
    newLoopCheck,
    syncWithSender,
    syncWithSender',
    makeOutput,
    makeOutputIO,
    enqueueOutputSIO,
) where

import Control.Concurrent
import Control.Concurrent.STM
import Control.Monad
import Network.Control
import Network.HTTP.Semantics.IO
import qualified System.ThreadManager as T
import qualified System.TimeManager as TM

import Network.HTTP2.H2.Context
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Types

syncWithSender
    :: Context
    -> Stream
    -> OutputType
    -> LoopCheck
    -> IO ()
syncWithSender :: Context -> Stream -> OutputType -> LoopCheck -> IO ()
syncWithSender ctx :: Context
ctx@Context{TVar WindowSize
TVar (Maybe WindowSize)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef WindowSize
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
WindowSize -> TokenHeaderTable -> IO ()
role :: Role
roleInfo :: RoleInfo
mySettings :: Settings
myFirstSettings :: IORef Bool
peerSettings :: IORef Settings
oddStreamTable :: TVar OddStreamTable
evenStreamTable :: TVar EvenStreamTable
continued :: IORef (Maybe HeaderContinuation)
myStreamId :: TVar WindowSize
peerStreamId :: IORef WindowSize
peerLastStreamId :: IORef WindowSize
outputBufferLimit :: IORef WindowSize
outputQ :: TQueue Output
outputQStreamID :: TVar WindowSize
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 :: WindowSize -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe WindowSize)
activeRequests :: TVar WindowSize
resetByMe :: IORef IntSet
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar WindowSize
peerGoAway :: Context -> TVar (Maybe WindowSize)
informationalCallback :: Context -> WindowSize -> 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 WindowSize
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef WindowSize
peerLastStreamId :: Context -> IORef WindowSize
peerStreamId :: Context -> IORef WindowSize
myStreamId :: Context -> TVar WindowSize
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
myFirstSettings :: Context -> IORef Bool
mySettings :: Context -> Settings
roleInfo :: Context -> RoleInfo
role :: Context -> Role
..} Stream
strm OutputType
otyp LoopCheck
lc = do
    (IO Sync
pop, Output
out) <- Stream -> OutputType -> IO (IO Sync, Output)
makeOutput Stream
strm OutputType
otyp
    TQueue Output -> Output -> IO ()
enqueueOutput TQueue Output
outputQ Output
out
    Context -> IO Sync -> LoopCheck -> IO ()
syncWithSender' Context
ctx IO Sync
pop LoopCheck
lc

makeOutput :: Stream -> OutputType -> IO (IO Sync, Output)
makeOutput :: Stream -> OutputType -> IO (IO Sync, Output)
makeOutput Stream
strm OutputType
otyp = do
    MVar Sync
var <- IO (MVar Sync)
forall a. IO (MVar a)
newEmptyMVar
    let push :: Maybe Output -> IO ()
push Maybe Output
mout = case Maybe Output
mout of
            Maybe Output
Nothing -> MVar Sync -> Sync -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar Sync
var Sync
Done
            Just Output
ot -> MVar Sync -> Sync -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar Sync
var (Sync -> IO ()) -> Sync -> IO ()
forall a b. (a -> b) -> a -> b
$ Output -> Sync
Cont Output
ot
        pop :: IO Sync
pop = MVar Sync -> IO Sync
forall a. MVar a -> IO a
takeMVar MVar Sync
var
        out :: Output
out =
            Output
                { outputStream :: Stream
outputStream = Stream
strm
                , outputType :: OutputType
outputType = OutputType
otyp
                , outputSync :: Maybe Output -> IO ()
outputSync = Maybe Output -> IO ()
push
                }
    (IO Sync, Output) -> IO (IO Sync, Output)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (IO Sync
pop, Output
out)

-- | An output for the 'runIO' interfaces, which have no thread waiting to
-- put the rest of a body back on the queue.
--
-- The rest used to go back at once, whatever the stream's window.  With
-- none left, the sender filled a DATA frame into no room; a file read into
-- no room reads 0 octets, which is the end of the file, so a body larger than
-- the window went out cut short with END_STREAM.  A streaming body with
-- nothing queued made the sender spin instead.  So the rest goes back once
-- it can go on, the way 'syncWithSender'' does it for the other interfaces;
-- only when it has to wait is a thread used for it.
makeOutputIO
    :: Context -> Stream -> Maybe (TBQueue StreamingChunk) -> OutputType -> Output
makeOutputIO :: Context
-> Stream -> Maybe (TBQueue StreamingChunk) -> OutputType -> Output
makeOutputIO Context{TVar WindowSize
TVar (Maybe WindowSize)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef WindowSize
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
WindowSize -> TokenHeaderTable -> IO ()
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar WindowSize
peerGoAway :: Context -> TVar (Maybe WindowSize)
informationalCallback :: Context -> WindowSize -> 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 WindowSize
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef WindowSize
peerLastStreamId :: Context -> IORef WindowSize
peerStreamId :: Context -> IORef WindowSize
myStreamId :: Context -> TVar WindowSize
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
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 WindowSize
peerStreamId :: IORef WindowSize
peerLastStreamId :: IORef WindowSize
outputBufferLimit :: IORef WindowSize
outputQ :: TQueue Output
outputQStreamID :: TVar WindowSize
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 :: WindowSize -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe WindowSize)
activeRequests :: TVar WindowSize
resetByMe :: IORef IntSet
..} Stream
strm Maybe (TBQueue StreamingChunk)
mtbq OutputType
otyp = Output
out
  where
    push :: Maybe Output -> IO ()
push Maybe Output
mout = case Maybe Output
mout of
        Maybe Output
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
        Just Output
ot -> do
            Maybe Bool
now <- STM (Maybe Bool) -> IO (Maybe Bool)
forall a. STM a -> IO a
atomically (STM (Maybe Bool) -> IO (Maybe Bool))
-> STM (Maybe Bool) -> IO (Maybe Bool)
forall a b. (a -> b) -> a -> b
$ (Bool -> Maybe Bool
forall a. a -> Maybe a
Just (Bool -> Maybe Bool) -> STM Bool -> STM (Maybe Bool)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> STM Bool
ready) STM (Maybe Bool) -> STM (Maybe Bool) -> STM (Maybe Bool)
forall a. STM a -> STM a -> STM a
`orElse` Maybe Bool -> STM (Maybe Bool)
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Maybe Bool
forall a. Maybe a
Nothing
            case Maybe Bool
now of
                Just Bool
True -> TQueue Output -> Output -> IO ()
enqueueOutput TQueue Output
outputQ Output
ot
                Just Bool
False -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
                Maybe Bool
Nothing ->
                    ThreadManager -> String -> IO () -> IO ()
T.forkManaged ThreadManager
threadManager String
"H2 output waiting for its window" (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                        Bool
ok <- STM Bool -> IO Bool
forall a. STM a -> IO a
atomically STM Bool
ready
                        Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
ok (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ TQueue Output -> Output -> IO ()
enqueueOutput TQueue Output
outputQ Output
ot
    ready :: STM Bool
ready = Stream -> Maybe (TBQueue StreamingChunk) -> STM Bool
readyToContinue Stream
strm Maybe (TBQueue StreamingChunk)
mtbq
    out :: Output
out =
        Output
            { outputStream :: Stream
outputStream = Stream
strm
            , outputType :: OutputType
outputType = OutputType
otyp
            , outputSync :: Maybe Output -> IO ()
outputSync = Maybe Output -> IO ()
push
            }

-- | Whether the rest of a stream's body can go on: waiting while the
-- stream's window is shut or a streaming body has nothing queued, and 'False'
-- once the stream is closed.
readyToContinue :: Stream -> Maybe (TBQueue StreamingChunk) -> STM Bool
readyToContinue :: Stream -> Maybe (TBQueue StreamingChunk) -> STM Bool
readyToContinue Stream{TVar StreamState
streamState :: TVar StreamState
streamState :: Stream -> TVar StreamState
streamState, TVar TxFlow
streamTxFlow :: TVar TxFlow
streamTxFlow :: Stream -> TVar TxFlow
streamTxFlow} Maybe (TBQueue StreamingChunk)
mtbq = do
    StreamState
state <- TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar TVar StreamState
streamState
    case StreamState
state of
        Closed{} -> Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
        StreamState
_ -> do
            Maybe (TBQueue StreamingChunk) -> STM ()
forall a. Maybe (TBQueue a) -> STM ()
waitStreaming' Maybe (TBQueue StreamingChunk)
mtbq
            TVar TxFlow -> STM ()
waitStreamWindowSizeSTM TVar TxFlow
streamTxFlow
            Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
True

enqueueOutputSIO :: Context -> Stream -> OutputType -> IO ()
enqueueOutputSIO :: Context -> Stream -> OutputType -> IO ()
enqueueOutputSIO ctx :: Context
ctx@Context{TVar WindowSize
TVar (Maybe WindowSize)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef WindowSize
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
WindowSize -> TokenHeaderTable -> IO ()
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar WindowSize
peerGoAway :: Context -> TVar (Maybe WindowSize)
informationalCallback :: Context -> WindowSize -> 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 WindowSize
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef WindowSize
peerLastStreamId :: Context -> IORef WindowSize
peerStreamId :: Context -> IORef WindowSize
myStreamId :: Context -> TVar WindowSize
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
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 WindowSize
peerStreamId :: IORef WindowSize
peerLastStreamId :: IORef WindowSize
outputBufferLimit :: IORef WindowSize
outputQ :: TQueue Output
outputQStreamID :: TVar WindowSize
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 :: WindowSize -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe WindowSize)
activeRequests :: TVar WindowSize
resetByMe :: IORef IntSet
..} Stream
strm OutputType
otyp = do
    let out :: Output
out = Context
-> Stream -> Maybe (TBQueue StreamingChunk) -> OutputType -> Output
makeOutputIO Context
ctx Stream
strm Maybe (TBQueue StreamingChunk)
forall a. Maybe a
Nothing OutputType
otyp
    TQueue Output -> Output -> IO ()
enqueueOutput TQueue Output
outputQ Output
out

syncWithSender' :: Context -> IO Sync -> LoopCheck -> IO ()
syncWithSender' :: Context -> IO Sync -> LoopCheck -> IO ()
syncWithSender' Context{TVar WindowSize
TVar (Maybe WindowSize)
TVar (Maybe SomeException)
TVar TxFlow
TVar EvenStreamTable
TVar OddStreamTable
IORef Bool
IORef WindowSize
IORef (Maybe HeaderContinuation)
IORef IntSet
IORef RxFlow
IORef Settings
STM Bool
SockAddr
Rate
TQueue Output
Manager
ThreadManager
DynamicTable
Settings
ControlQ
RoleInfo
Role
WindowSize -> TokenHeaderTable -> IO ()
resetByMe :: Context -> IORef IntSet
activeRequests :: Context -> TVar WindowSize
peerGoAway :: Context -> TVar (Maybe WindowSize)
informationalCallback :: Context -> WindowSize -> 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 WindowSize
outputQ :: Context -> TQueue Output
outputBufferLimit :: Context -> IORef WindowSize
peerLastStreamId :: Context -> IORef WindowSize
peerStreamId :: Context -> IORef WindowSize
myStreamId :: Context -> TVar WindowSize
continued :: Context -> IORef (Maybe HeaderContinuation)
evenStreamTable :: Context -> TVar EvenStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
peerSettings :: Context -> IORef Settings
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 WindowSize
peerStreamId :: IORef WindowSize
peerLastStreamId :: IORef WindowSize
outputBufferLimit :: IORef WindowSize
outputQ :: TQueue Output
outputQStreamID :: TVar WindowSize
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 :: WindowSize -> TokenHeaderTable -> IO ()
peerGoAway :: TVar (Maybe WindowSize)
activeRequests :: TVar WindowSize
resetByMe :: IORef IntSet
..} IO Sync
pop LoopCheck
lc = IO ()
loop
  where
    loop :: IO ()
loop = do
        Sync
s <- IO Sync
pop
        case Sync
s of
            Sync
Done -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
            Cont Output
newout -> do
                (Handle -> IO ()) -> Maybe Handle -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ Handle -> IO ()
TM.tickle (LoopCheck -> Maybe Handle
lcTimeHandle LoopCheck
lc)
                Bool
cont <- LoopCheck -> IO Bool
checkLoop LoopCheck
lc
                Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
cont (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
                    TQueue Output -> Output -> IO ()
enqueueOutput TQueue Output
outputQ Output
newout
                    IO ()
loop

newLoopCheck
    :: Stream -> Maybe (TBQueue StreamingChunk) -> Maybe TM.Handle -> IO LoopCheck
newLoopCheck :: Stream
-> Maybe (TBQueue StreamingChunk) -> Maybe Handle -> IO LoopCheck
newLoopCheck Stream
strm Maybe (TBQueue StreamingChunk)
mtbq Maybe Handle
mth = do
    TVar Bool
tovar <- Bool -> IO (TVar Bool)
forall a. a -> IO (TVar a)
newTVarIO Bool
False
    LoopCheck -> IO LoopCheck
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (LoopCheck -> IO LoopCheck) -> LoopCheck -> IO LoopCheck
forall a b. (a -> b) -> a -> b
$
        LoopCheck
            { lcState :: TVar StreamState
lcState = Stream -> TVar StreamState
streamState Stream
strm
            , lcTBQ :: Maybe (TBQueue StreamingChunk)
lcTBQ = Maybe (TBQueue StreamingChunk)
mtbq
            , lcTimeout :: TVar Bool
lcTimeout = TVar Bool
tovar
            , lcWindow :: TVar TxFlow
lcWindow = Stream -> TVar TxFlow
streamTxFlow Stream
strm
            , lcTimeHandle :: Maybe Handle
lcTimeHandle = Maybe Handle
mth
            }

data LoopCheck = LoopCheck
    { LoopCheck -> TVar StreamState
lcState :: TVar StreamState
    , LoopCheck -> Maybe (TBQueue StreamingChunk)
lcTBQ :: Maybe (TBQueue StreamingChunk)
    , LoopCheck -> TVar Bool
lcTimeout :: TVar Bool
    , LoopCheck -> TVar TxFlow
lcWindow :: TVar TxFlow
    , LoopCheck -> Maybe Handle
lcTimeHandle :: Maybe TM.Handle
    }

checkLoop :: LoopCheck -> IO Bool
checkLoop :: LoopCheck -> IO Bool
checkLoop LoopCheck{Maybe (TBQueue StreamingChunk)
Maybe Handle
TVar Bool
TVar TxFlow
TVar StreamState
lcTimeHandle :: LoopCheck -> Maybe Handle
lcState :: LoopCheck -> TVar StreamState
lcTBQ :: LoopCheck -> Maybe (TBQueue StreamingChunk)
lcTimeout :: LoopCheck -> TVar Bool
lcWindow :: LoopCheck -> TVar TxFlow
lcState :: TVar StreamState
lcTBQ :: Maybe (TBQueue StreamingChunk)
lcTimeout :: TVar Bool
lcWindow :: TVar TxFlow
lcTimeHandle :: Maybe Handle
..} = STM Bool -> IO Bool
forall a. STM a -> IO a
atomically (STM Bool -> IO Bool) -> STM Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
    Bool
tout <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar TVar Bool
lcTimeout
    StreamState
state <- TVar StreamState -> STM StreamState
forall a. TVar a -> STM a
readTVar TVar StreamState
lcState
    if
        | Bool
tout -> Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
        | Closed{} <- StreamState
state -> Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
False
        | Bool
otherwise -> do
            Maybe (TBQueue StreamingChunk) -> STM ()
forall a. Maybe (TBQueue a) -> STM ()
waitStreaming' Maybe (TBQueue StreamingChunk)
lcTBQ
            TVar TxFlow -> STM ()
waitStreamWindowSizeSTM TVar TxFlow
lcWindow
            Bool -> STM Bool
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Bool
True

waitStreaming' :: Maybe (TBQueue a) -> STM ()
waitStreaming' :: forall a. Maybe (TBQueue a) -> STM ()
waitStreaming' Maybe (TBQueue a)
Nothing = () -> STM ()
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
waitStreaming' (Just TBQueue a
tbq) = do
    Bool
isEmpty <- TBQueue a -> STM Bool
forall a. TBQueue a -> STM Bool
isEmptyTBQueue TBQueue a
tbq
    Bool -> STM ()
check (Bool -> Bool
not Bool
isEmpty)

waitStreamWindowSizeSTM :: TVar TxFlow -> STM ()
waitStreamWindowSizeSTM :: TVar TxFlow -> STM ()
waitStreamWindowSizeSTM TVar TxFlow
txf = do
    WindowSize
w <- TxFlow -> WindowSize
txWindowSize (TxFlow -> WindowSize) -> STM TxFlow -> STM WindowSize
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar TxFlow -> STM TxFlow
forall a. TVar a -> STM a
readTVar TVar TxFlow
txf
    Bool -> STM ()
check (WindowSize
w WindowSize -> WindowSize -> Bool
forall a. Ord a => a -> a -> Bool
> WindowSize
0)