-- |
-- Module      : Streamly.Internal.Network.Inet.TCP
-- Copyright   : (c) 2019 Composewell Technologies
--
-- License     : BSD3
-- Maintainer  : streamly@composewell.com
-- Stability   : experimental
-- Portability : GHC
--
-- Combinators to build Inet/TCP clients and servers.

module Streamly.Internal.Network.Inet.TCP
    (
    -- * Setup
    -- | To execute the code examples provided in this module in ghci, please
    -- run the following commands first.
    --
    -- $setup

    -- * TCP Servers
    -- ** Streams
      accept
    , acceptLocal
    , acceptOnAddr
    , acceptOnAddrWith

    -- ** Unfolds
    , acceptor
    , acceptorLocal
    , acceptorWith
    , acceptorOnAddr
    , acceptorOnAddrWith

    -- * TCP clients
    -- | IP Address based operations.
    , connect
    , withConnectionM

    -- ** Unfolds
    , usingConnection
    , reader

    -- ** Streams
    , withConnection
    -- *** Source
    , read
    -- , readUtf8
    -- , readLines
    -- , readFrames
    -- , readByChunks

    -- -- * Array Read
    -- , readArrayUpto
    -- , readArrayOf

    -- , readChunksUpto
    -- , readChunksOf
    -- , readChunks

    -- *** Sink
    , write
    -- , writeUtf8
    -- , writeUtf8ByLines
    -- , writeByFrames
    , writeWithBufferOf
    , putBytes
    , putBytesWithBufferOf

    -- -- * Array Write
    -- , writeArray
    , writeChunks
    , putChunks

    -- ** Transformation
    , pipeBytes
    {-
    -- ** Sink Servers

    -- These abstractions can be applied to any setting where we need to do a
    -- sink processing of multiple streams e.g. output from multiple processes
    -- or data coming from multiple files.

    -- handle connections concurrently using a specified fold
    -- , handleConnections

    -- handle frames concurrently using a specified fold
    , handleFrames

    -- merge frames from all connection into a single stream. Frames can be
    -- created by a specified fold.
    , mergeFrames

    -- * UDP Servers
    , datagrams
    , datagramsOn
    -}

    -- * Deprecated
    , acceptorOnPort
    , acceptorOnPortLocal
    )
where

#include "inline.hs"

import Control.Exception (onException)
import Control.Monad.Catch (MonadCatch, MonadMask, bracket)
import Control.Monad.IO.Class (MonadIO(..))
import Data.Word (Word8)
import Network.Socket
       (Socket, PortNumber, SocketOption(..), Family(..), SockAddr(..),
        SocketType(..), defaultProtocol, maxListenQueue, tupleToHostAddress,
        socket)
import Prelude hiding (read)

import Streamly.Internal.Control.Concurrent (MonadAsync)
import Streamly.Internal.Control.ForkLifted (fork)
import Streamly.Data.Array (Array)
import Streamly.Internal.Data.Fold ( Fold(..) )
import Streamly.Data.Stream (Stream)
import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
import Streamly.Data.MutByteArray (Unbox)
import Streamly.Data.Unfold (Unfold)
import Streamly.Internal.Network.Socket (SockSpec(..))
import Streamly.Internal.System.IO (defaultChunkSize)

import qualified Control.Monad.Catch as MC
import qualified Network.Socket as Net

import qualified Streamly.Data.Array as A
import qualified Streamly.Data.Fold as FL
import qualified Streamly.Data.Stream as S
import qualified Streamly.Data.Unfold as UF
import qualified Streamly.Internal.Data.Array as A
    (pinnedChunksOf, unsafePinnedCreateOf)
import qualified Streamly.Internal.Data.Unfold as UF (bracketIO)
import qualified Streamly.Internal.Data.Fold as FL (Step(..), reduce)

import qualified Streamly.Internal.Data.Stream.Lifted as S (bracket)
import qualified Streamly.Internal.Network.Socket as ISK

-- $setup
-- >>> :m
--
-- >>> import qualified Streamly.Data.Unfold as Unfold
-- >>> import qualified Streamly.Network.Inet.TCP as TCP

-------------------------------------------------------------------------------
-- Accept (unfolds)
-------------------------------------------------------------------------------

