{-# LANGUAGE CPP #-}
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE PolyKinds #-}
{-# LANGUAGE RankNTypes #-}
{-# LANGUAGE ScopedTypeVariables #-}
module Network.Mux.Bearer.Pipe (
PipeChannel (..)
, pipeChannelFromHandles
#if defined(mingw32_HOST_OS)
, pipeChannelFromNamedPipe
#endif
, pipeAsMuxBearer
) where
import Control.Monad.Class.MonadThrow
import Control.Monad.Class.MonadTime.SI
import Control.Tracer
import qualified Data.ByteString.Lazy as BL
import System.IO (Handle, hFlush)
#if defined(mingw32_HOST_OS)
import Data.Foldable (traverse_)
import qualified System.Win32.Types as Win32 (HANDLE)
import qualified System.Win32.Async as Win32.Async
#endif
import Network.Mux.Types (MuxBearer)
import qualified Network.Mux.Types as Mx
import qualified Network.Mux.Trace as Mx
import qualified Network.Mux.Codec as Mx
import qualified Network.Mux.Time as Mx
import qualified Network.Mux.Timeout as Mx
data PipeChannel = PipeChannel {
PipeChannel -> Int -> IO ByteString
readHandle :: Int -> IO BL.ByteString,
PipeChannel -> ByteString -> IO ()
writeHandle :: BL.ByteString -> IO ()
}
pipeChannelFromHandles :: Handle
-> Handle
-> PipeChannel
pipeChannelFromHandles :: Handle -> Handle -> PipeChannel
pipeChannelFromHandles Handle
r Handle
w = PipeChannel {
readHandle :: Int -> IO ByteString
readHandle = Handle -> Int -> IO ByteString
BL.hGet Handle
r,
writeHandle :: ByteString -> IO ()
writeHandle = \ByteString
a -> Handle -> ByteString -> IO ()
BL.hPut Handle
w ByteString
a IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Handle -> IO ()
hFlush Handle
w
}
#if defined(mingw32_HOST_OS)
pipeChannelFromNamedPipe :: Win32.HANDLE
-> PipeChannel
pipeChannelFromNamedPipe h = PipeChannel {
readHandle = fmap BL.fromStrict . Win32.Async.readHandle h,
writeHandle = traverse_ (Win32.Async.writeHandle h) . BL.toChunks
}
#endif
pipeAsMuxBearer
:: Mx.SDUSize
-> Tracer IO Mx.MuxTrace
-> PipeChannel
-> MuxBearer IO
pipeAsMuxBearer :: SDUSize -> Tracer IO MuxTrace -> PipeChannel -> MuxBearer IO
pipeAsMuxBearer SDUSize
sduSize Tracer IO MuxTrace
tracer PipeChannel
channel =
Mx.MuxBearer {
read :: TimeoutFn IO -> IO (MuxSDU, Time)
Mx.read = TimeoutFn IO -> IO (MuxSDU, Time)
readPipe,
write :: TimeoutFn IO -> MuxSDU -> IO Time
Mx.write = TimeoutFn IO -> MuxSDU -> IO Time
writePipe,
sduSize :: SDUSize
Mx.sduSize = SDUSize
sduSize
}
where
readPipe :: Mx.TimeoutFn IO -> IO (Mx.MuxSDU, Time)
readPipe :: TimeoutFn IO -> IO (MuxSDU, Time)
readPipe TimeoutFn IO
_ = do
Tracer IO MuxTrace -> MuxTrace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO MuxTrace
tracer MuxTrace
Mx.MuxTraceRecvHeaderStart
hbuf <- Int -> [ByteString] -> IO ByteString
recvLen' Int
8 []
case Mx.decodeMuxSDU hbuf of
Left MuxError
e -> MuxError -> IO (MuxSDU, Time)
forall e a. Exception e => e -> IO a
forall (m :: * -> *) e a. (MonadThrow m, Exception e) => e -> m a
throwIO MuxError
e
Right header :: MuxSDU
header@Mx.MuxSDU { MuxSDUHeader
msHeader :: MuxSDUHeader
msHeader :: MuxSDU -> MuxSDUHeader
Mx.msHeader } -> do
Tracer IO MuxTrace -> MuxTrace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO MuxTrace
tracer (MuxTrace -> IO ()) -> MuxTrace -> IO ()
forall a b. (a -> b) -> a -> b
$ MuxSDUHeader -> MuxTrace
Mx.MuxTraceRecvHeaderEnd MuxSDUHeader
msHeader
blob <- Int -> [ByteString] -> IO ByteString
recvLen' (Word16 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Word16 -> Int) -> Word16 -> Int
forall a b. (a -> b) -> a -> b
$ MuxSDUHeader -> Word16
Mx.mhLength MuxSDUHeader
msHeader) []
ts <- getMonotonicTime
traceWith tracer (Mx.MuxTraceRecvDeltaQObservation msHeader ts)
return (header {Mx.msBlob = blob}, ts)
recvLen' :: Int -> [BL.ByteString] -> IO BL.ByteString
recvLen' :: Int -> [ByteString] -> IO ByteString
recvLen' Int
0 [ByteString]
bufs = ByteString -> IO ByteString
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return (ByteString -> IO ByteString) -> ByteString -> IO ByteString
forall a b. (a -> b) -> a -> b
$ [ByteString] -> ByteString
BL.concat ([ByteString] -> ByteString) -> [ByteString] -> ByteString
forall a b. (a -> b) -> a -> b
$ [ByteString] -> [ByteString]
forall a. [a] -> [a]
reverse [ByteString]
bufs
recvLen' Int
l [ByteString]
bufs = do
Tracer IO MuxTrace -> MuxTrace -> IO ()
forall (m :: * -> *) a. Tracer m a -> a -> m ()
traceWith Tracer IO MuxTrace
tracer (MuxTrace -> IO ()) -> MuxTrace -> IO ()
forall a b. (a -> b) -> a -> b
$ Int -> MuxTrace
Mx.MuxTraceRecvStart Int
l
buf <- PipeChannel -> Int -> IO ByteString
readHandle PipeChannel
channel Int
l
IO ByteString -> (IOException -> IO ByteString) -> IO ByteString
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` String -> IOException -> IO ByteString
forall (m :: * -> *) a.
MonadThrow m =>
String -> IOException -> m a
Mx.handleIOException String
"readHandle errored"
if BL.null buf
then throwIO $ Mx.MuxError Mx.MuxBearerClosed "Pipe closed when reading data"
else do
traceWith tracer $ Mx.MuxTraceRecvEnd (fromIntegral $ BL.length buf)
recvLen' (l - fromIntegral (BL.length buf)) (buf : bufs)
writePipe :: Mx.TimeoutFn IO -> Mx.MuxSDU -> IO Time
writePipe :: TimeoutFn IO -> MuxSDU -> IO Time
writePipe TimeoutFn IO
_ MuxSDU
sdu = do
ts <- IO Time
forall (m :: * -> *). MonadMonotonicTime m => m Time
getMonotonicTime
let ts32 = Time -> Word32
Mx.timestampMicrosecondsLow32Bits Time
ts
sdu' = MuxSDU -> RemoteClockModel -> MuxSDU
Mx.setTimestamp MuxSDU
sdu (Word32 -> RemoteClockModel
Mx.RemoteClockModel Word32
ts32)
buf = MuxSDU -> ByteString
Mx.encodeMuxSDU MuxSDU
sdu'
traceWith tracer $ Mx.MuxTraceSendStart (Mx.msHeader sdu')
writeHandle channel buf
`catch` Mx.handleIOException "writeHandle errored"
traceWith tracer Mx.MuxTraceSendEnd
return ts