git.iko.soy/streamly
on 00a2f7b help sign in
streamly/test/Streamly/Test/Network/Socket.hs
147 lines 5.5 kB history raw
1-- |
2-- Module : Streamly.Test.Network.Socket
3-- Copyright : (c) 2020 Composewell technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
9module Main (main) where
11import Control.Concurrent (threadDelay, killThread, forkIO)
12import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)
13import Control.Monad.IO.Class (liftIO)
14import Data.Function ((&))
15import Data.Word (Word8)
16import Network.Socket (Socket, PortNumber)
17import Streamly.Internal.Control.Monad (discard)
18import Streamly.Prelude (SerialT)
19import Test.QuickCheck (Property)
20import Test.QuickCheck.Monadic (monadicIO, assert, run)
22import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as Array
23import qualified Streamly.Internal.Network.Inet.TCP as TCP
24import qualified Streamly.Internal.Network.Socket as Socket
25import qualified Streamly.Internal.Unicode.Stream as Unicode
26import qualified Streamly.Prelude as Stream
28import Test.Hspec
29import Test.Hspec.QuickCheck
31------------------------------------------------------------------------------
32-- Command Handlers
33------------------------------------------------------------------------------
34testData :: String
35testData = "Test data 1234567891012131415!@#$%^&*()`~ABCD"
37testDataSource :: String
38testDataSource = concat $ replicate 1000 testData
40------------------------------------------------------------------------------
41-- Parse and handle commands on a socket
42------------------------------------------------------------------------------
43handlerChunksWithBuffer :: Socket -> IO ()
44handlerChunksWithBuffer sk =
45 Stream.unfold Socket.readChunksWithBufferOf (100, sk)
46 & Stream.fold (Socket.writeChunks sk)
47 & discard
49handlerChunks :: Socket -> IO ()
50handlerChunks sk =
51 Stream.unfold Socket.readChunks sk
52 & Stream.fold (Socket.writeChunks sk)
53 & discard
55handlerwithbuffer :: Socket -> IO ()
56handlerwithbuffer sk =
57 Stream.unfold Socket.readWithBufferOf (100, sk)
58 & Stream.fold (Socket.writeWithBufferOf 100 sk)
59 & discard
61handlerRW :: Socket -> IO ()
62handlerRW sk =
63 Stream.unfold Socket.read sk
64 & Stream.fold (Socket.write sk)
65 & discard
66------------------------------------------------------------------------------
67-- Accept connections and handle connected sockets
68------------------------------------------------------------------------------
70-- Ideally we should choose an available port automatically. However, to test
71-- the APIs that do not allow choosing a port automatically we still need to
72-- use a hardcoded port or search an available port on the system.
73--
74-- This is unreliable and the test may fail on systems where this port is not
75-- available. We choose a higher port number so that the likelihood of it being
76-- available is more.
77--
78-- Also, we cannot run the test after running it once until the timeout.
79basePort :: PortNumber
80basePort = 64000
82server :: PortNumber -> MVar () -> (Socket -> IO ()) -> IO ()
83server port sem handler = do
84 putMVar sem ()
85 Stream.serially (Stream.unfold TCP.acceptOnPort port)
86 & Stream.asyncly . Stream.mapM (Socket.handleWithM handler)
87 & Stream.drain
89remoteAddr :: (Word8,Word8,Word8,Word8)
90remoteAddr = (127, 0, 0, 1)
92sender :: PortNumber -> MVar () -> SerialT IO Char
93sender port sem = do
94 _ <- liftIO $ takeMVar sem
95 liftIO $ threadDelay 1000000 -- wait for server
96 Stream.replicate 1000 testData -- SerialT IO String
97 & Stream.concatMap Stream.fromList -- SerialT IO Char
98 & Unicode.encodeLatin1 -- SerialT IO Word8
99 & TCP.transformBytesWith remoteAddr port -- SerialT IO Word8
100 & Unicode.decodeLatin1 -- SerialT IO Char
102execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (SerialT IO Char)
103execute port size handler = do
104 sem <- newEmptyMVar
105 tid <- forkIO $ server port sem handler
106 let lst = sender port sem
107 & Stream.take size
108 & Stream.finally (killThread tid)
109 return lst
111validateWithBufferOf :: Property
112validateWithBufferOf = monadicIO $ do
113 res <- run $ do
114 ls2 <- execute basePort 45000 handlerwithbuffer
115 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
116 assert res
118validateRW :: Property
119validateRW = monadicIO $ do
120 res <- run $ do
121 ls2 <- execute (basePort + 1) Array.defaultChunkSize handlerRW
122 let dataChunk = take Array.defaultChunkSize testDataSource
123 Stream.eqBy (==) (Stream.fromList dataChunk) ls2
124 assert res
126validateChunks :: Property
127validateChunks = monadicIO $ do
128 res <- run $ do
129 ls2 <- execute (basePort + 2) 45000 handlerChunks
130 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
131 assert res
133validateChunksWithBufferOf :: Property
134validateChunksWithBufferOf = monadicIO $ do
135 res <- run $ do
136 ls2 <- execute (basePort + 3) 45000 handlerChunksWithBuffer
137 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
138 assert res
140main :: IO ()
141main = hspec $ do
142 modifyMaxSuccess (const 1) $ do
143 describe "Read/Write" $ do
144 prop "read/write" validateRW
145 prop "readWithBufferOf/writeWithBufferOf" validateWithBufferOf
146 prop "readChunks/writeChunks" validateChunks
147 prop "readChunksWithBufferOf" validateChunksWithBufferOf