{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Network.HTTP2.H2.Sender (
frameSender,
) where
import Control.Concurrent.MVar (putMVar)
import Data.IORef (modifyIORef', readIORef, writeIORef)
import Data.IntMap.Strict (IntMap)
import Foreign.Ptr (minusPtr, plusPtr)
import Network.ByteOrder
import Network.HTTP.Semantics.Client
import Network.HTTP.Semantics.IO
import qualified UnliftIO.Exception as E
import UnliftIO.STM
import Imports
import Network.HPACK (setLimitForEncoding, toTokenHeaderTable)
import Network.HTTP2.Frame
import Network.HTTP2.H2.Context
import Network.HTTP2.H2.EncodeFrame
import Network.HTTP2.H2.HPACK
import Network.HTTP2.H2.Manager hiding (start)
import Network.HTTP2.H2.Queue
import Network.HTTP2.H2.Settings
import Network.HTTP2.H2.Stream
import Network.HTTP2.H2.StreamTable
import Network.HTTP2.H2.Types
import Network.HTTP2.H2.Window
{-# INLINE waitStreaming #-}
waitStreaming :: TBQueue a -> IO ()
waitStreaming :: forall a. TBQueue a -> IO ()
waitStreaming TBQueue a
tbq = STM () -> IO ()
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
Bool
isEmpty <- TBQueue a -> STM Bool
forall a. TBQueue a -> STM Bool
isEmptyTBQueue TBQueue a
tbq
Bool -> STM ()
checkSTM (Bool -> Bool
not Bool
isEmpty)
data Switch
= C Control
| O (Output Stream)
| Flush
wrapException :: E.SomeException -> IO ()
wrapException :: SomeException -> IO ()
wrapException SomeException
se
| Just (HTTP2Error
e :: HTTP2Error) <- SomeException -> Maybe HTTP2Error
forall e. Exception e => SomeException -> Maybe e
E.fromException SomeException
se = HTTP2Error -> IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
E.throwIO HTTP2Error
e
| Bool
otherwise = HTTP2Error -> IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
E.throwIO (HTTP2Error -> IO ()) -> HTTP2Error -> IO ()
forall a b. (a -> b) -> a -> b
$ SomeException -> HTTP2Error
BadThingHappen SomeException
se
updatePeerSettings :: Context -> SettingsList -> IO ()
updatePeerSettings :: Context -> SettingsList -> IO ()
updatePeerSettings Context{IORef Settings
peerSettings :: IORef Settings
peerSettings :: Context -> IORef Settings
peerSettings, TVar OddStreamTable
oddStreamTable :: TVar OddStreamTable
oddStreamTable :: Context -> TVar OddStreamTable
oddStreamTable, TVar EvenStreamTable
evenStreamTable :: TVar EvenStreamTable
evenStreamTable :: Context -> TVar EvenStreamTable
evenStreamTable} SettingsList
peerAlist = do
WindowSize
oldws <- Settings -> WindowSize
initialWindowSize (Settings -> WindowSize) -> IO Settings -> IO WindowSize
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Settings -> IO Settings
forall a. IORef a -> IO a
readIORef IORef Settings
peerSettings
IORef Settings -> (Settings -> Settings) -> IO ()
forall a. IORef a -> (a -> a) -> IO ()
modifyIORef' IORef Settings
peerSettings ((Settings -> Settings) -> IO ())
-> (Settings -> Settings) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Settings
old -> Settings -> SettingsList -> Settings
fromSettingsList Settings
old SettingsList
peerAlist
WindowSize
newws <- Settings -> WindowSize
initialWindowSize (Settings -> WindowSize) -> IO Settings -> IO WindowSize
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IORef Settings -> IO Settings
forall a. IORef a -> IO a
readIORef IORef Settings
peerSettings
let dif :: WindowSize
dif = WindowSize
newws WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
oldws
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (WindowSize
dif WindowSize -> WindowSize -> Bool
forall a. Eq a => a -> a -> Bool
/= WindowSize
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
TVar OddStreamTable -> IO (IntMap Stream)
getOddStreams TVar OddStreamTable
oddStreamTable IO (IntMap Stream) -> (IntMap Stream -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= WindowSize -> IntMap Stream -> IO ()
updateAllStreamTxFlow WindowSize
dif
TVar EvenStreamTable -> IO (IntMap Stream)
getEvenStreams TVar EvenStreamTable
evenStreamTable IO (IntMap Stream) -> (IntMap Stream -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= WindowSize -> IntMap Stream -> IO ()
updateAllStreamTxFlow WindowSize
dif
where
updateAllStreamTxFlow :: WindowSize -> IntMap Stream -> IO ()
updateAllStreamTxFlow :: WindowSize -> IntMap Stream -> IO ()
updateAllStreamTxFlow WindowSize
siz IntMap Stream
strms =
IntMap Stream -> (Stream -> IO ()) -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ IntMap Stream
strms ((Stream -> IO ()) -> IO ()) -> (Stream -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Stream
strm -> Stream -> WindowSize -> IO ()
increaseStreamWindowSize Stream
strm WindowSize
siz
frameSender :: Context -> Config -> Manager -> IO ()
frameSender :: Context -> Config -> Manager -> IO ()
frameSender
ctx :: Context
ctx@Context{TQueue (Output Stream)
outputQ :: TQueue (Output Stream)
outputQ :: Context -> TQueue (Output Stream)
outputQ, TQueue Control
controlQ :: TQueue Control
controlQ :: Context -> TQueue Control
controlQ, DynamicTable
encodeDynamicTable :: DynamicTable
encodeDynamicTable :: Context -> DynamicTable
encodeDynamicTable, IORef WindowSize
outputBufferLimit :: IORef WindowSize
outputBufferLimit :: Context -> IORef WindowSize
outputBufferLimit}
Config{WindowSize
Buffer
Manager
SockAddr
WindowSize -> IO FieldValue
PositionReadMaker
FieldValue -> IO ()
confWriteBuffer :: Buffer
confBufferSize :: WindowSize
confSendAll :: FieldValue -> IO ()
confReadN :: WindowSize -> IO FieldValue
confPositionReadMaker :: PositionReadMaker
confTimeoutManager :: Manager
confMySockAddr :: SockAddr
confPeerSockAddr :: SockAddr
confWriteBuffer :: Config -> Buffer
confBufferSize :: Config -> WindowSize
confSendAll :: Config -> FieldValue -> IO ()
confReadN :: Config -> WindowSize -> IO FieldValue
confPositionReadMaker :: Config -> PositionReadMaker
confTimeoutManager :: Config -> Manager
confMySockAddr :: Config -> SockAddr
confPeerSockAddr :: Config -> SockAddr
..}
Manager
mgr = WindowSize -> IO ()
loop WindowSize
0 IO () -> (SomeException -> IO ()) -> IO ()
forall (m :: * -> *) e a.
(MonadUnliftIO m, Exception e) =>
m a -> (e -> m a) -> m a
`E.catch` SomeException -> IO ()
wrapException
where
loop :: Offset -> IO ()
loop :: WindowSize -> IO ()
loop WindowSize
off = do
Switch
x <- STM Switch -> IO Switch
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM Switch -> IO Switch) -> STM Switch -> IO Switch
forall a b. (a -> b) -> a -> b
$ WindowSize -> STM Switch
dequeue WindowSize
off
case Switch
x of
C Control
ctl -> WindowSize -> IO ()
flushN WindowSize
off IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Control -> IO ()
control Control
ctl IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WindowSize -> IO ()
loop WindowSize
0
O Output Stream
out -> Output Stream -> WindowSize -> IO WindowSize
outputOrEnqueueAgain Output Stream
out WindowSize
off IO WindowSize -> (WindowSize -> IO WindowSize) -> IO WindowSize
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= WindowSize -> IO WindowSize
flushIfNecessary IO WindowSize -> (WindowSize -> IO ()) -> IO ()
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= WindowSize -> IO ()
loop
Switch
Flush -> WindowSize -> IO ()
flushN WindowSize
off IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> WindowSize -> IO ()
loop WindowSize
0
flushN :: Offset -> IO ()
flushN :: WindowSize -> IO ()
flushN WindowSize
0 = () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
flushN WindowSize
n = Buffer -> WindowSize -> (FieldValue -> IO ()) -> IO ()
forall a. Buffer -> WindowSize -> (FieldValue -> IO a) -> IO a
bufferIO Buffer
confWriteBuffer WindowSize
n FieldValue -> IO ()
confSendAll
flushIfNecessary :: Offset -> IO Offset
flushIfNecessary :: WindowSize -> IO WindowSize
flushIfNecessary WindowSize
off = do
WindowSize
buflim <- IORef WindowSize -> IO WindowSize
forall a. IORef a -> IO a
readIORef IORef WindowSize
outputBufferLimit
if WindowSize
off WindowSize -> WindowSize -> Bool
forall a. Ord a => a -> a -> Bool
<= WindowSize
buflim WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
512
then WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
else do
WindowSize -> IO ()
flushN WindowSize
off
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
0
dequeue :: Offset -> STM Switch
dequeue :: WindowSize -> STM Switch
dequeue WindowSize
off = do
Bool
isEmptyC <- TQueue Control -> STM Bool
forall a. TQueue a -> STM Bool
isEmptyTQueue TQueue Control
controlQ
if Bool
isEmptyC
then do
Context -> STM ()
waitConnectionWindowSize Context
ctx
Bool
isEmptyO <- TQueue (Output Stream) -> STM Bool
forall a. TQueue a -> STM Bool
isEmptyTQueue TQueue (Output Stream)
outputQ
if Bool
isEmptyO
then if WindowSize
off WindowSize -> WindowSize -> Bool
forall a. Eq a => a -> a -> Bool
/= WindowSize
0 then Switch -> STM Switch
forall a. a -> STM a
forall (m :: * -> *) a. Monad m => a -> m a
return Switch
Flush else STM Switch
forall a. STM a
retrySTM
else Output Stream -> Switch
O (Output Stream -> Switch) -> STM (Output Stream) -> STM Switch
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TQueue (Output Stream) -> STM (Output Stream)
forall a. TQueue a -> STM a
readTQueue TQueue (Output Stream)
outputQ
else Control -> Switch
C (Control -> Switch) -> STM Control -> STM Switch
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TQueue Control -> STM Control
forall a. TQueue a -> STM a
readTQueue TQueue Control
controlQ
copyAll :: [FieldValue] -> Buffer -> IO Buffer
copyAll [] Buffer
buf = Buffer -> IO Buffer
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Buffer
buf
copyAll (FieldValue
x : [FieldValue]
xs) Buffer
buf = Buffer -> FieldValue -> IO Buffer
copy Buffer
buf FieldValue
x IO Buffer -> (Buffer -> IO Buffer) -> IO Buffer
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= [FieldValue] -> Buffer -> IO Buffer
copyAll [FieldValue]
xs
control :: Control -> IO ()
control :: Control -> IO ()
control (CFinish HTTP2Error
e) = HTTP2Error -> IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
E.throwIO HTTP2Error
e
control (CGoaway FieldValue
bs MVar ()
mvar) = do
Buffer
buf <- [FieldValue] -> Buffer -> IO Buffer
copyAll [FieldValue
bs] Buffer
confWriteBuffer
let off :: WindowSize
off = Buffer
buf Buffer -> Buffer -> WindowSize
forall a b. Ptr a -> Ptr b -> WindowSize
`minusPtr` Buffer
confWriteBuffer
WindowSize -> IO ()
flushN WindowSize
off
MVar () -> () -> IO ()
forall a. MVar a -> a -> IO ()
putMVar MVar ()
mvar ()
HTTP2Error -> IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
E.throwIO HTTP2Error
GoAwayIsSent
control (CFrames Maybe SettingsList
ms [FieldValue]
xs) = do
Buffer
buf <- [FieldValue] -> Buffer -> IO Buffer
copyAll [FieldValue]
xs Buffer
confWriteBuffer
let off :: WindowSize
off = Buffer
buf Buffer -> Buffer -> WindowSize
forall a b. Ptr a -> Ptr b -> WindowSize
`minusPtr` Buffer
confWriteBuffer
WindowSize -> IO ()
flushN WindowSize
off
case Maybe SettingsList
ms of
Maybe SettingsList
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just SettingsList
peerAlist -> do
Context -> SettingsList -> IO ()
updatePeerSettings Context
ctx SettingsList
peerAlist
case SettingsKey -> SettingsList -> Maybe WindowSize
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup SettingsKey
SettingsMaxFrameSize SettingsList
peerAlist of
Maybe WindowSize
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just WindowSize
payloadLen -> do
let dlim :: WindowSize
dlim = WindowSize
payloadLen WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength
buflim :: WindowSize
buflim
| WindowSize
confBufferSize WindowSize -> WindowSize -> Bool
forall a. Ord a => a -> a -> Bool
>= WindowSize
dlim = WindowSize
dlim
| Bool
otherwise = WindowSize
confBufferSize
IORef WindowSize -> WindowSize -> IO ()
forall a. IORef a -> a -> IO ()
writeIORef IORef WindowSize
outputBufferLimit WindowSize
buflim
case SettingsKey -> SettingsList -> Maybe WindowSize
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup SettingsKey
SettingsTokenHeaderTableSize SettingsList
peerAlist of
Maybe WindowSize
Nothing -> () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()
Just WindowSize
siz -> WindowSize -> DynamicTable -> IO ()
setLimitForEncoding WindowSize
siz DynamicTable
encodeDynamicTable
output :: Output Stream -> Offset -> WindowSize -> IO Offset
output :: Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output out :: Output Stream
out@(Output Stream
strm OutObj{} (ONext DynaNext
curr TrailersMaker
tlrmkr) Maybe (TBQueue StreamingChunk)
_ IO ()
sentinel) WindowSize
off0 WindowSize
lim = do
WindowSize
buflim <- IORef WindowSize -> IO WindowSize
forall a. IORef a -> IO a
readIORef IORef WindowSize
outputBufferLimit
let payloadOff :: WindowSize
payloadOff = WindowSize
off0 WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength
datBuf :: Ptr b
datBuf = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
payloadOff
datBufSiz :: WindowSize
datBufSiz = WindowSize
buflim WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
payloadOff
Next WindowSize
datPayloadLen Bool
reqflush Maybe DynaNext
mnext <- DynaNext
curr Buffer
forall {b}. Ptr b
datBuf WindowSize
datBufSiz WindowSize
lim
NextTrailersMaker TrailersMaker
tlrmkr' <- TrailersMaker -> Buffer -> WindowSize -> IO NextTrailersMaker
runTrailersMaker TrailersMaker
tlrmkr Buffer
forall {b}. Ptr b
datBuf WindowSize
datPayloadLen
Stream
-> WindowSize
-> WindowSize
-> Maybe DynaNext
-> TrailersMaker
-> IO ()
-> Output Stream
-> Bool
-> IO WindowSize
fillDataHeaderEnqueueNext
Stream
strm
WindowSize
off0
WindowSize
datPayloadLen
Maybe DynaNext
mnext
TrailersMaker
tlrmkr'
IO ()
sentinel
Output Stream
out
Bool
reqflush
output out :: Output Stream
out@(Output Stream
strm (OutObj [Header]
hdr OutBody
body TrailersMaker
tlrmkr) OutputType
OObj Maybe (TBQueue StreamingChunk)
mtbq IO ()
_) WindowSize
off0 WindowSize
lim = do
let sid :: WindowSize
sid = Stream -> WindowSize
streamNumber Stream
strm
endOfStream :: Bool
endOfStream = case OutBody
body of
OutBody
OutBodyNone -> Bool
True
OutBody
_ -> Bool
False
(TokenHeaderList
ths, ValueTable
_) <- [Header] -> IO (TokenHeaderList, ValueTable)
toTokenHeaderTable ([Header] -> IO (TokenHeaderList, ValueTable))
-> [Header] -> IO (TokenHeaderList, ValueTable)
forall a b. (a -> b) -> a -> b
$ [Header] -> [Header]
fixHeaders [Header]
hdr
WindowSize
off' <- WindowSize
-> TokenHeaderList -> Bool -> WindowSize -> IO WindowSize
headerContinue WindowSize
sid TokenHeaderList
ths Bool
endOfStream WindowSize
off0
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when Bool
endOfStream (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ Context -> Stream -> ClosedCode -> IO ()
halfClosedLocal Context
ctx Stream
strm ClosedCode
Finished
WindowSize
off <- WindowSize -> IO WindowSize
flushIfNecessary WindowSize
off'
case OutBody
body of
OutBody
OutBodyNone -> WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
OutBodyFile (FileSpec FilePath
path FileOffset
fileoff FileOffset
bytecount) -> do
(PositionRead
pread, Sentinel
sentinel') <- PositionReadMaker
confPositionReadMaker FilePath
path
IO ()
refresh <- case Sentinel
sentinel' of
Closer IO ()
closer -> Manager -> IO () -> IO (IO ())
timeoutClose Manager
mgr IO ()
closer
Refresher IO ()
refresher -> IO () -> IO (IO ())
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return IO ()
refresher
let next :: DynaNext
next = PositionRead -> FileOffset -> FileOffset -> IO () -> DynaNext
fillFileBodyGetNext PositionRead
pread FileOffset
fileoff FileOffset
bytecount IO ()
refresh
out' :: Output Stream
out' = Output Stream
out{outputType = ONext next tlrmkr}
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output Output Stream
out' WindowSize
off WindowSize
lim
OutBodyBuilder Builder
builder -> do
let next :: DynaNext
next = Builder -> DynaNext
fillBuilderBodyGetNext Builder
builder
out' :: Output Stream
out' = Output Stream
out{outputType = ONext next tlrmkr}
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output Output Stream
out' WindowSize
off WindowSize
lim
OutBodyStreaming (Builder -> IO ()) -> IO () -> IO ()
_ ->
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output (Maybe (TBQueue StreamingChunk)
-> TrailersMaker -> Output Stream -> Output Stream
setNextForStreaming Maybe (TBQueue StreamingChunk)
mtbq TrailersMaker
tlrmkr Output Stream
out) WindowSize
off WindowSize
lim
OutBodyStreamingUnmask (forall x. IO x -> IO x) -> (Builder -> IO ()) -> IO () -> IO ()
_ ->
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output (Maybe (TBQueue StreamingChunk)
-> TrailersMaker -> Output Stream -> Output Stream
setNextForStreaming Maybe (TBQueue StreamingChunk)
mtbq TrailersMaker
tlrmkr Output Stream
out) WindowSize
off WindowSize
lim
output out :: Output Stream
out@(Output Stream
strm OutObj
_ (OPush TokenHeaderList
ths WindowSize
pid) Maybe (TBQueue StreamingChunk)
_ IO ()
_) WindowSize
off0 WindowSize
lim = do
let sid :: WindowSize
sid = Stream -> WindowSize
streamNumber Stream
strm
WindowSize
len <- WindowSize
-> WindowSize -> TokenHeaderList -> WindowSize -> IO WindowSize
pushPromise WindowSize
pid WindowSize
sid TokenHeaderList
ths WindowSize
off0
WindowSize
off <- WindowSize -> IO WindowSize
flushIfNecessary (WindowSize -> IO WindowSize) -> WindowSize -> IO WindowSize
forall a b. (a -> b) -> a -> b
$ WindowSize
off0 WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
len
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output Output Stream
out{outputType = OObj} WindowSize
off WindowSize
lim
output Output Stream
_ WindowSize
_ WindowSize
_ = IO WindowSize
forall a. HasCallStack => a
undefined
setNextForStreaming
:: Maybe (TBQueue StreamingChunk)
-> TrailersMaker
-> Output Stream
-> Output Stream
setNextForStreaming :: Maybe (TBQueue StreamingChunk)
-> TrailersMaker -> Output Stream -> Output Stream
setNextForStreaming Maybe (TBQueue StreamingChunk)
mtbq TrailersMaker
tlrmkr Output Stream
out =
let tbq :: TBQueue StreamingChunk
tbq = Maybe (TBQueue StreamingChunk) -> TBQueue StreamingChunk
forall a. HasCallStack => Maybe a -> a
fromJust Maybe (TBQueue StreamingChunk)
mtbq
takeQ :: IO (Maybe StreamingChunk)
takeQ = STM (Maybe StreamingChunk) -> IO (Maybe StreamingChunk)
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM (Maybe StreamingChunk) -> IO (Maybe StreamingChunk))
-> STM (Maybe StreamingChunk) -> IO (Maybe StreamingChunk)
forall a b. (a -> b) -> a -> b
$ TBQueue StreamingChunk -> STM (Maybe StreamingChunk)
forall a. TBQueue a -> STM (Maybe a)
tryReadTBQueue TBQueue StreamingChunk
tbq
next :: DynaNext
next = IO (Maybe StreamingChunk) -> DynaNext
fillStreamBodyGetNext IO (Maybe StreamingChunk)
takeQ
in Output Stream
out{outputType = ONext next tlrmkr}
outputOrEnqueueAgain :: Output Stream -> Offset -> IO Offset
outputOrEnqueueAgain :: Output Stream -> WindowSize -> IO WindowSize
outputOrEnqueueAgain out :: Output Stream
out@(Output Stream
strm OutObj
_ OutputType
otyp Maybe (TBQueue StreamingChunk)
_ IO ()
_) WindowSize
off = (SomeException -> IO WindowSize) -> IO WindowSize -> IO WindowSize
forall (m :: * -> *) e a.
(MonadUnliftIO m, Exception e) =>
(e -> m a) -> m a -> m a
E.handle SomeException -> IO WindowSize
resetStream (IO WindowSize -> IO WindowSize) -> IO WindowSize -> IO WindowSize
forall a b. (a -> b) -> a -> b
$ do
StreamState
state <- Stream -> IO StreamState
readStreamState Stream
strm
if StreamState -> Bool
isHalfClosedLocal StreamState
state
then WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
else case OutputType
otyp of
OWait IO ()
wait -> do
IO ()
-> TQueue (Output Stream) -> Output Stream -> Manager -> IO ()
forkAndEnqueueWhenReady IO ()
wait TQueue (Output Stream)
outputQ Output Stream
out{outputType = OObj} Manager
mgr
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
OutputType
_ -> case Maybe (TBQueue StreamingChunk)
mtbq of
Just TBQueue StreamingChunk
tbq -> TBQueue StreamingChunk -> IO WindowSize
forall {a}. TBQueue a -> IO WindowSize
checkStreaming TBQueue StreamingChunk
tbq
Maybe (TBQueue StreamingChunk)
_ -> IO WindowSize
checkStreamWindowSize
where
mtbq :: Maybe (TBQueue StreamingChunk)
mtbq = Output Stream -> Maybe (TBQueue StreamingChunk)
forall a. Output a -> Maybe (TBQueue StreamingChunk)
outputStrmQ Output Stream
out
checkStreaming :: TBQueue a -> IO WindowSize
checkStreaming TBQueue a
tbq = do
Bool
isEmpty <- STM Bool -> IO Bool
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM Bool -> IO Bool) -> STM Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ TBQueue a -> STM Bool
forall a. TBQueue a -> STM Bool
isEmptyTBQueue TBQueue a
tbq
if Bool
isEmpty
then do
IO ()
-> TQueue (Output Stream) -> Output Stream -> Manager -> IO ()
forkAndEnqueueWhenReady (TBQueue a -> IO ()
forall a. TBQueue a -> IO ()
waitStreaming TBQueue a
tbq) TQueue (Output Stream)
outputQ Output Stream
out Manager
mgr
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
else IO WindowSize
checkStreamWindowSize
checkStreamWindowSize :: IO WindowSize
checkStreamWindowSize = do
WindowSize
sws <- Stream -> IO WindowSize
getStreamWindowSize Stream
strm
if WindowSize
sws WindowSize -> WindowSize -> Bool
forall a. Eq a => a -> a -> Bool
== WindowSize
0
then do
IO ()
-> TQueue (Output Stream) -> Output Stream -> Manager -> IO ()
forkAndEnqueueWhenReady (Stream -> IO ()
waitStreamWindowSize Stream
strm) TQueue (Output Stream)
outputQ Output Stream
out Manager
mgr
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
else do
WindowSize
cws <- Context -> IO WindowSize
getConnectionWindowSize Context
ctx
let lim :: WindowSize
lim = WindowSize -> WindowSize -> WindowSize
forall a. Ord a => a -> a -> a
min WindowSize
cws WindowSize
sws
Output Stream -> WindowSize -> WindowSize -> IO WindowSize
output Output Stream
out WindowSize
off WindowSize
lim
resetStream :: SomeException -> IO WindowSize
resetStream SomeException
e = do
Context -> Stream -> ClosedCode -> IO ()
closed Context
ctx Stream
strm (SomeException -> ClosedCode
ResetByMe SomeException
e)
let rst :: FieldValue
rst = ErrorCode -> WindowSize -> FieldValue
resetFrame ErrorCode
InternalError (WindowSize -> FieldValue) -> WindowSize -> FieldValue
forall a b. (a -> b) -> a -> b
$ Stream -> WindowSize
streamNumber Stream
strm
TQueue Control -> Control -> IO ()
enqueueControl TQueue Control
controlQ (Control -> IO ()) -> Control -> IO ()
forall a b. (a -> b) -> a -> b
$ Maybe SettingsList -> [FieldValue] -> Control
CFrames Maybe SettingsList
forall a. Maybe a
Nothing [FieldValue
rst]
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
headerContinue :: StreamId -> TokenHeaderList -> Bool -> Offset -> IO Offset
headerContinue :: WindowSize
-> TokenHeaderList -> Bool -> WindowSize -> IO WindowSize
headerContinue WindowSize
sid TokenHeaderList
ths0 Bool
endOfStream WindowSize
off0 = do
WindowSize
buflim <- IORef WindowSize -> IO WindowSize
forall a. IORef a -> IO a
readIORef IORef WindowSize
outputBufferLimit
let offkv :: WindowSize
offkv = WindowSize
off0 WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength
bufkv :: Ptr b
bufkv = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
offkv
limkv :: WindowSize
limkv = WindowSize
buflim WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
offkv
(TokenHeaderList
ths, WindowSize
kvlen) <- Context
-> Buffer
-> WindowSize
-> TokenHeaderList
-> IO (TokenHeaderList, WindowSize)
hpackEncodeHeader Context
ctx Buffer
forall {b}. Ptr b
bufkv WindowSize
limkv TokenHeaderList
ths0
if WindowSize
kvlen WindowSize -> WindowSize -> Bool
forall a. Eq a => a -> a -> Bool
== WindowSize
0
then WindowSize -> TokenHeaderList -> FrameType -> IO WindowSize
continue WindowSize
off0 TokenHeaderList
ths FrameType
FrameHeaders
else do
let flag :: FrameFlags
flag = TokenHeaderList -> FrameFlags
forall {a}. [a] -> FrameFlags
getFlag TokenHeaderList
ths
buf :: Ptr b
buf = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
off0
off :: WindowSize
off = WindowSize
offkv WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
kvlen
FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
FrameHeaders WindowSize
kvlen WindowSize
sid FrameFlags
flag Buffer
forall {b}. Ptr b
buf
WindowSize -> TokenHeaderList -> FrameType -> IO WindowSize
continue WindowSize
off TokenHeaderList
ths FrameType
FrameContinuation
where
eos :: FrameFlags -> FrameFlags
eos = if Bool
endOfStream then FrameFlags -> FrameFlags
setEndStream else FrameFlags -> FrameFlags
forall a. a -> a
id
getFlag :: [a] -> FrameFlags
getFlag [] = FrameFlags -> FrameFlags
eos (FrameFlags -> FrameFlags) -> FrameFlags -> FrameFlags
forall a b. (a -> b) -> a -> b
$ FrameFlags -> FrameFlags
setEndHeader FrameFlags
defaultFlags
getFlag [a]
_ = FrameFlags -> FrameFlags
eos (FrameFlags -> FrameFlags) -> FrameFlags -> FrameFlags
forall a b. (a -> b) -> a -> b
$ FrameFlags
defaultFlags
continue :: Offset -> TokenHeaderList -> FrameType -> IO Offset
continue :: WindowSize -> TokenHeaderList -> FrameType -> IO WindowSize
continue WindowSize
off [] FrameType
_ = WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
continue WindowSize
off TokenHeaderList
ths FrameType
ft = do
WindowSize -> IO ()
flushN WindowSize
off
WindowSize
buflim <- IORef WindowSize -> IO WindowSize
forall a. IORef a -> IO a
readIORef IORef WindowSize
outputBufferLimit
let bufHeaderPayload :: Ptr b
bufHeaderPayload = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
frameHeaderLength
headerPayloadLim :: WindowSize
headerPayloadLim = WindowSize
buflim WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
frameHeaderLength
(TokenHeaderList
ths', WindowSize
kvlen') <-
Context
-> Buffer
-> WindowSize
-> TokenHeaderList
-> IO (TokenHeaderList, WindowSize)
hpackEncodeHeaderLoop Context
ctx Buffer
forall {b}. Ptr b
bufHeaderPayload WindowSize
headerPayloadLim TokenHeaderList
ths
Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (TokenHeaderList
ths TokenHeaderList -> TokenHeaderList -> Bool
forall a. Eq a => a -> a -> Bool
== TokenHeaderList
ths') (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
HTTP2Error -> IO ()
forall (m :: * -> *) e a. (MonadIO m, Exception e) => e -> m a
E.throwIO (HTTP2Error -> IO ()) -> HTTP2Error -> IO ()
forall a b. (a -> b) -> a -> b
$
ErrorCode -> WindowSize -> ReasonPhrase -> HTTP2Error
ConnectionErrorIsSent ErrorCode
CompressionError WindowSize
sid ReasonPhrase
"cannot compress the header"
let flag :: FrameFlags
flag = TokenHeaderList -> FrameFlags
forall {a}. [a] -> FrameFlags
getFlag TokenHeaderList
ths'
off' :: WindowSize
off' = WindowSize
frameHeaderLength WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
kvlen'
FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
ft WindowSize
kvlen' WindowSize
sid FrameFlags
flag Buffer
confWriteBuffer
WindowSize -> TokenHeaderList -> FrameType -> IO WindowSize
continue WindowSize
off' TokenHeaderList
ths' FrameType
FrameContinuation
fillDataHeaderEnqueueNext
:: Stream
-> Offset
-> Int
-> Maybe DynaNext
-> (Maybe ByteString -> IO NextTrailersMaker)
-> IO ()
-> Output Stream
-> Bool
-> IO Offset
fillDataHeaderEnqueueNext :: Stream
-> WindowSize
-> WindowSize
-> Maybe DynaNext
-> TrailersMaker
-> IO ()
-> Output Stream
-> Bool
-> IO WindowSize
fillDataHeaderEnqueueNext
strm :: Stream
strm@Stream{WindowSize
streamNumber :: Stream -> WindowSize
streamNumber :: WindowSize
streamNumber}
WindowSize
off
WindowSize
datPayloadLen
Maybe DynaNext
Nothing
TrailersMaker
tlrmkr
IO ()
tell
Output Stream
_
Bool
reqflush = do
let buf :: Ptr b
buf = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
off
off' :: WindowSize
off' = WindowSize
off WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
datPayloadLen
(Maybe [Header]
mtrailers, FrameFlags
flag) <- do
Trailers [Header]
trailers <- TrailersMaker
tlrmkr Maybe FieldValue
forall a. Maybe a
Nothing
if [Header] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [Header]
trailers
then (Maybe [Header], FrameFlags) -> IO (Maybe [Header], FrameFlags)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (Maybe [Header]
forall a. Maybe a
Nothing, FrameFlags -> FrameFlags
setEndStream FrameFlags
defaultFlags)
else (Maybe [Header], FrameFlags) -> IO (Maybe [Header], FrameFlags)
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ([Header] -> Maybe [Header]
forall a. a -> Maybe a
Just [Header]
trailers, FrameFlags
defaultFlags)
FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
FrameData WindowSize
datPayloadLen WindowSize
streamNumber FrameFlags
flag Buffer
forall {b}. Ptr b
buf
WindowSize
off'' <- Maybe [Header] -> WindowSize -> IO WindowSize
handleTrailers Maybe [Header]
mtrailers WindowSize
off'
IO () -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void IO ()
tell
Context -> Stream -> ClosedCode -> IO ()
halfClosedLocal Context
ctx Stream
strm ClosedCode
Finished
Context -> Stream -> WindowSize -> IO ()
decreaseWindowSize Context
ctx Stream
strm WindowSize
datPayloadLen
if Bool
reqflush
then do
WindowSize -> IO ()
flushN WindowSize
off''
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
0
else WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off''
where
handleTrailers :: Maybe [Header] -> WindowSize -> IO WindowSize
handleTrailers Maybe [Header]
Nothing WindowSize
off0 = WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off0
handleTrailers (Just [Header]
trailers) WindowSize
off0 = do
(TokenHeaderList
ths, ValueTable
_) <- [Header] -> IO (TokenHeaderList, ValueTable)
toTokenHeaderTable [Header]
trailers
WindowSize
-> TokenHeaderList -> Bool -> WindowSize -> IO WindowSize
headerContinue WindowSize
streamNumber TokenHeaderList
ths Bool
True WindowSize
off0
fillDataHeaderEnqueueNext
Stream
_
WindowSize
off
WindowSize
0
(Just DynaNext
next)
TrailersMaker
tlrmkr
IO ()
_
Output Stream
out
Bool
reqflush = do
let out' :: Output Stream
out' = Output Stream
out{outputType = ONext next tlrmkr}
TQueue (Output Stream) -> Output Stream -> IO ()
enqueueOutput TQueue (Output Stream)
outputQ Output Stream
out'
if Bool
reqflush
then do
WindowSize -> IO ()
flushN WindowSize
off
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
0
else WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off
fillDataHeaderEnqueueNext
strm :: Stream
strm@Stream{WindowSize
streamNumber :: Stream -> WindowSize
streamNumber :: WindowSize
streamNumber}
WindowSize
off
WindowSize
datPayloadLen
(Just DynaNext
next)
TrailersMaker
tlrmkr
IO ()
_
Output Stream
out
Bool
reqflush = do
let buf :: Ptr b
buf = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
off
off' :: WindowSize
off' = WindowSize
off WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
datPayloadLen
flag :: FrameFlags
flag = FrameFlags
defaultFlags
FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
FrameData WindowSize
datPayloadLen WindowSize
streamNumber FrameFlags
flag Buffer
forall {b}. Ptr b
buf
Context -> Stream -> WindowSize -> IO ()
decreaseWindowSize Context
ctx Stream
strm WindowSize
datPayloadLen
let out' :: Output Stream
out' = Output Stream
out{outputType = ONext next tlrmkr}
TQueue (Output Stream) -> Output Stream -> IO ()
enqueueOutput TQueue (Output Stream)
outputQ Output Stream
out'
if Bool
reqflush
then do
WindowSize -> IO ()
flushN WindowSize
off'
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
0
else WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
off'
pushPromise :: StreamId -> StreamId -> TokenHeaderList -> Offset -> IO Int
pushPromise :: WindowSize
-> WindowSize -> TokenHeaderList -> WindowSize -> IO WindowSize
pushPromise WindowSize
pid WindowSize
sid TokenHeaderList
ths WindowSize
off = do
let offsid :: WindowSize
offsid = WindowSize
off WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
frameHeaderLength
bufsid :: Ptr b
bufsid = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
offsid
Word32 -> Buffer -> WindowSize -> IO ()
poke32 (WindowSize -> Word32
forall a b. (Integral a, Num b) => a -> b
fromIntegral WindowSize
sid) Buffer
forall {b}. Ptr b
bufsid WindowSize
0
let offkv :: WindowSize
offkv = WindowSize
offsid WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
4
bufkv :: Ptr b
bufkv = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
offkv
limkv :: WindowSize
limkv = WindowSize
confBufferSize WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
- WindowSize
offkv
(TokenHeaderList
_, WindowSize
kvlen) <- Context
-> Buffer
-> WindowSize
-> TokenHeaderList
-> IO (TokenHeaderList, WindowSize)
hpackEncodeHeader Context
ctx Buffer
forall {b}. Ptr b
bufkv WindowSize
limkv TokenHeaderList
ths
let flag :: FrameFlags
flag = FrameFlags -> FrameFlags
setEndHeader FrameFlags
defaultFlags
buf :: Ptr b
buf = Buffer
confWriteBuffer Buffer -> WindowSize -> Ptr b
forall a b. Ptr a -> WindowSize -> Ptr b
`plusPtr` WindowSize
off
len :: WindowSize
len = WindowSize
kvlen WindowSize -> WindowSize -> WindowSize
forall a. Num a => a -> a -> a
+ WindowSize
4
FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
FramePushPromise WindowSize
len WindowSize
pid FrameFlags
flag Buffer
forall {b}. Ptr b
buf
WindowSize -> IO WindowSize
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return WindowSize
len
{-# INLINE fillFrameHeader #-}
fillFrameHeader :: FrameType -> Int -> StreamId -> FrameFlags -> Buffer -> IO ()
fillFrameHeader :: FrameType
-> WindowSize -> WindowSize -> FrameFlags -> Buffer -> IO ()
fillFrameHeader FrameType
ftyp WindowSize
len WindowSize
sid FrameFlags
flag Buffer
buf = FrameType -> FrameHeader -> Buffer -> IO ()
encodeFrameHeaderBuf FrameType
ftyp FrameHeader
hinfo Buffer
buf
where
hinfo :: FrameHeader
hinfo =
FrameHeader
{ payloadLength :: WindowSize
payloadLength = WindowSize
len
, flags :: FrameFlags
flags = FrameFlags
flag
, streamId :: WindowSize
streamId = WindowSize
sid
}