git.iko.soy/streamly
on 0717458 help sign in
streamly/test/Streamly/Test/Network/Socket.hs
155 lines 5.5 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 = do
98 _ <- liftIO $ takeMVar sem
99 liftIO $ threadDelay 1000000 -- wait for server
100 Stream.replicate 1000 testData -- Stream IO String
101 & Stream.concatMap Stream.fromList -- Stream IO Char
102 & Unicode.encodeLatin1 -- Stream IO Word8
103 & TCP.pipeBytes remoteAddr port -- Stream IO Word8
104 & Unicode.decodeLatin1 -- Stream IO Char
106execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO (Stream IO Char)
107execute port size handler = do
108 sem <- newEmptyMVar
109 tid <- forkIO $ server port sem handler
110 let lst = sender port sem
111 & Stream.take size
112 & Stream.finally (killThread tid)
113 return lst
115validateWith :: Property
116validateWith = monadicIO $ do
117 res <- run $ do
118 ls2 <- execute basePort 45000 handlerwithbuffer
119 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
120 assert res
122validateRW :: Property
123validateRW = monadicIO $ do
124 res <- run $ do
125 ls2 <- execute (basePort + 1) defaultChunkSize handlerRW
126 let dataChunk = take defaultChunkSize testDataSource
127 Stream.eqBy (==) (Stream.fromList dataChunk) ls2
128 assert res
130validateChunks :: Property
131validateChunks = monadicIO $ do
132 res <- run $ do
133 ls2 <- execute (basePort + 2) 45000 handlerChunks
134 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
135 assert res
137validateChunksWith :: Property
138validateChunksWith = monadicIO $ do
139 res <- run $ do
140 ls2 <- execute (basePort + 3) 45000 handlerChunksWithBuffer
141 Stream.eqBy (==) (Stream.fromList testDataSource) ls2
142 assert res
144moduleName :: String
145moduleName = "Network.Socket"
147main :: IO ()
148main = hspec $ do
149 modifyMaxSuccess (const 1) $ do
150 describe moduleName $ do
151 describe "Read/Write" $ do
152 prop "read/write" validateRW
153 prop "readWith/writeWith" validateWith
154 prop "readChunks/writeChunks" validateChunks
155 prop "readChunksWith" validateChunksWith