677c94d · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Simplify connections' stream API, add fromSocket

677c94dHarendra Kumar7 years ago

src/Streamly/Network/Socket.hs+27−10

compare with master

unifiedsplit

@@ -34,11 +34,12 @@ module Streamly.Network.Socket

34 (

35 -- ** Listen for Connections

36 ServerSpec(..)

37− , recvConnectionsWith

37+ , recvConnectionsWith

38− , recvConnectionsOn

38+ , recvConnectionsOn

40 -- ** Read a stream from a connection

41+ , fromSocket

41− , read

42+ , read

42 -- , readUtf8

43 -- , readUtf8

43 -- , readLines

44 -- , readLines

44 -- , readFrames

45 -- , readFrames

@@ -66,6 +67,7 @@ module Streamly.Network.Socket

66where

67where

68import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)

69import Control.Concurrent (threadWaitWrite, rtsSupportsBoundThreads)

70+import Control.Monad.Catch (MonadCatch)

69import Control.Monad.IO.Class (MonadIO(..))

71import Control.Monad.IO.Class (MonadIO(..))

70import Control.Monad (when)

72import Control.Monad (when)

71import Data.Word (Word8)

73import Data.Word (Word8)

@@ -124,9 +126,10 @@ initListener tcpListenQ ServerSpec{..} =

124-- generates a stream of connected sockets and the endpoint they are connected

126-- generates a stream of connected sockets and the endpoint they are connected

125-- to. The first argument is the maximum number of pending connections in the

127-- to. The first argument is the maximum number of pending connections in the

126-- backlog.

128-- backlog.

127−{-# INLINE recvConnectionsWith #-}

129+{-# INLINE recvConnectionTuplesWith #-}

130+recvConnectionTuplesWith :: MonadAsync m

128−recvConnectionsWith :: MonadAsync m => Int -> ServerSpec -> SerialT m (Socket, SockAddr)

131+ => Int -> ServerSpec -> SerialT m (Socket, SockAddr)

129−recvConnectionsWith tcpListenQ opts = S.unfoldrM step Nothing

132+recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing

130 where

133 where

131 step Nothing = do

134 step Nothing = do

132 listener <- liftIO $ initListener tcpListenQ opts

135 listener <- liftIO $ initListener tcpListenQ opts

@@ -139,15 +142,29 @@ recvConnectionsWith tcpListenQ opts = S.unfoldrM step Nothing

139 -- XXX error handling

142 -- XXX error handling

140 return $ Just (r, Just listener)

143 return $ Just (r, Just listener)

141

144

142−{-# INLINE recvConnectionsOn #-}

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

143−recvConnectionsOn :: MonadAsync m => PortNumber -> SerialT m (Socket, SockAddr)

152+ => PortNumber -> SerialT m (Socket, SockAddr)

144−recvConnectionsOn port =

153+recvConnectionTuplesOn port =

145− recvConnectionsWith maxListenQueue ServerSpec

154+ recvConnectionTuplesWith maxListenQueue ServerSpec

146 { serverAddressFamily = AF_INET

155 { serverAddressFamily = AF_INET

147 , serverAddress = SockAddrInet port 0

156 , serverAddress = SockAddrInet port 0

148 , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]

157 , serverSockOpts = [(NoDelay,1), (ReuseAddr,1)]

149 }

158 }

150

159

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+

151-------------------------------------------------------------------------------

168-------------------------------------------------------------------------------

152-- Array IO (Input)

169-- Array IO (Input)

153-------------------------------------------------------------------------------

170-------------------------------------------------------------------------------