@@ -0,0 +1,128 @@
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
8+
9+module Main (main) where
10+
11+import Control.Concurrent (threadDelay, killThread, forkIO)
12+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar)
13+import Control.Monad.IO.Class (liftIO)
14+import Data.Function ((&))
15+import Data.Word (Word8)
16+import Network.Socket (Socket, PortNumber)
17+import Streamly.Internal.Control.Monad (discard)
18+import Streamly.Prelude (SerialT)
19+import Test.QuickCheck (Property)
20+import Test.QuickCheck.Monadic (monadicIO, assert, run)
21+
22+import qualified Streamly.Internal.Data.Array.Storable.Foreign.Types as A
23+import qualified Streamly.Internal.Data.Unfold as UF
24+import qualified Streamly.Internal.Network.Inet.TCP as TCP
25+import qualified Streamly.Internal.Network.Socket as SK
26+import qualified Streamly.Internal.Unicode.Stream as U
27+import qualified Streamly.Prelude as S
28+
29+import Test.Hspec
30+import Test.Hspec.QuickCheck
31+
32+------------------------------------------------------------------------------
33+-- Command Handlers
34+------------------------------------------------------------------------------
35+testData :: String
36+testData = "Test data 1234567891012131415!@#$%^&*()`~ABCD"
37+
38+testDataSource :: String
39+testDataSource = concat $ replicate 1000 testData
40+
41+------------------------------------------------------------------------------
42+-- Parse and handle commands on a socket
43+------------------------------------------------------------------------------
44+handlerChunksWithBuffer :: Socket -> IO ()
45+handlerChunksWithBuffer sk =
46+ S.unfold SK.readChunksWithBufferOf (100, sk)
47+ & S.fold (SK.writeChunks sk)
48+ & discard
49+
50+handlerChunks :: Socket -> IO ()
51+handlerChunks sk =
52+ S.unfold SK.readChunks sk
53+ & S.fold (SK.writeChunks sk)
54+ & discard
55+
56+handlerwithbuffer :: Socket -> IO ()
57+handlerwithbuffer sk =
58+ S.unfold SK.readWithBufferOf (100, sk)
59+ & S.fold (SK.writeWithBufferOf 100 sk)
60+ & discard
61+
62+handlerRW :: Socket -> IO ()
63+handlerRW sk =
64+ S.unfold SK.read sk
65+ & S.fold (SK.write sk)
66+ & discard
67+------------------------------------------------------------------------------
68+-- Accept connecttions and handle connected sockets
69+------------------------------------------------------------------------------
70+
71+server :: PortNumber -> MVar () -> (Socket -> IO ()) -> IO ()
72+server port sem handler = do
73+ putMVar sem ()
74+ (S.serially $ S.unfold TCP.acceptOnPort port)
75+ & (S.asyncly . S.mapM (SK.handleWithM handler))
76+ & S.drain
77+
78+remoteAddr :: (Word8,Word8,Word8,Word8)
79+remoteAddr = (127, 0, 0, 1)
80+
81+sender :: PortNumber -> MVar () -> SerialT IO Char
82+sender port sem = do
83+ _ <- liftIO $ takeMVar sem
84+ liftIO $ threadDelay 1000000 -- wait for server
85+ S.replicate 1 testDataSource -- SerialT IO String
86+ & U.unwords UF.fromList -- SerialT IO Char
87+ & U.encodeLatin1 -- SerialT IO Word8
88+ & TCP.transformBytesWith remoteAddr port -- SerialT IO Word8
89+ & U.decodeLatin1 -- SerialT IO Char
90+
91+execute :: PortNumber -> Int -> (Socket -> IO ()) -> IO [Char]
92+execute port size handler = do
93+ sem <- newEmptyMVar
94+ tid <- forkIO $ server port sem handler
95+ lst <- sender port sem
96+ & S.take size
97+ & S.toList
98+ killThread tid
99+ return $ lst
100+
101+validateWithBufferOf :: Property
102+validateWithBufferOf = monadicIO $ do
103+ ls2 <- run $ execute 8000 45000 handlerwithbuffer
104+ assert (testDataSource == ls2)
105+
106+validateRW :: Property
107+validateRW = monadicIO $ do
108+ ls2 <- run $ execute 8001 A.defaultChunkSize handlerRW
109+ assert (take A.defaultChunkSize testDataSource == ls2)
110+
111+validateChunks :: Property
112+validateChunks = monadicIO $ do
113+ ls2 <- run $ execute 8002 45000 handlerChunks
114+ assert (testDataSource == ls2)
115+
116+validateChunksWithBufferOf :: Property
117+validateChunksWithBufferOf = monadicIO $ do
118+ ls2 <- run $ execute 8003 45000 handlerChunksWithBuffer
119+ assert (testDataSource == ls2)
120+
121+main :: IO ()
122+main = hspec $ do
123+ modifyMaxSuccess (const 1) $ do
124+ describe "Socket Tests: " $ do
125+ prop "read/write" validateRW
126+ prop "readWithBufferOf/writeWithBufferOf" validateWithBufferOf
127+ prop "readChunksWithBufferOf" validateChunksWithBufferOf
128+ prop "readChunks/writeChunks" validateChunks