git.iko.soy/streamly
on 46314a2 help sign in
streamly/test/Streamly/Test/Network/Socket.hs
151 lines 5.6 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.Internal.System.IO (defaultChunkSize)
19import Streamly.Prelude (SerialT)
20import Test.QuickCheck (Property)
21import Test.QuickCheck.Monadic (monadicIO, assert, run)
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.fromSerial (Stream.unfold TCP.acceptOnPort port)
86 & Stream.fromAsync . Stream.mapM (Socket.forSocketM 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.processBytes 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) defaultChunkSize handlerRW
122 let dataChunk = take 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
140moduleName :: String
141moduleName = "Network.Socket"
143main :: IO ()
144main = hspec $ do
145 modifyMaxSuccess (const 1) $ do
146 describe moduleName $ do
147 describe "Read/Write" $ do
148 prop "read/write" validateRW
149 prop "readWithBufferOf/writeWithBufferOf" validateWithBufferOf
150 prop "readChunks/writeChunks" validateChunks
151 prop "readChunksWithBufferOf" validateChunksWithBufferOf