git.iko.soy/streamly
on 1946eb8 help sign in
streamly/test/Streamly/Test/Network/Socket.hs
160 lines 5.6 kB history raw
1{-# OPTIONS_GHC -Wno-deprecations #-}
3-- |
4-- Module : Streamly.Test.Network.Socket
5-- Copyright : (c) 2020 Composewell technologies
6-- License : BSD-3-Clause
7-- Maintainer : streamly@composewell.com
8-- Stability : experimental
9-- Portability : GHC
11module Main (main) where
13import Control.Concurrent (threadDelay, killThread, forkIO)
14import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)
15import Control.Monad.IO.Class (liftIO)
16import Data.Function ((&))
17import Data.Word (Word8)
18import Network.Socket (Socket, PortNumber)
19import Streamly.Internal.Control.Monad (discard)
20import Streamly.Internal.System.IO (defaultChunkSize)
21import Streamly.Internal.Data.Stream (Stream)
22import Test.QuickCheck (Property)
23import Test.QuickCheck.Monadic (monadicIO, assert, run)
25import qualified Streamly.Data.Fold as Fold
26import qualified Streamly.Data.Stream.Prelude as Stream
27import qualified Streamly.Internal.Network.Inet.TCP as TCP
28import qualified Streamly.Internal.Network.Socket as Socket
29import qualified Streamly.Internal.Unicode.Stream as Unicode
31import Test.Hspec
32import Test.Hspec.QuickCheck
34------------------------------------------------------------------------------
35-- Command Handlers
36------------------------------------------------------------------------------
37testData :: String
38testData = "Test data 1234567891012131415!@#$%^&*()`~ABCD"
40testDataSource :: String
41testDataSource = concat $ replicate 1000 testData
43------------------------------------------------------------------------------
44-- Parse and handle commands on a socket
45------------------------------------------------------------------------------
47handlerChunksWithBuffer :: Socket -> IO ()
48handlerChunksWithBuffer sk =
49 Stream.unfold Socket.chunkReaderWith (100, sk)
50 & Stream.fold (Socket.writeChunks sk)
51 & discard
53handlerChunks :: Socket -> IO ()
54handlerChunks sk =
55 Stream.unfold Socket.chunkReader sk
56 & Stream.fold (Socket.writeChunks sk)
57 & discard
59handlerwithbuffer :: Socket -> IO ()
60handlerwithbuffer sk =
61 Stream.unfold Socket.readerWith (100, sk)
62 & Stream.fold (Socket.writeWith 100 sk)
63 & discard
65handlerRW :: Socket -> IO ()
66handlerRW sk =
67 Stream.unfold Socket.reader sk
68 & Stream.fold (Socket.write sk)
69 & discard
71------------------------------------------------------------------------------
72-- Accept connections and handle connected sockets
73------------------------------------------------------------------------------
75-- Ideally we should choose an available port automatically. However, to test
76-- the APIs that do not allow choosing a port automatically we still need to
77-- use a hardcoded port or search an available port on the system.
78--
79-- This is unreliable and the test may fail on systems where this port is not
80-- available. We choose a higher port number so that the likelihood of it being
81-- available is more.
82--
83-- Also, we cannot run the test after running it once until the timeout.
84basePort :: PortNumber
85basePort = 64000
87server :: PortNumber -> MVar () -> (Socket -> IO ()) -> IO ()
88server port sem handler = do
89 putMVar sem ()
90 Stream.unfold TCP.acceptorOnPort port
91 & Stream.mapM (Socket.forSocketM handler)
92 & Stream.fold Fold.drain
94remoteAddr :: (Word8,Word8,Word8,Word8)
95remoteAddr = (127, 0, 0, 1)
97sender :: PortNumber -> MVar () -> Stream IO Char
98sender port sem = Stream.before action stream
100 where
102 action = liftIO (takeMVar sem >> threadDelay 1000000) -- wait for server
104 stream =
105 Stream.replicate 1000 testData -- Stream IO String
106 & Stream.concatMap Stream.fromList -- Stream IO Char
107 & Unicode.encodeLatin1 -- Stream IO Word8
108 & TCP.pipeBytes remoteAddr port -- Stream IO Word8
109 & Unicode.decodeLatin1 -- Stream IO Char
111execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (Stream IO Char)
112execute port size handler = do
113 sem <- newEmptyMVar
114 tid <- forkIO $ server port sem handler
115 let lst = sender port sem
116 & Stream.take size
117 & Stream.finally (killThread tid)
118 return lst
120validateWith :: Property
121validateWith = monadicIO $ do
122 res <- run $ do
123 ls2 <- execute basePort 45000 handlerwithbuffer
124 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
125 assert res
127validateRW :: Property
128validateRW = monadicIO $ do
129 res <- run $ do
130 ls2 <- execute (basePort + 1) defaultChunkSize handlerRW
131 let dataChunk = take defaultChunkSize testDataSource
132 Stream.eqBy (==) (Stream.fromList dataChunk) ls2
133 assert res
135validateChunks :: Property
136validateChunks = monadicIO $ do
137 res <- run $ do
138 ls2 <- execute (basePort + 2) 45000 handlerChunks
139 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
140 assert res
142validateChunksWith :: Property
143validateChunksWith = monadicIO $ do
144 res <- run $ do
145 ls2 <- execute (basePort + 3) 45000 handlerChunksWithBuffer
146 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
147 assert res
149moduleName :: String
150moduleName = "Network.Socket"
152main :: IO ()
153main = hspec $ do
154 modifyMaxSuccess (const 1) $ do
155 describe moduleName $ do
156 describe "Read/Write" $ do
157 prop "read/write" validateRW
158 prop "readWith/writeWith" validateWith
159 prop "readChunks/writeChunks" validateChunks
160 prop "readChunksWith" validateChunksWith