a2e7676 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Split the server combinators in a separate module

a2e7676Harendra Kumar7 years ago

examples/FileSinkServer.hs+2−1

compare with master

unifiedsplit

@@ -13,6 +13,7 @@ import qualified Streamly.FileSystem.File as File

13import qualified Streamly.Fold as FL

14import qualified Streamly.Mem.Array as A

15import qualified Streamly.Network.Socket as NS

16+import qualified Streamly.Network.Server as NS

16import qualified Streamly.Prelude as S

17import qualified Streamly.Prelude as S

18main :: IO ()

19main :: IO ()

@@ -22,7 +23,7 @@ main = do

22 $ encodeChar8Unchecked

23 $ encodeChar8Unchecked

23 $ S.concatMap A.read

24 $ S.concatMap A.read

24 $ S.concatMapBy parallel recv

25 $ S.concatMapBy parallel recv

25− $ NS.recvConnectionsOn 8090

26+ $ NS.connectionsOnAllAddrs 8090

27 where

28 where

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

@@ -32,13 +32,8 @@

33module Streamly.Network.Socket

34 (

35− -- ** Listen for Connections

36− ServerSpec(..)

37− , recvConnectionsWith

38− , recvConnectionsOn

39−

40− -- ** Read a stream from a connection

35+ -- * Read from connection

41− , fromSocket

36+ fromSocket

42 , read

37 , read

43 -- , readUtf8

38 -- , readUtf8

44 -- , readLines

39 -- , readLines

@@ -53,7 +48,7 @@ module Streamly.Network.Socket

53 -- , readArraysOf

48 -- , readArraysOf

54 , readArrays

49 , readArrays

56− -- ** Write a stream to a connection

51+ -- * Write to connection

57 , write

52 , write

58 -- , writeUtf8

53 -- , writeUtf8

59 -- , writeUtf8ByLines

54 -- , writeUtf8ByLines

@@ -63,6 +58,8 @@ module Streamly.Network.Socket

63 -- -- * Array Write

58 -- -- * Array Write

64 , writeArray

59 , writeArray

65 , writeArrays

60 , writeArrays

61+

62+ -- reading/writing datagrams

66 )

63 )

67where

64where

@@ -76,10 +73,7 @@ import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

76import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)

73import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)

77import Foreign.Storable (Storable(..))

74import Foreign.Storable (Storable(..))

78import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

75import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

79−import Network.Socket

76+import Network.Socket (Socket, sendBuf, recvBuf)

80− (Socket, PortNumber, SocketOption(..), Family(..),

81− SockAddr(..), withSocketsDo, SocketType(..), socket, accept, bind,

82− defaultProtocol, setSocketOption, maxListenQueue, sendBuf, recvBuf)

83#if MIN_VERSION_network(3,1,0)

77#if MIN_VERSION_network(3,1,0)

84import Network.Socket (withFdSocket)

78import Network.Socket (withFdSocket)

85#else

79#else

@@ -89,7 +83,6 @@ import Prelude hiding (read)

90import qualified Network.Socket as Net

84import qualified Network.Socket as Net

92−import Streamly (MonadAsync)

93import Streamly.Mem.Array.Types (Array(..))

86import Streamly.Mem.Array.Types (Array(..))

94import Streamly.Streams.Serial (SerialT)

87import Streamly.Streams.Serial (SerialT)

95import Streamly.Streams.StreamK.Type (IsStream, mkStream)

88import Streamly.Streams.StreamK.Type (IsStream, mkStream)

@@ -100,71 +93,6 @@ import qualified Streamly.Mem.Array as A

100import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)

93import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)

101import qualified Streamly.Prelude as S

94import qualified Streamly.Prelude as S

102

103−-------------------------------------------------------------------------------

104−-- Listen

105−-------------------------------------------------------------------------------

106−

107−-- | Specify the configuration of a server.

108−data ServerSpec = ServerSpec

109− {

110− serverAddressFamily :: !Family

111− , serverAddress :: !SockAddr

112− , serverSockOpts :: ![(SocketOption, Int)]

113− }

114−

115−initListener :: Int -> ServerSpec -> IO Socket

116−initListener tcpListenQ ServerSpec{..} =

117− withSocketsDo $ do

118− sock <- socket serverAddressFamily Stream defaultProtocol

119− mapM_ (\(opt, val) -> setSocketOption sock opt val) serverSockOpts

120− bind sock serverAddress

121− Net.listen sock tcpListenQ

122− return sock

123−

124−-- | Start a TCP stream server. The server listens for connections on the

125−-- supplied address (address family, local interface IP address and port) and

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

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

128−-- backlog.

129−{-# INLINE recvConnectionTuplesWith #-}

130−recvConnectionTuplesWith :: MonadAsync m

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

132−recvConnectionTuplesWith tcpListenQ opts = S.unfoldrM step Nothing

133− where

134− step Nothing = do

135− listener <- liftIO $ initListener tcpListenQ opts

136− r <- liftIO $ accept listener

137− -- XXX error handling

138− return $ Just (r, Just listener)

139−

140− step (Just listener) = do

141− r <- liftIO $ accept listener

142− -- XXX error handling

143− return $ Just (r, Just listener)

144−

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

155− { serverAddressFamily = AF_INET

156− , serverAddress = SockAddrInet port 0

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

158− }

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−

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

96-------------------------------------------------------------------------------

169-- Array IO (Input)

97-- Array IO (Input)

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

98-------------------------------------------------------------------------------

@@ -307,6 +235,10 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h

307read :: (IsStream t, MonadIO m) => Socket -> t m Word8

235read :: (IsStream t, MonadIO m) => Socket -> t m Word8

308read = A.flattenArrays . readArrays

236read = A.flattenArrays . readArrays

309

237

238+-- | Read a stream of Word8 from a socket, closing the socket when done.

239+fromSocket :: (MonadCatch m, MonadIO m) => Socket -> SerialT m Word8

240+fromSocket sk = S.finally (liftIO (Net.close sk)) (read sk)

241+

310-------------------------------------------------------------------------------

242-------------------------------------------------------------------------------

311-- Writing

243-- Writing

312-------------------------------------------------------------------------------

244-------------------------------------------------------------------------------

streamly.cabal+1

@@ -218,6 +218,7 @@ library

218 , Streamly.FileSystem.File

219 , Streamly.FileSystem.Handle

220 , Streamly.Network.Socket

221+ , Streamly.Network.Server

221

222

222 -- Time

223 -- Time

223 , Streamly.Time.Units

224 , Streamly.Time.Units