{-# INLINE acceptorOnAddrWith #-}
acceptorOnAddrWith
    :: MonadIO m
    => [(SocketOption, Int)]
    -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddrWith :: forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)]
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddrWith [(SocketOption, Int)]
opts = (((Word8, Word8, Word8, Word8), PortNumber)
 -> (Int, SockSpec, SockAddr))
-> Unfold m (Int, SockSpec, SockAddr) Socket
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
forall a c (m :: * -> *) b.
(a -> c) -> Unfold m c b -> Unfold m a b
UF.lmap ((Word8, Word8, Word8, Word8), PortNumber)
-> (Int, SockSpec, SockAddr)
f Unfold m (Int, SockSpec, SockAddr) Socket
forall (m :: * -> *).
MonadIO m =>
Unfold m (Int, SockSpec, SockAddr) Socket
ISK.acceptor
    where
    f :: ((Word8, Word8, Word8, Word8), PortNumber)
-> (Int, SockSpec, SockAddr)
f ((Word8, Word8, Word8, Word8)
addr, PortNumber
port) =
        (Int
maxListenQueue
        , SockSpec
            { sockFamily :: Family
sockFamily = Family
AF_INET
            , sockType :: SocketType
sockType = SocketType
Stream
            , sockProto :: ProtocolNumber
sockProto = ProtocolNumber
defaultProtocol -- TCP
            , sockOpts :: [(SocketOption, Int)]
sockOpts = [(SocketOption, Int)]
opts
            }
        , PortNumber -> HostAddress -> SockAddr
SockAddrInet PortNumber
port ((Word8, Word8, Word8, Word8) -> HostAddress
tupleToHostAddress (Word8, Word8, Word8, Word8)
addr)
        )

-- | Unfold a tuple @(ipAddr, port)@ into a stream of connected TCP sockets.
-- @ipAddr@ is the local IP address and @port@ is the local port on which
-- connections are accepted.
--
{-# INLINE acceptorOnAddr #-}
acceptorOnAddr
    :: MonadIO m
    => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddr :: forall (m :: * -> *).
MonadIO m =>
Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddr = [(SocketOption, Int)]
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)]
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddrWith []

{-# INLINE acceptorWith #-}
acceptorWith :: MonadIO m
    => [(SocketOption, Int)]
    -> Unfold m PortNumber Socket
acceptorWith :: forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)] -> Unfold m PortNumber Socket
acceptorWith [(SocketOption, Int)]
opts = (Word8, Word8, Word8, Word8)
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
-> Unfold m PortNumber Socket
forall a (m :: * -> *) b c. a -> Unfold m (a, b) c -> Unfold m b c
UF.first (Word8
0,Word8
0,Word8
0,Word8
0) ([(SocketOption, Int)]
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)]
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddrWith [(SocketOption, Int)]
opts)

