{-# LANGUAGE NamedFieldPuns #-}

module Network.HTTP2.H2.Stream where

import Control.Concurrent
import Control.Concurrent.STM
import qualified Control.Exception as E
import Control.Monad
import Data.IORef
import Data.Maybe (fromMaybe)
import Network.Control
import Network.HTTP.Semantics.IO

import Network.HTTP2.Frame
import Network.HTTP2.H2.StreamTable
import Network.HTTP2.H2.Types

----------------------------------------------------------------

isIdle :: StreamState -> Bool
isIdle :: StreamState -> Bool
isIdle StreamState
Idle = Bool
True
isIdle StreamState
_ = Bool
False

isOpen :: StreamState -> Bool
isOpen :: StreamState -> Bool
isOpen Open{} = Bool
True
isOpen StreamState
_ = Bool
False

isHalfClosedRemote :: StreamState -> Bool
isHalfClosedRemote :: StreamState -> Bool
isHalfClosedRemote StreamState
HalfClosedRemote = Bool
True
isHalfClosedRemote (Closed ClosedCode
_) = Bool
True
isHalfClosedRemote StreamState
_ = Bool
False

isHalfClosedLocal :: StreamState -> Bool
isHalfClosedLocal :: StreamState -> Bool
isHalfClosedLocal (Open (Just ClosedCode
_) OpenState
_) = Bool
True
isHalfClosedLocal (Closed ClosedCode
_) = Bool
True
isHalfClosedLocal StreamState
_ = Bool
False

isClosed :: StreamState -> Bool
isClosed :: StreamState -> Bool
isClosed Closed{} = Bool
True
isClosed StreamState
_ = Bool
False

isReserved :: StreamState -> Bool
isReserved :: StreamState -> Bool
isReserved StreamState
Reserved = Bool
True
isReserved StreamState
_ = Bool
False

----------------------------------------------------------------

newOddStream :: StreamId -> WindowSize -> WindowSize -> IO Stream
newOddStream :: StreamId -> StreamId -> StreamId -> IO Stream
newOddStream StreamId
sid StreamId
txwin StreamId
rxwin =
    StreamId
-> TVar StreamState
-> MVar (Either SomeException InpObj)
-> TVar TxFlow
-> IORef RxFlow
-> IORef (Maybe RxQ)
-> IORef (Maybe ByteString)
-> TVar Bool
-> Stream
Stream StreamId
sid
        (TVar StreamState
 -> MVar (Either SomeException InpObj)
 -> TVar TxFlow
 -> IORef RxFlow
 -> IORef (Maybe RxQ)
 -> IORef (Maybe ByteString)
 -> TVar Bool
 -> Stream)
-> IO (TVar StreamState)
-> IO
     (MVar (Either SomeException InpObj)
      -> TVar TxFlow
      -> IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StreamState -> IO (TVar StreamState)
forall a. a -> IO (TVar a)
newTVarIO StreamState
Idle
        IO
  (MVar (Either SomeException InpObj)
   -> TVar TxFlow
   -> IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (MVar (Either SomeException InpObj))
-> IO
     (TVar TxFlow
      -> IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> IO (MVar (Either SomeException InpObj))
forall a. IO (MVar a)
newEmptyMVar
        IO
  (TVar TxFlow
   -> IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (TVar TxFlow)
-> IO
     (IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> TxFlow -> IO (TVar TxFlow)
forall a. a -> IO (TVar a)
newTVarIO (StreamId -> TxFlow
newTxFlow StreamId
txwin)
        IO
  (IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (IORef RxFlow)
-> IO
     (IORef (Maybe RxQ)
      -> IORef (Maybe ByteString) -> TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> RxFlow -> IO (IORef RxFlow)
forall a. a -> IO (IORef a)
newIORef (StreamId -> RxFlow
newRxFlow StreamId
rxwin)
        IO
  (IORef (Maybe RxQ)
   -> IORef (Maybe ByteString) -> TVar Bool -> Stream)
-> IO (IORef (Maybe RxQ))
-> IO (IORef (Maybe ByteString) -> TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe RxQ -> IO (IORef (Maybe RxQ))
forall a. a -> IO (IORef a)
newIORef Maybe RxQ
forall a. Maybe a
Nothing
        IO (IORef (Maybe ByteString) -> TVar Bool -> Stream)
-> IO (IORef (Maybe ByteString)) -> IO (TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe ByteString -> IO (IORef (Maybe ByteString))
forall a. a -> IO (IORef a)
newIORef Maybe ByteString
forall a. Maybe a
Nothing
        IO (TVar Bool -> Stream) -> IO (TVar Bool) -> IO Stream
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Bool -> IO (TVar Bool)
forall a. a -> IO (TVar a)
newTVarIO Bool
False

newEvenStream :: StreamId -> WindowSize -> WindowSize -> IO Stream
newEvenStream :: StreamId -> StreamId -> StreamId -> IO Stream
newEvenStream StreamId
sid StreamId
txwin StreamId
rxwin =
    StreamId
-> TVar StreamState
-> MVar (Either SomeException InpObj)
-> TVar TxFlow
-> IORef RxFlow
-> IORef (Maybe RxQ)
-> IORef (Maybe ByteString)
-> TVar Bool
-> Stream
Stream StreamId
sid
        (TVar StreamState
 -> MVar (Either SomeException InpObj)
 -> TVar TxFlow
 -> IORef RxFlow
 -> IORef (Maybe RxQ)
 -> IORef (Maybe ByteString)
 -> TVar Bool
 -> Stream)
-> IO (TVar StreamState)
-> IO
     (MVar (Either SomeException InpObj)
      -> TVar TxFlow
      -> IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StreamState -> IO (TVar StreamState)
forall a. a -> IO (TVar a)
newTVarIO StreamState
Reserved
        IO
  (MVar (Either SomeException InpObj)
   -> TVar TxFlow
   -> IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (MVar (Either SomeException InpObj))
-> IO
     (TVar TxFlow
      -> IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> IO (MVar (Either SomeException InpObj))
forall a. IO (MVar a)
newEmptyMVar
        IO
  (TVar TxFlow
   -> IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (TVar TxFlow)
-> IO
     (IORef RxFlow
      -> IORef (Maybe RxQ)
      -> IORef (Maybe ByteString)
      -> TVar Bool
      -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> TxFlow -> IO (TVar TxFlow)
forall a. a -> IO (TVar a)
newTVarIO (StreamId -> TxFlow
newTxFlow StreamId
txwin)
        IO
  (IORef RxFlow
   -> IORef (Maybe RxQ)
   -> IORef (Maybe ByteString)
   -> TVar Bool
   -> Stream)
-> IO (IORef RxFlow)
-> IO
     (IORef (Maybe RxQ)
      -> IORef (Maybe ByteString) -> TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> RxFlow -> IO (IORef RxFlow)
forall a. a -> IO (IORef a)
newIORef (StreamId -> RxFlow
newRxFlow StreamId
rxwin)
        IO
  (IORef (Maybe RxQ)
   -> IORef (Maybe ByteString) -> TVar Bool -> Stream)
-> IO (IORef (Maybe RxQ))
-> IO (IORef (Maybe ByteString) -> TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe RxQ -> IO (IORef (Maybe RxQ))
forall a. a -> IO (IORef a)
newIORef Maybe RxQ
forall a. Maybe a
Nothing
        IO (IORef (Maybe ByteString) -> TVar Bool -> Stream)
-> IO (IORef (Maybe ByteString)) -> IO (TVar Bool -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe ByteString -> IO (IORef (Maybe ByteString))
forall a. a -> IO (IORef a)
newIORef Maybe ByteString
forall a. Maybe a
Nothing
        IO (TVar Bool -> Stream) -> IO (TVar Bool) -> IO Stream
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Bool -> IO (TVar Bool)
forall a. a -> IO (TVar a)
newTVarIO Bool
False

----------------------------------------------------------------

{-# INLINE readStreamState #-}
readStreamState :: Stream -> IO StreamState
readStreamState :: Stream -> IO StreamState
readStreamState Stream{TVar StreamState
streamState :: TVar StreamState
streamState :: Stream -> TVar StreamState
streamState} = TVar StreamState -> IO StreamState
forall a. TVar a -> IO a
readTVarIO TVar StreamState
streamState

----------------------------------------------------------------

closeAllStreams
    :: TVar OddStreamTable -> TVar EvenStreamTable -> Maybe E.SomeException -> IO ()
closeAllStreams :: TVar OddStreamTable
-> TVar EvenStreamTable -> Maybe SomeException -> IO ()
closeAllStreams TVar OddStreamTable
ovar TVar EvenStreamTable
evar Maybe SomeException
mErr = do
    IntMap Stream
ostrms <- TVar OddStreamTable -> IO (IntMap Stream)
clearOddStreamTable TVar OddStreamTable
ovar
    (Stream -> IO ()) -> IntMap Stream -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ Stream -> IO ()
finalize IntMap Stream
ostrms
    IntMap Stream
estrms <- TVar EvenStreamTable -> IO (IntMap Stream)
clearEvenStreamTable TVar EvenStreamTable
evar
    (Stream -> IO ()) -> IntMap Stream -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ Stream -> IO ()
finalize IntMap Stream
estrms
  where
    -- We treat /every/ exception, including 'ConectionIsClosed', as abnormal
    -- termination: we should only report a clean termination when we receive an
    -- explicit @END_STREAM@ frame.
    finalize :: Stream -> IO ()
finalize Stream
strm = do
        StreamState
st <- Stream -> IO StreamState
readStreamState Stream
strm
        IO Bool -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Bool -> IO ()) -> IO Bool -> IO ()
forall a b. (a -> b) -> a -> b
$ MVar (Either SomeException InpObj)
-> Either SomeException InpObj -> IO Bool
forall a. MVar a -> a -> IO Bool
tryPutMVar (Stream -> MVar (Either SomeException InpObj)
streamInput Stream
strm) Either SomeException InpObj
forall a. Either SomeException a
err
        case StreamState
st of
            Open Maybe ClosedCode
_ (Body RxQ
q Maybe StreamId
_ IORef StreamId
_ IORef (Maybe TokenHeaderTable)
_) ->
                STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ RxQ -> Either SomeException (ByteString, Bool) -> STM ()
forall a. TQueue a -> a -> STM ()
writeTQueue RxQ
q (Either SomeException (ByteString, Bool) -> STM ())
-> Either SomeException (ByteString, Bool) -> STM ()
forall a b. (a -> b) -> a -> b
$ Either SomeException (ByteString, Bool)
-> (SomeException -> Either SomeException (ByteString, Bool))
-> Maybe SomeException
-> Either SomeException (ByteString, Bool)
forall b a. b -> (a -> b) -> Maybe a -> b
maybe ((ByteString, Bool) -> Either SomeException (ByteString, Bool)
forall a b. b -> Either a b
Right (ByteString
forall a. Monoid a => a
mempty, Bool
True)) SomeException -> Either SomeException (ByteString, Bool)
forall a b. a -> Either a b
Left Maybe SomeException
mErr
            StreamState
_otherwise ->
                () -> IO ()
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

    err :: Either E.SomeException a
    err :: forall a. Either SomeException a
err = SomeException -> Either SomeException a
forall a b. a -> Either a b
Left (SomeException -> Either SomeException a)
-> SomeException -> Either SomeException a
forall a b. (a -> b) -> a -> b
$ SomeException -> Maybe SomeException -> SomeException
forall a. a -> Maybe a -> a
fromMaybe (HTTP2Error -> SomeException
forall e. Exception e => e -> SomeException
E.toException HTTP2Error
ConnectionIsClosed) Maybe SomeException
mErr

----------------------------------------------------------------

nextForStreaming
    :: TBQueue StreamingChunk
    -> DynaNext
nextForStreaming :: TBQueue StreamingChunk -> DynaNext
nextForStreaming TBQueue StreamingChunk
tbq =
    let takeQ :: IO (Maybe StreamingChunk)
takeQ = STM (Maybe StreamingChunk) -> IO (Maybe StreamingChunk)
forall a. STM a -> IO 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 DynaNext
next