@@ -32,13 +32,8 @@
3232
3333module Streamly.Network.Socket
3434 (
35− -- ** Listen for Connections
36− ServerSpec(..)
37− , recvConnectionsWith
38− , recvConnectionsOn
39−
40− -- ** Read a stream from a connection
41− , fromSocket
35+ -- * Read from connection
36+ fromSocket
4237 , read
4338 -- , readUtf8
4439 -- , readLines
@@ -53,7 +48,7 @@ module Streamly.Network.Socket
5348 -- , readArraysOf
5449 , readArrays
5550
56− -- ** Write a stream to a connection
51+ -- * Write to connection
5752 , write
5853 -- , writeUtf8
5954 -- , writeUtf8ByLines
@@ -63,6 +58,8 @@ module Streamly.Network.Socket
6358 -- -- * Array Write
6459 , writeArray
6560 , writeArrays
61+
62+ -- reading/writing datagrams
6663 )
6764where
6865
@@ -76,10 +73,7 @@ import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
7673import Foreign.Ptr (minusPtr, plusPtr, Ptr, castPtr)
7774import Foreign.Storable (Storable(..))
7875import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
79−import Network.Socket
80− (Socket, PortNumber, SocketOption(..), Family(..),
81− SockAddr(..), withSocketsDo, SocketType(..), socket, accept, bind,
82− defaultProtocol, setSocketOption, maxListenQueue, sendBuf, recvBuf)
76+import Network.Socket (Socket, sendBuf, recvBuf)
8377#if MIN_VERSION_network(3,1,0)
8478import Network.Socket (withFdSocket)
8579#else
@@ -89,7 +83,6 @@ import Prelude hiding (read)
8983
9084import qualified Network.Socket as Net
9185
92−import Streamly (MonadAsync)
9386import Streamly.Mem.Array.Types (Array(..))
9487import Streamly.Streams.Serial (SerialT)
9588import Streamly.Streams.StreamK.Type (IsStream, mkStream)
@@ -100,71 +93,6 @@ import qualified Streamly.Mem.Array as A
10093import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)
10194import qualified Streamly.Prelude as S
10295
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−
16896-------------------------------------------------------------------------------
16997-- Array IO (Input)
17098-------------------------------------------------------------------------------
@@ -307,6 +235,10 @@ readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
307235read :: (IsStream t, MonadIO m) => Socket -> t m Word8
308236read = A.flattenArrays . readArrays
309237
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+
310242-------------------------------------------------------------------------------
311243-- Writing
312244-------------------------------------------------------------------------------