@@ -34,11 +34,12 @@ module Streamly.Network.Socket
3434 (
3535 -- ** Listen for Connections
3636 ServerSpec(..)
37− , recvConnectionsWith
38− , recvConnectionsOn
37+ , recvConnectionsWith
38+ , recvConnectionsOn
3939
4040 -- ** Read a stream from a connection
41− , read
41+ , fromSocket
42+ , read
4243 -- , readUtf8
4344 -- , readLines
4445 -- , readFrames
@@ -66,6 +67,7 @@ module Streamly.Network.Socket
6667where
6768
6869import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)
70+import Control.Monad.Catch (MonadCatch)
6971import Control.Monad.IO.Class (MonadIO(..))
7072import Control.Monad (when)
7173import Data.Word (Word8)
@@ -124,9 +126,10 @@ initListener tcpListenQ ServerSpec{..} =
124126-- generates a stream of connected sockets and the endpoint they are connected
125127-- to. The first argument is the maximum number of pending connections in the
126128-- backlog.
127−{-# INLINE recvConnectionsWith #-}
128−recvConnectionsWith :: MonadAsync m => Int -> ServerSpec -> SerialT m (Socket, SockAddr)
129−recvConnectionsWith tcpListenQ opts = S.unfoldrM step Nothing
129+{-# INLINE recvConnectionTuplesWith #-}
130+recvConnectionTuplesWith :: MonadAsync m
131+ => Int -> ServerSpec -> SerialT m (Socket, SockAddr)
132+recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing
130133 where
131134 step Nothing = do
132135 listener <- liftIO $ initListener tcpListenQ opts
@@ -139,15 +142,29 @@ recvConnectionsWith tcpListenQ opts = S.unfoldrM step Nothing
139142 -- XXX error handling
140143 return $ Just (r, Just listener)
141144
142−{-# INLINE recvConnectionsOn #-}
143−recvConnectionsOn :: MonadAsync m => PortNumber -> SerialT m (Socket, SockAddr)
144−recvConnectionsOn port =
145− recvConnectionsWith maxListenQueue ServerSpec
145+{-# INLINE recvConnectionsWith #-}
146+recvConnectionsWith :: MonadAsync m => Int -> ServerSpec -> SerialT m Socket
147+recvConnectionsWith tcpListenQ opts = fmap fst $
148+ recvConnectionTuplesWith tcpListenQ opts
149+
150+{-# INLINE recvConnectionTuplesOn #-}
151+recvConnectionTuplesOn :: MonadAsync m
152+ => PortNumber -> SerialT m (Socket, SockAddr)
153+recvConnectionTuplesOn port =
154+ recvConnectionTuplesWith maxListenQueue ServerSpec
146155 { serverAddressFamily = AF_INET
147156 , serverAddress = SockAddrInet port 0
148157 , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]
149158 }
150159
160+{-# INLINE recvConnectionsOn #-}
161+recvConnectionsOn :: MonadAsync m => PortNumber -> SerialT m Socket
162+recvConnectionsOn = fmap fst . recvConnectionTuplesOn
163+
164+-- | Read a stream of Word8 from a socket, closing the socket when done.
165+fromSocket :: (MonadCatch m, MonadIO m) => Socket -> SerialT m Word8
166+fromSocket sk = S.finally (liftIO (Net.close sk)) (read sk)
167+
151168-------------------------------------------------------------------------------
152169-- Array IO (Input)
153170-------------------------------------------------------------------------------