{-# LANGUAGE RecordWildCards #-}

module Network.QUIC.Stream.Types (
    Stream (..),
    newStream,
    TxStreamData (..),
    StreamState (..),
    RecvStreamQ (..),
    RxStreamData (..),
    Length,
    syncFinTx,
    waitFinTx,
) where

import qualified Data.ByteString as BS
import Network.Control
import UnliftIO.Concurrent
import UnliftIO.STM

import {-# SOURCE #-} Network.QUIC.Connection.Types
import Network.QUIC.Imports
import Network.QUIC.Stream.Frag
import Network.QUIC.Stream.Skew
import qualified Network.QUIC.Stream.Skew as Skew
import Network.QUIC.Types

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

-- | An abstract data type for streams.
data Stream = Stream
    { Stream -> StreamId
streamId :: StreamId
    -- ^ Getting stream identifier.
    , Stream -> Connection
streamConnection :: Connection
    , -- "counter" is equivalent to "offset".
      -- It is duplicated but used for API level flow control.
      Stream -> TVar TxFlow
streamFlowTx :: TVar TxFlow -- counter, maxDax
    , Stream -> IORef RxFlow
streamFlowRx :: IORef RxFlow -- counter, maxDax
    , Stream -> IORef StreamState
streamStateTx :: IORef StreamState -- offset, fin
    , Stream -> IORef StreamState
streamStateRx :: IORef StreamState -- offset, fin
    , Stream -> RecvStreamQ
streamRecvQ :: RecvStreamQ -- input bytestring
    , Stream -> IORef (Skew RxStreamData)
streamReass :: IORef (Skew RxStreamData) -- input stream fragments to streamQ
    , Stream -> MVar ()
streamSyncFinTx :: MVar ()
    }

instance Show Stream where
    show :: Stream -> String
show Stream
s = StreamId -> String
forall a. Show a => a -> String
show (StreamId -> String) -> StreamId -> String
forall a b. (a -> b) -> a -> b
$ Stream -> StreamId
streamId Stream
s

newStream :: Connection -> Int -> Int -> StreamId -> IO Stream
newStream :: Connection -> StreamId -> StreamId -> StreamId -> IO Stream
newStream Connection
conn StreamId
sid StreamId
txLim StreamId
rxLim =
    StreamId
-> Connection
-> TVar TxFlow
-> IORef RxFlow
-> IORef StreamState
-> IORef StreamState
-> RecvStreamQ
-> IORef (Skew RxStreamData)
-> MVar ()
-> Stream
Stream StreamId
sid Connection
conn
        (TVar TxFlow
 -> IORef RxFlow
 -> IORef StreamState
 -> IORef StreamState
 -> RecvStreamQ
 -> IORef (Skew RxStreamData)
 -> MVar ()
 -> Stream)
-> IO (TVar TxFlow)
-> IO
     (IORef RxFlow
      -> IORef StreamState
      -> IORef StreamState
      -> RecvStreamQ
      -> IORef (Skew RxStreamData)
      -> MVar ()
      -> Stream)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TxFlow -> IO (TVar TxFlow)
forall (m :: * -> *) a. MonadIO m => a -> m (TVar a)
newTVarIO (StreamId -> TxFlow
newTxFlow StreamId
txLim)
        IO
  (IORef RxFlow
   -> IORef StreamState
   -> IORef StreamState
   -> RecvStreamQ
   -> IORef (Skew RxStreamData)
   -> MVar ()
   -> Stream)
-> IO (IORef RxFlow)
-> IO
     (IORef StreamState
      -> IORef StreamState
      -> RecvStreamQ
      -> IORef (Skew RxStreamData)
      -> MVar ()
      -> 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
rxLim)
        IO
  (IORef StreamState
   -> IORef StreamState
   -> RecvStreamQ
   -> IORef (Skew RxStreamData)
   -> MVar ()
   -> Stream)
-> IO (IORef StreamState)
-> IO
     (IORef StreamState
      -> RecvStreamQ -> IORef (Skew RxStreamData) -> MVar () -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> StreamState -> IO (IORef StreamState)
forall a. a -> IO (IORef a)
newIORef StreamState
emptyStreamState
        IO
  (IORef StreamState
   -> RecvStreamQ -> IORef (Skew RxStreamData) -> MVar () -> Stream)
-> IO (IORef StreamState)
-> IO
     (RecvStreamQ -> IORef (Skew RxStreamData) -> MVar () -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> StreamState -> IO (IORef StreamState)
forall a. a -> IO (IORef a)
newIORef StreamState
emptyStreamState
        IO (RecvStreamQ -> IORef (Skew RxStreamData) -> MVar () -> Stream)
-> IO RecvStreamQ
-> IO (IORef (Skew RxStreamData) -> MVar () -> 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 RecvStreamQ
newRecvStreamQ
        IO (IORef (Skew RxStreamData) -> MVar () -> Stream)
-> IO (IORef (Skew RxStreamData)) -> IO (MVar () -> Stream)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Skew RxStreamData -> IO (IORef (Skew RxStreamData))
forall a. a -> IO (IORef a)
newIORef Skew RxStreamData
forall a. Skew a
Skew.empty
        IO (MVar () -> Stream) -> IO (MVar ()) -> 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
<*> IO (MVar ())
forall (m :: * -> *) a. MonadIO m => m (MVar a)
newEmptyMVar

syncFinTx :: Stream -> IO ()
syncFinTx :: Stream -> IO ()
syncFinTx Stream
s = 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 () -> () -> IO Bool
forall (m :: * -> *) a. MonadIO m => MVar a -> a -> m Bool
tryPutMVar (Stream -> MVar ()
streamSyncFinTx Stream
s) ()

waitFinTx :: Stream -> IO ()
waitFinTx :: Stream -> IO ()
waitFinTx Stream
s = MVar () -> IO ()
forall (m :: * -> *) a. MonadIO m => MVar a -> m a
takeMVar (MVar () -> IO ()) -> MVar () -> IO ()
forall a b. (a -> b) -> a -> b
$ Stream -> MVar ()
streamSyncFinTx Stream
s

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

type Length = Int

data TxStreamData = TxStreamData Stream [StreamData] Length Fin

data RxStreamData = RxStreamData
    { RxStreamData -> StreamData
rxstrmData :: StreamData
    , RxStreamData -> StreamId
rxstrmOff :: Offset
    , RxStreamData -> StreamId
rxstrmLen :: Length
    , RxStreamData -> Bool
rxstrmFin :: Fin
    }
    deriving (RxStreamData -> RxStreamData -> Bool
(RxStreamData -> RxStreamData -> Bool)
-> (RxStreamData -> RxStreamData -> Bool) -> Eq RxStreamData
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: RxStreamData -> RxStreamData -> Bool
== :: RxStreamData -> RxStreamData -> Bool
$c/= :: RxStreamData -> RxStreamData -> Bool
/= :: RxStreamData -> RxStreamData -> Bool
Eq, StreamId -> RxStreamData -> ShowS
[RxStreamData] -> ShowS
RxStreamData -> String
(StreamId -> RxStreamData -> ShowS)
-> (RxStreamData -> String)
-> ([RxStreamData] -> ShowS)
-> Show RxStreamData
forall a.
(StreamId -> a -> ShowS)
-> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: StreamId -> RxStreamData -> ShowS
showsPrec :: StreamId -> RxStreamData -> ShowS
$cshow :: RxStreamData -> String
show :: RxStreamData -> String
$cshowList :: [RxStreamData] -> ShowS
showList :: [RxStreamData] -> ShowS
Show)

instance Frag RxStreamData where
    currOff :: RxStreamData -> StreamId
currOff RxStreamData
r = RxStreamData -> StreamId
rxstrmOff RxStreamData
r
    nextOff :: RxStreamData -> StreamId
nextOff RxStreamData
r = RxStreamData -> StreamId
rxstrmOff RxStreamData
r StreamId -> StreamId -> StreamId
forall a. Num a => a -> a -> a
+ RxStreamData -> StreamId
rxstrmLen RxStreamData
r
    shrink :: StreamId -> RxStreamData -> RxStreamData
shrink StreamId
off' (RxStreamData StreamData
bs StreamId
off StreamId
len Bool
fin) =
        let n :: StreamId
n = StreamId
off' StreamId -> StreamId -> StreamId
forall a. Num a => a -> a -> a
- StreamId
off
            bs' :: StreamData
bs' = StreamId -> StreamData -> StreamData
BS.drop StreamId
n StreamData
bs
            len' :: StreamId
len' = StreamId
len StreamId -> StreamId -> StreamId
forall a. Num a => a -> a -> a
- StreamId
n
         in StreamData -> StreamId -> StreamId -> Bool -> RxStreamData
RxStreamData StreamData
bs' StreamId
off' StreamId
len' Bool
fin

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

data StreamState = StreamState
    { StreamState -> StreamId
streamOffset :: Offset
    , StreamState -> Bool
streamFin :: Fin
    }
    deriving (StreamState -> StreamState -> Bool
(StreamState -> StreamState -> Bool)
-> (StreamState -> StreamState -> Bool) -> Eq StreamState
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: StreamState -> StreamState -> Bool
== :: StreamState -> StreamState -> Bool
$c/= :: StreamState -> StreamState -> Bool
/= :: StreamState -> StreamState -> Bool
Eq, StreamId -> StreamState -> ShowS
[StreamState] -> ShowS
StreamState -> String
(StreamId -> StreamState -> ShowS)
-> (StreamState -> String)
-> ([StreamState] -> ShowS)
-> Show StreamState
forall a.
(StreamId -> a -> ShowS)
-> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: StreamId -> StreamState -> ShowS
showsPrec :: StreamId -> StreamState -> ShowS
$cshow :: StreamState -> String
show :: StreamState -> String
$cshowList :: [StreamState] -> ShowS
showList :: [StreamState] -> ShowS
Show)

emptyStreamState :: StreamState
emptyStreamState :: StreamState
emptyStreamState = StreamId -> Bool -> StreamState
StreamState StreamId
0 Bool
False

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

data RecvStreamQ = RecvStreamQ
    { RecvStreamQ -> TQueue StreamData
recvStreamQ :: TQueue ByteString
    , RecvStreamQ -> IORef (Maybe StreamData)
pendingData :: IORef (Maybe ByteString)
    , RecvStreamQ -> IORef Bool
endOfStream :: IORef Bool
    }

newRecvStreamQ :: IO RecvStreamQ
newRecvStreamQ :: IO RecvStreamQ
newRecvStreamQ = TQueue StreamData
-> IORef (Maybe StreamData) -> IORef Bool -> RecvStreamQ
RecvStreamQ (TQueue StreamData
 -> IORef (Maybe StreamData) -> IORef Bool -> RecvStreamQ)
-> IO (TQueue StreamData)
-> IO (IORef (Maybe StreamData) -> IORef Bool -> RecvStreamQ)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> IO (TQueue StreamData)
forall (m :: * -> *) a. MonadIO m => m (TQueue a)
newTQueueIO IO (IORef (Maybe StreamData) -> IORef Bool -> RecvStreamQ)
-> IO (IORef (Maybe StreamData)) -> IO (IORef Bool -> RecvStreamQ)
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe StreamData -> IO (IORef (Maybe StreamData))
forall a. a -> IO (IORef a)
newIORef Maybe StreamData
forall a. Maybe a
Nothing IO (IORef Bool -> RecvStreamQ) -> IO (IORef Bool) -> IO RecvStreamQ
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 (IORef Bool)
forall a. a -> IO (IORef a)
newIORef Bool
False