-- | Like 'acceptorOnAddr' but binds on the IPv4 address @0.0.0.0@ i.e.  on all
-- IPv4 addresses/interfaces of the machine and listens for TCP connections on
-- the specified port.
--
-- >>> acceptor = Unfold.first (0,0,0,0) TCP.acceptorOnAddr
--
{-# INLINE acceptor #-}
acceptor :: MonadIO m => Unfold m PortNumber Socket
acceptor :: forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptor = (Word8, Word8, Word8, Word8)
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
-> Unfold m PortNumber Socket
forall a (m :: * -> *) b c. a -> Unfold m (a, b) c -> Unfold m b c
UF.first (Word8
0,Word8
0,Word8
0,Word8
0) Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
forall (m :: * -> *).
MonadIO m =>
Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddr

{-# DEPRECATED acceptorOnPort "Use \"acceptor\" instead." #-}
{-# INLINE acceptorOnPort #-}
acceptorOnPort :: MonadIO m => Unfold m PortNumber Socket
acceptorOnPort :: forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptorOnPort = Unfold m PortNumber Socket
forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptor

-- | Like 'acceptor' but binds on the localhost IPv4 address @127.0.0.1@. The
-- server can only be accessed from the local host, it cannot be accessed from
-- other hosts on the network.
--
-- >>> acceptorLocal = Unfold.first (127,0,0,1) TCP.acceptorOnAddr
--
{-# INLINE acceptorLocal #-}
acceptorLocal :: MonadIO m => Unfold m PortNumber Socket
acceptorLocal :: forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptorLocal = (Word8, Word8, Word8, Word8)
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
-> Unfold m PortNumber Socket
forall a (m :: * -> *) b c. a -> Unfold m (a, b) c -> Unfold m b c
UF.first (Word8
127,Word8
0,Word8
0,Word8
1) Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
forall (m :: * -> *).
MonadIO m =>
Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket
acceptorOnAddr

{-# DEPRECATED acceptorOnPortLocal "Use \"acceptorLocal\" instead." #-}
{-# INLINE acceptorOnPortLocal #-}
acceptorOnPortLocal :: MonadIO m => Unfold m PortNumber Socket
acceptorOnPortLocal :: forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptorOnPortLocal = Unfold m PortNumber Socket
forall (m :: * -> *). MonadIO m => Unfold m PortNumber Socket
acceptorLocal

-------------------------------------------------------------------------------
-- Accept (streams)
-------------------------------------------------------------------------------

-- | Like 'acceptOnAddr' but with the ability to specify a list of socket
-- options.
--
-- /Pre-release/
{-# INLINE acceptOnAddrWith #-}
acceptOnAddrWith
    :: MonadIO m
    => [(SocketOption, Int)]
    -> (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m Socket
acceptOnAddrWith :: forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)]
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
acceptOnAddrWith [(SocketOption, Int)]
opts (Word8, Word8, Word8, Word8)
addr PortNumber
port =
    Int -> SockSpec -> SockAddr -> Stream m Socket
forall (m :: * -> *).
MonadIO m =>
Int -> SockSpec -> SockAddr -> Stream m Socket
ISK.accept Int
maxListenQueue SockSpec
        { sockFamily :: Family
sockFamily = Family
AF_INET
        , sockType :: SocketType
sockType = SocketType
Stream
        , sockProto :: ProtocolNumber
sockProto = ProtocolNumber
defaultProtocol
        , sockOpts :: [(SocketOption, Int)]
sockOpts = [(SocketOption, Int)]
opts
        }
        (PortNumber -> HostAddress -> SockAddr
SockAddrInet PortNumber
port ((Word8, Word8, Word8, Word8) -> HostAddress
tupleToHostAddress (Word8, Word8, Word8, Word8)
addr))

-- | Like 'accept' but binds on the specified IPv4 address.
--
-- >>> acceptOnAddr = TCP.acceptOnAddrWith []
--
-- /Pre-release/
{-# INLINE acceptOnAddr #-}
acceptOnAddr
    :: MonadIO m
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m Socket
acceptOnAddr :: forall (m :: * -> *).
MonadIO m =>
(Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
acceptOnAddr = [(SocketOption, Int)]
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
forall (m :: * -> *).
MonadIO m =>
[(SocketOption, Int)]
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
acceptOnAddrWith []

-- | Start a TCP stream server that binds on the IPV4 address @0.0.0.0@ and
-- listens for TCP connections from remote hosts on the specified server port.
-- The server generates a stream of connected sockets.
--
-- >>> accept = TCP.acceptOnAddr (0,0,0,0)
--
-- /Pre-release/
{-# INLINE accept #-}
accept :: MonadIO m => PortNumber -> Stream m Socket
accept :: forall (m :: * -> *). MonadIO m => PortNumber -> Stream m Socket
accept = (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
forall (m :: * -> *).
MonadIO m =>
(Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
acceptOnAddr (Word8
0,Word8
0,Word8
0,Word8
0)

-- | Like 'accept' but binds on the localhost IPv4 address @127.0.0.1@.
-- The server can only be accessed from the local host, it cannot be accessed
-- from other hosts on the network.
--
-- >>> acceptLocal = TCP.acceptOnAddr (127,0,0,1)
--
-- /Pre-release/
{-# INLINE acceptLocal #-}
acceptLocal :: MonadIO m => PortNumber -> Stream m Socket
acceptLocal :: forall (m :: * -> *). MonadIO m => PortNumber -> Stream m Socket
acceptLocal = (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
forall (m :: * -> *).
MonadIO m =>
(Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Socket
acceptOnAddr (Word8
127,Word8
0,Word8
0,Word8
1)

-------------------------------------------------------------------------------
-- TCP Clients
-------------------------------------------------------------------------------

-- | Connect to the specified IP address and port number. Returns a connected
-- socket or throws an exception.
--
connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect :: (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port = do
    Socket
sock <- Family -> SocketType -> ProtocolNumber -> IO Socket
socket Family
AF_INET SocketType
Stream ProtocolNumber
defaultProtocol
    Socket -> SockAddr -> IO ()
Net.connect Socket
sock (PortNumber -> HostAddress -> SockAddr
SockAddrInet PortNumber
port ((Word8, Word8, Word8, Word8) -> HostAddress
Net.tupleToHostAddress (Word8, Word8, Word8, Word8)
addr))
        IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
`onException` Socket -> IO ()
Net.close Socket
sock
    Socket -> IO Socket
forall a. a -> IO a
forall (m :: * -> *) a. Monad m => a -> m a
return Socket
sock

-- | Connect to a remote host using IP address and port and run the supplied
-- action on the resulting socket.  'withConnectionM' makes sure that the
-- socket is closed on normal termination or in case of an exception.  If
-- closing the socket raises an exception, then this exception will be raised
-- by 'withConnectionM'.
--
-- /Pre-release/
{-# INLINABLE withConnectionM #-}
withConnectionM :: (MonadMask m, MonadIO m)
    => (Word8, Word8, Word8, Word8) -> PortNumber -> (Socket -> m ()) -> m ()
withConnectionM :: forall (m :: * -> *).
(MonadMask m, MonadIO m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> (Socket -> m ()) -> m ()
withConnectionM (Word8, Word8, Word8, Word8)
addr PortNumber
port =
    m Socket -> (Socket -> m ()) -> (Socket -> m ()) -> m ()
forall (m :: * -> *) a c b.
MonadMask m =>
m a -> (a -> m c) -> (a -> m b) -> m b
bracket (IO Socket -> m Socket
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Socket -> m Socket) -> IO Socket -> m Socket
forall a b. (a -> b) -> a -> b
$ (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port) (IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> (Socket -> IO ()) -> Socket -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Socket -> IO ()
Net.close)

-------------------------------------------------------------------------------
-- Connect (unfolds)
-------------------------------------------------------------------------------

-- | Transform an 'Unfold' from a 'Socket' to an unfold from a remote IP
-- address and port. The resulting unfold opens a socket, uses it using the
-- supplied unfold and then makes sure that the socket is closed on normal
-- termination or in case of an exception.  If closing the socket raises an
-- exception, then this exception will be raised by 'usingConnection'.
--
-- /Pre-release/
{-# INLINE usingConnection #-}
usingConnection :: (MonadCatch m, MonadAsync m)
    => Unfold m Socket a
    -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
usingConnection :: forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
Unfold m Socket a
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
usingConnection = (((Word8, Word8, Word8, Word8), PortNumber) -> IO Socket)
-> (Socket -> IO ())
-> Unfold m Socket a
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
forall (m :: * -> *) a c d b.
(MonadIO m, MonadCatch m) =>
(a -> IO c) -> (c -> IO d) -> Unfold m c b -> Unfold m a b
UF.bracketIO (\((Word8, Word8, Word8, Word8)
addr, PortNumber
port) -> (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port) Socket -> IO ()
Net.close

-------------------------------------------------------------------------------
-- Connect (streams)
-------------------------------------------------------------------------------

-- | @'withConnection' addr port act@ opens a connection to the specified IPv4
-- host address and port and passes the resulting socket handle to the
-- computation @act@.  The handle will be closed on exit from 'withConnection',
-- whether by normal termination or by raising an exception.  If closing the
-- handle raises an exception, then this exception will be raised by
-- 'withConnection' rather than any exception raised by 'act'.
--
-- /Pre-release/
{-# INLINE withConnection #-}
withConnection :: (MonadCatch m, MonadAsync m)
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> (Socket -> Stream m a)
    -> Stream m a
withConnection :: forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> (Socket -> Stream m a) -> Stream m a
withConnection (Word8, Word8, Word8, Word8)
addr PortNumber
port = IO Socket
-> (Socket -> IO ()) -> (Socket -> Stream m a) -> Stream m a
forall (m :: * -> *) b c a.
(MonadIO m, MonadCatch m) =>
IO b -> (b -> IO c) -> (b -> Stream m a) -> Stream m a
S.bracketIO ((Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port) Socket -> IO ()
Net.close

-------------------------------------------------------------------------------
-- Read Addr to Stream
-------------------------------------------------------------------------------

-- | Read a stream from the supplied IPv4 host address and port number.
--
{-# INLINE reader #-}
reader :: (MonadCatch m, MonadAsync m)
    => Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
reader :: forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
reader = Unfold m (Array Word8) Word8
-> Unfold
     m ((Word8, Word8, Word8, Word8), PortNumber) (Array Word8)
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Word8
forall (m :: * -> *) b c a.
Monad m =>
Unfold m b c -> Unfold m a b -> Unfold m a c
UF.many Unfold m (Array Word8) Word8
forall (m :: * -> *) a. (Monad m, Unbox a) => Unfold m (Array a) a
A.reader (Unfold m Socket (Array Word8)
-> Unfold
     m ((Word8, Word8, Word8, Word8), PortNumber) (Array Word8)
forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
Unfold m Socket a
-> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) a
usingConnection Unfold m Socket (Array Word8)
forall (m :: * -> *). MonadIO m => Unfold m Socket (Array Word8)
ISK.chunkReader)

{-# INLINE concatChunks #-}
concatChunks :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a
concatChunks :: forall (m :: * -> *) a.
(Monad m, Unbox a) =>
Stream m (Array a) -> Stream m a
concatChunks = Unfold m (Array a) a -> Stream m (Array a) -> Stream m a
forall (m :: * -> *) a b.
Monad m =>
Unfold m a b -> Stream m a -> Stream m b
S.unfoldMany Unfold m (Array a) a
forall (m :: * -> *) a. (Monad m, Unbox a) => Unfold m (Array a) a
A.reader

-- | Read a stream from the supplied IPv4 host address and port number.
--
-- /Pre-release/
{-# INLINE read #-}
read :: (MonadCatch m, MonadAsync m)
    => (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Word8
read :: forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Word8
read (Word8, Word8, Word8, Word8)
addr PortNumber
port = Stream m (Array Word8) -> Stream m Word8
forall (m :: * -> *) a.
(Monad m, Unbox a) =>
Stream m (Array a) -> Stream m a
concatChunks (Stream m (Array Word8) -> Stream m Word8)
-> Stream m (Array Word8) -> Stream m Word8
forall a b. (a -> b) -> a -> b
$ (Word8, Word8, Word8, Word8)
-> PortNumber
-> (Socket -> Stream m (Array Word8))
-> Stream m (Array Word8)
forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> (Socket -> Stream m a) -> Stream m a
withConnection (Word8, Word8, Word8, Word8)
addr PortNumber
port Socket -> Stream m (Array Word8)
forall (m :: * -> *). MonadIO m => Socket -> Stream m (Array Word8)
ISK.readChunks

-------------------------------------------------------------------------------
-- Writing
-------------------------------------------------------------------------------

-- | Write a stream of arrays to the supplied IPv4 host address and port
-- number.
--
-- /Pre-release/
{-# INLINE putChunks #-}
putChunks
    :: (MonadCatch m, MonadAsync m)
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m (Array Word8)
    -> m ()
putChunks :: forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Stream m (Array Word8) -> m ()
putChunks (Word8, Word8, Word8, Word8)
addr PortNumber
port Stream m (Array Word8)
xs =
    Fold m () () -> Stream m () -> m ()
forall (m :: * -> *) a b.
Monad m =>
Fold m a b -> Stream m a -> m b
S.fold Fold m () ()
forall (m :: * -> *) a. Monad m => Fold m a ()
FL.drain
        (Stream m () -> m ()) -> Stream m () -> m ()
forall a b. (a -> b) -> a -> b
$ (Word8, Word8, Word8, Word8)
-> PortNumber -> (Socket -> Stream m ()) -> Stream m ()
forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> (Socket -> Stream m a) -> Stream m a
withConnection (Word8, Word8, Word8, Word8)
addr PortNumber
port (\Socket
sk -> m () -> Stream m ()
forall (m :: * -> *) a. Applicative m => m a -> Stream m a
S.fromEffect (m () -> Stream m ()) -> m () -> Stream m ()
forall a b. (a -> b) -> a -> b
$ Socket -> Stream m (Array Word8) -> m ()
forall (m :: * -> *) a.
(MonadIO m, Unbox a) =>
Socket -> Stream m (Array a) -> m ()
ISK.putChunks Socket
sk Stream m (Array Word8)
xs)

-- | Write a stream of arrays to the supplied IPv4 host address and port
-- number.
--
{-# INLINE writeChunks #-}
writeChunks
    :: (MonadIO m, MonadCatch m)
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Fold m (Array Word8) ()
writeChunks :: forall (m :: * -> *).
(MonadIO m, MonadCatch m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Fold m (Array Word8) ()
writeChunks (Word8, Word8, Word8, Word8)
addr PortNumber
port = (Tuple' (Fold m (Array Word8) ()) Socket
 -> Array Word8
 -> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) ()))
-> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) ())
-> (Tuple' (Fold m (Array Word8) ()) Socket -> m ())
-> (Tuple' (Fold m (Array Word8) ()) Socket -> m ())
-> Fold m (Array Word8) ()
forall (m :: * -> *) a b s.
(s -> a -> m (Step s b))
-> m (Step s b) -> (s -> m b) -> (s -> m b) -> Fold m a b
Fold Tuple' (Fold m (Array Word8) ()) Socket
-> Array Word8
-> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) ())
forall {m :: * -> *} {a} {b} {b}.
(MonadCatch m, MonadIO m) =>
Tuple' (Fold m a b) Socket
-> a -> m (Step (Tuple' (Fold m a b) Socket) b)
step m (Step (Tuple' (Fold m (Array Word8) ()) Socket) ())
forall {b}. m (Step (Tuple' (Fold m (Array Word8) ()) Socket) b)
initial Tuple' (Fold m (Array Word8) ()) Socket -> m ()
forall {m :: * -> *} {p}. Monad m => p -> m ()
extract Tuple' (Fold m (Array Word8) ()) Socket -> m ()
forall {m :: * -> *} {a}.
MonadIO m =>
Tuple' (Fold m a ()) Socket -> m ()
final
    where
    initial :: m (Step (Tuple' (Fold m (Array Word8) ()) Socket) b)
initial = do
        Socket
skt <- IO Socket -> m Socket
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO ((Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port)
        Fold m (Array Word8) ()
fld <- Fold m (Array Word8) () -> m (Fold m (Array Word8) ())
forall (m :: * -> *) a b. Monad m => Fold m a b -> m (Fold m a b)
FL.reduce (Socket -> Fold m (Array Word8) ()
forall (m :: * -> *) a.
(MonadIO m, Unbox a) =>
Socket -> Fold m (Array a) ()
ISK.writeChunks Socket
skt)
                    m (Fold m (Array Word8) ()) -> m () -> m (Fold m (Array Word8) ())
forall (m :: * -> *) a b. MonadCatch m => m a -> m b -> m a
`MC.onException` IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Socket -> IO ()
Net.close Socket
skt)
        Step (Tuple' (Fold m (Array Word8) ()) Socket) b
-> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) b)
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return (Step (Tuple' (Fold m (Array Word8) ()) Socket) b
 -> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) b))
-> Step (Tuple' (Fold m (Array Word8) ()) Socket) b
-> m (Step (Tuple' (Fold m (Array Word8) ()) Socket) b)
forall a b. (a -> b) -> a -> b
$ Tuple' (Fold m (Array Word8) ()) Socket
-> Step (Tuple' (Fold m (Array Word8) ()) Socket) b
forall s b. s -> Step s b
FL.Partial (Fold m (Array Word8) ()
-> Socket -> Tuple' (Fold m (Array Word8) ()) Socket
forall a b. a -> b -> Tuple' a b
Tuple' Fold m (Array Word8) ()
fld Socket
skt)
    step :: Tuple' (Fold m a b) Socket
-> a -> m (Step (Tuple' (Fold m a b) Socket) b)
step (Tuple' Fold m a b
fld Socket
skt) a
x = do
        Fold m a b
r <- a -> Fold m a b -> m (Fold m a b)
forall (m :: * -> *) a b.
Monad m =>
a -> Fold m a b -> m (Fold m a b)
FL.addOne a
x Fold m a b
fld m (Fold m a b) -> m () -> m (Fold m a b)
forall (m :: * -> *) a b. MonadCatch m => m a -> m b -> m a
`MC.onException` IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (Socket -> IO ()
Net.close Socket
skt)
        Step (Tuple' (Fold m a b) Socket) b
-> m (Step (Tuple' (Fold m a b) Socket) b)
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return (Step (Tuple' (Fold m a b) Socket) b
 -> m (Step (Tuple' (Fold m a b) Socket) b))
-> Step (Tuple' (Fold m a b) Socket) b
-> m (Step (Tuple' (Fold m a b) Socket) b)
forall a b. (a -> b) -> a -> b
$ Tuple' (Fold m a b) Socket -> Step (Tuple' (Fold m a b) Socket) b
forall s b. s -> Step s b
FL.Partial (Fold m a b -> Socket -> Tuple' (Fold m a b) Socket
forall a b. a -> b -> Tuple' a b
Tuple' Fold m a b
r Socket
skt)

    extract :: p -> m ()
extract p
_ = () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

    final :: Tuple' (Fold m a ()) Socket -> m ()
final (Tuple' (Fold s -> a -> m (Step s ())
_ m (Step s ())
initial1 s -> m ()
_ s -> m ()
final1) Socket
skt) = do
        IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Socket -> IO ()
Net.close Socket
skt
        Step s ()
res <- m (Step s ())
initial1
        case Step s ()
res of
            FL.Partial s
fs -> s -> m ()
final1 s
fs
            FL.Done () -> () -> m ()
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return ()

-- | Like 'write' but provides control over the write buffer. Output will
-- be written to the IO device as soon as we collect the specified number of
-- input elements.
--
-- /Pre-release/
{-# INLINE putBytesWithBufferOf #-}
putBytesWithBufferOf
    :: (MonadCatch m, MonadAsync m)
    => Int
    -> (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m Word8
    -> m ()
putBytesWithBufferOf :: forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
Int
-> (Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> m ()
putBytesWithBufferOf Int
n (Word8, Word8, Word8, Word8)
addr PortNumber
port Stream m Word8
m =
    (Word8, Word8, Word8, Word8)
-> PortNumber -> Stream m (Array Word8) -> m ()
forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Stream m (Array Word8) -> m ()
putChunks (Word8, Word8, Word8, Word8)
addr PortNumber
port (Stream m (Array Word8) -> m ()) -> Stream m (Array Word8) -> m ()
forall a b. (a -> b) -> a -> b
$ Int -> Stream m Word8 -> Stream m (Array Word8)
forall (m :: * -> *) a.
(MonadIO m, Unbox a) =>
Int -> Stream m a -> Stream m (Array a)
A.pinnedChunksOf Int
n Stream m Word8
m

-- | Like 'write' but provides control over the write buffer. Output will
-- be written to the IO device as soon as we collect the specified number of
-- input elements.
--
{-# INLINE writeWithBufferOf #-}
writeWithBufferOf
    :: (MonadIO m, MonadCatch m)
    => Int
    -> (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Fold m Word8 ()
writeWithBufferOf :: forall (m :: * -> *).
(MonadIO m, MonadCatch m) =>
Int
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
writeWithBufferOf Int
n (Word8, Word8, Word8, Word8)
addr PortNumber
port =
    Int
-> Fold m Word8 (Array Word8)
-> Fold m (Array Word8) ()
-> Fold m Word8 ()
forall (m :: * -> *) a b c.
Monad m =>
Int -> Fold m a b -> Fold m b c -> Fold m a c
FL.groupsOf Int
n (Int -> Fold m Word8 (Array Word8)
forall (m :: * -> *) a.
(MonadIO m, Unbox a) =>
Int -> Fold m a (Array a)
A.unsafePinnedCreateOf Int
n) ((Word8, Word8, Word8, Word8)
-> PortNumber -> Fold m (Array Word8) ()
forall (m :: * -> *).
(MonadIO m, MonadCatch m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Fold m (Array Word8) ()
writeChunks (Word8, Word8, Word8, Word8)
addr PortNumber
port)

-- | Write a stream to the supplied IPv4 host address and port number.
--
-- /Pre-release/
{-# INLINE putBytes #-}
putBytes :: (MonadCatch m, MonadAsync m)
    => (Word8, Word8, Word8, Word8) -> PortNumber -> Stream m Word8 -> m ()
putBytes :: forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Stream m Word8 -> m ()
putBytes = Int
-> (Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> m ()
forall (m :: * -> *).
(MonadCatch m, MonadAsync m) =>
Int
-> (Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> m ()
putBytesWithBufferOf Int
defaultChunkSize

-- | Write a stream to the supplied IPv4 host address and port number.
--
{-# INLINE write #-}
write :: (MonadIO m, MonadCatch m)
    => (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
write :: forall (m :: * -> *).
(MonadIO m, MonadCatch m) =>
(Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
write = Int
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
forall (m :: * -> *).
(MonadIO m, MonadCatch m) =>
Int
-> (Word8, Word8, Word8, Word8) -> PortNumber -> Fold m Word8 ()
writeWithBufferOf Int
defaultChunkSize

-------------------------------------------------------------------------------
-- Transformations
-------------------------------------------------------------------------------

{-# INLINE withInputConnect #-}
withInputConnect
    :: (MonadCatch m, MonadAsync m)
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m Word8
    -> (Socket -> Stream m a)
    -> Stream m a
withInputConnect :: forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> (Socket -> Stream m a)
-> Stream m a
withInputConnect (Word8, Word8, Word8, Word8)
addr PortNumber
port Stream m Word8
input Socket -> Stream m a
f = m (Socket, ThreadId)
-> ((Socket, ThreadId) -> m ())
-> ((Socket, ThreadId) -> Stream m a)
-> Stream m a
forall (m :: * -> *) b c a.
(MonadAsync m, MonadCatch m) =>
m b -> (b -> m c) -> (b -> Stream m a) -> Stream m a
S.bracket m (Socket, ThreadId)
pre (Socket, ThreadId) -> m ()
forall {m :: * -> *} {b}. MonadIO m => (Socket, b) -> m ()
post (Socket, ThreadId) -> Stream m a
forall {b}. (Socket, b) -> Stream m a
handler

    where

    pre :: m (Socket, ThreadId)
pre = do
        Socket
sk <- IO Socket -> m Socket
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO Socket -> m Socket) -> IO Socket -> m Socket
forall a b. (a -> b) -> a -> b
$ (Word8, Word8, Word8, Word8) -> PortNumber -> IO Socket
connect (Word8, Word8, Word8, Word8)
addr PortNumber
port
        ThreadId
tid <- m () -> m ThreadId
forall (m :: * -> *). MonadRunInIO m => m () -> m ThreadId
fork (Socket -> Stream m Word8 -> m ()
forall (m :: * -> *). MonadIO m => Socket -> Stream m Word8 -> m ()
ISK.putBytes Socket
sk Stream m Word8
input)
        (Socket, ThreadId) -> m (Socket, ThreadId)
forall a. a -> m a
forall (m :: * -> *) a. Monad m => a -> m a
return (Socket
sk, ThreadId
tid)

    handler :: (Socket, b) -> Stream m a
handler (Socket
sk, b
_) = Socket -> Stream m a
f Socket
sk

    -- XXX kill the thread immediately?
    post :: (Socket, b) -> m ()
post (Socket
sk, b
_) = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ Socket -> IO ()
Net.close Socket
sk

-- | Send an input stream to a remote host and produce the output stream from
-- the host. The server host just acts as a transformation function on the
-- input stream.  Both sending and receiving happen asynchronously.
--
-- /Pre-release/
--
{-# INLINE pipeBytes #-}
pipeBytes
    :: (MonadAsync m, MonadCatch m)
    => (Word8, Word8, Word8, Word8)
    -> PortNumber
    -> Stream m Word8
    -> Stream m Word8
pipeBytes :: forall (m :: * -> *).
(MonadAsync m, MonadCatch m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber -> Stream m Word8 -> Stream m Word8
pipeBytes (Word8, Word8, Word8, Word8)
addr PortNumber
port Stream m Word8
input = (Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> (Socket -> Stream m Word8)
-> Stream m Word8
forall (m :: * -> *) a.
(MonadCatch m, MonadAsync m) =>
(Word8, Word8, Word8, Word8)
-> PortNumber
-> Stream m Word8
-> (Socket -> Stream m a)
-> Stream m a
withInputConnect (Word8, Word8, Word8, Word8)
addr PortNumber
port Stream m Word8
input Socket -> Stream m Word8
forall (m :: * -> *). MonadIO m => Socket -> Stream m Word8
ISK.read