git.iko.soy/streamly
on ffcb268 help sign in
streamly/test/Streamly/Test/Network/Socket.hs
159 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.Internal.Network.Inet.TCP as TCP
26import qualified Streamly.Internal.Network.Socket as Socket
27import qualified Streamly.Internal.Unicode.Stream as Unicode
28import qualified Streamly.Prelude as Stream
30import Test.Hspec
31import Test.Hspec.QuickCheck
33------------------------------------------------------------------------------
34-- Command Handlers
35------------------------------------------------------------------------------
36testData :: String
37testData = "Test data 1234567891012131415!@#$%^&*()`~ABCD"
39testDataSource :: String
40testDataSource = concat $ replicate 1000 testData
42------------------------------------------------------------------------------
43-- Parse and handle commands on a socket
44------------------------------------------------------------------------------
46handlerChunksWithBuffer :: Socket -> IO ()
47handlerChunksWithBuffer sk =
48 Stream.unfold Socket.chunkReaderWith (100, sk)
49 & Stream.fold (Socket.writeChunks sk)
50 & discard
52handlerChunks :: Socket -> IO ()
53handlerChunks sk =
54 Stream.unfold Socket.chunkReader sk
55 & Stream.fold (Socket.writeChunks sk)
56 & discard
58handlerwithbuffer :: Socket -> IO ()
59handlerwithbuffer sk =
60 Stream.unfold Socket.readerWith (100, sk)
61 & Stream.fold (Socket.writeWith 100 sk)
62 & discard
64handlerRW :: Socket -> IO ()
65handlerRW sk =
66 Stream.unfold Socket.reader sk
67 & Stream.fold (Socket.write sk)
68 & discard
70------------------------------------------------------------------------------
71-- Accept connections and handle connected sockets
72------------------------------------------------------------------------------
74-- Ideally we should choose an available port automatically. However, to test
75-- the APIs that do not allow choosing a port automatically we still need to
76-- use a hardcoded port or search an available port on the system.
77--
78-- This is unreliable and the test may fail on systems where this port is not
79-- available. We choose a higher port number so that the likelihood of it being
80-- available is more.
81--
82-- Also, we cannot run the test after running it once until the timeout.
83basePort :: PortNumber
84basePort = 64000
86server :: PortNumber -> MVar () -> (Socket -> IO ()) -> IO ()
87server port sem handler = do
88 putMVar sem ()
89 Stream.fromSerial (Stream.unfold TCP.acceptorOnPort port)
90 & Stream.fromAsync . Stream.mapM (Socket.forSocketM handler)
91 & Stream.drain
93remoteAddr :: (Word8,Word8,Word8,Word8)
94remoteAddr = (127, 0, 0, 1)
96sender :: PortNumber -> MVar () -> Stream IO Char
97sender port sem = Stream.before action stream
99 where
101 action = liftIO (takeMVar sem >> threadDelay 1000000) -- wait for server
103 stream =
104 Stream.replicate 1000 testData -- Stream IO String
105 & Stream.concatMap Stream.fromList -- Stream IO Char
106 & Unicode.encodeLatin1 -- Stream IO Word8
107 & TCP.pipeBytes remoteAddr port -- Stream IO Word8
108 & Unicode.decodeLatin1 -- Stream IO Char
110execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (Stream IO Char)
111execute port size handler = do
112 sem <- newEmptyMVar
113 tid <- forkIO $ server port sem handler
114 let lst = sender port sem
115 & Stream.take size
116 & Stream.finally (killThread tid)
117 return lst
119validateWith :: Property
120validateWith = monadicIO $ do
121 res <- run $ do
122 ls2 <- execute basePort 45000 handlerwithbuffer
123 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
124 assert res
126validateRW :: Property
127validateRW = monadicIO $ do
128 res <- run $ do
129 ls2 <- execute (basePort + 1) defaultChunkSize handlerRW
130 let dataChunk = take defaultChunkSize testDataSource
131 Stream.eqBy (==) (Stream.fromList dataChunk) ls2
132 assert res
134validateChunks :: Property
135validateChunks = monadicIO $ do
136 res <- run $ do
137 ls2 <- execute (basePort + 2) 45000 handlerChunks
138 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
139 assert res
141validateChunksWith :: Property
142validateChunksWith = monadicIO $ do
143 res <- run $ do
144 ls2 <- execute (basePort + 3) 45000 handlerChunksWithBuffer
145 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
146 assert res
148moduleName :: String
149moduleName = "Network.Socket"
151main :: IO ()
152main = hspec $ do
153 modifyMaxSuccess (const 1) $ do
154 describe moduleName $ do
155 describe "Read/Write" $ do
156 prop "read/write" validateRW
157 prop "readWith/writeWith" validateWith
158 prop "readChunks/writeChunks" validateChunks
159 prop "readChunksWith" validateChunksWith