typed-protocols-examples: pipelined tests

* using TBQueue based channels in ST and IO
* using unix named pipes and 'handleAsChannel'
* using sockets and 'socketAsChannel'

typed-protocols-examples/src/Network/TypedProtocol/Channel.hs

@@ -203,7 +203,8 @@ createPipelineTestChannels sz = do
 --
 -- The Handles should be open in the appropriate read or write mode, and in
 -- binary mode. Writes are flushed after each write, so it is safe to use
--- a buffering mode.
+-- a buffering mode.  On unix named pipes can be used, see
+-- 'Network.TypedProtocol.ReqResp.Test.prop_namedPipePipelined_IO'
 --
 -- For bidirectional handles it is safe to pass the same handle for both.
 --

typed-protocols-examples/test/Network/TypedProtocol/PingPong/Tests.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP                      #-}
 {-# LANGUAGE DataKinds                #-}
 {-# LANGUAGE FlexibleInstances        #-}
 {-# LANGUAGE GADTs                    #-}
@@ -34,10 +35,17 @@ import           Control.Monad.Class.MonadThrow
 import           Control.Monad.IOSim (runSimOrThrow)
 import           Control.Monad.ST (runST)
 import           Control.Tracer (nullTracer)
+
 import           Data.Functor.Identity (Identity (..))
+import           Data.List (inits, tails)
 
 import qualified Data.ByteString.Lazy as LBS
-import           Data.List (inits, tails)
+#if !defined(mingw32_HOST_OS)
+import qualified Network.Socket as Socket
+import           System.Directory (removeFile)
+import           System.IO
+import qualified System.Posix.Files as Posix
+#endif
 
 import           Test.QuickCheck
 import           Test.Tasty (TestTree, testGroup)
@@ -61,6 +69,10 @@ tests = testGroup "Network.TypedProtocol.PingPong"
   , testProperty "connect_pipelined 5" prop_connect_pipelined5
   , testProperty "channel ST"          prop_channel_ST
   , testProperty "channel IO"          prop_channel_IO
+#if !defined(mingw32_HOST_OS)
+  , testProperty "namedPipePipelined"  prop_namedPipePipelined_IO
+  , testProperty "socketPipelined"     prop_socketPipelined_IO
+#endif
   , testGroup "Codec"
     [ testProperty "codec"             prop_codec_PingPong
     , testProperty "codec 2-splits"    prop_codec_splits2_PingPong
@@ -313,6 +325,64 @@ prop_channel_ST n =
     runSimOrThrow (prop_channel n)
 
 
+#if !defined(mingw32_HOST_OS)
+prop_namedPipePipelined_IO :: NonNegative Int
+                           -> Property
+prop_namedPipePipelined_IO (NonNegative n) = ioProperty $ do
+    let client = pingPongClientPeer (pingPongClientCount n)
+        server = pingPongServerPeer  pingPongServerCount
+
+    let cliPath = "client.sock"
+        srvPath = "server.sock"
+        mode = Posix.ownerModes
+
+    Posix.createNamedPipe cliPath mode
+    Posix.createNamedPipe srvPath mode
+
+    bracket   (openFile cliPath ReadWriteMode)
+              (\_ -> removeFile cliPath)
+            $ \cliHandle ->
+      bracket (openFile srvPath ReadWriteMode)
+              (\_ -> removeFile srvPath)
+           $ \srvHandle -> do
+              ((), n') <- runConnectedPeers (return ( handlesAsChannel cliHandle srvHandle
+                                                    , handlesAsChannel srvHandle cliHandle
+                                                    ))
+                                            nullTracer
+                                            CBOR.codecPingPong client server
+              return (n' == n)
+#endif
+
+
+#if !defined(mingw32_HOST_OS)
+prop_socketPipelined_IO :: NonNegative Int
+                        -> Property
+prop_socketPipelined_IO (NonNegative n) = ioProperty $ do
+    ai : _ <- Socket.getAddrInfo Nothing (Just "127.0.0.1") Nothing
+    bracket
+      ((,) <$> Socket.openSocket ai
+           <*> Socket.openSocket ai)
+      ( \ (sock, sock') -> Socket.close sock
+                        >> Socket.close sock')
+      $ \ (sock, sock') -> do
+          Socket.bind sock (Socket.addrAddress ai)
+          addr <- Socket.getSocketName sock
+          Socket.listen sock 1
+          Socket.connect sock' addr
+          bracket (fst <$> Socket.accept sock) Socket.close
+                $ \sock'' -> do
+            let client = pingPongClientPeer (pingPongClientCount n)
+                server = pingPongServerPeer  pingPongServerCount
+
+            ((), n') <- runConnectedPeers (return ( socketAsChannel sock'
+                                                  , socketAsChannel sock''
+                                                  ))
+                                          nullTracer
+                                          CBOR.codecPingPong client server
+            return (n' == n)
+#endif
+
+
 --
 -- Codec properties
 --

typed-protocols-examples/test/Network/TypedProtocol/ReqResp/Tests.hs

@@ -1,4 +1,5 @@
 {-# LANGUAGE BangPatterns             #-}
+{-# LANGUAGE CPP                      #-}
 {-# LANGUAGE DataKinds                #-}
 {-# LANGUAGE FlexibleInstances        #-}
 {-# LANGUAGE GADTs                    #-}
@@ -29,16 +30,25 @@ import           Network.TypedProtocol.ReqResp.Examples
 import           Network.TypedProtocol.ReqResp.Server
 import           Network.TypedProtocol.ReqResp.Type
 
+import           Control.Exception (throw)
 import           Control.Monad.Class.MonadAsync
+import           Control.Monad.Class.MonadST
 import           Control.Monad.Class.MonadSTM
 import           Control.Monad.Class.MonadThrow
-import           Control.Monad.IOSim (runSimOrThrow)
+import           Control.Monad.Class.MonadTimer.SI
+import           Control.Monad.IOSim
 import           Control.Monad.ST (runST)
 import           Control.Tracer (nullTracer)
 
 import           Data.Functor.Identity (Identity (..))
-import           Data.List (mapAccumL)
+import           Data.List (intercalate, mapAccumL)
 import           Data.Tuple (swap)
+#if !defined(mingw32_HOST_OS)
+import qualified Network.Socket as Socket
+import           System.Directory (removeFile)
+import           System.IO
+import qualified System.Posix.Files as Posix
+#endif
 
 import           Network.TypedProtocol.PingPong.Tests (splits2, splits2BS,
                      splits3, splits3BS)
@@ -61,6 +71,12 @@ tests = testGroup "Network.TypedProtocol.ReqResp"
   , testProperty "connectPipelined"    prop_connectPipelined
   , testProperty "channel ST"          prop_channel_ST
   , testProperty "channel IO"          prop_channel_IO
+  , testProperty "channelPipelined ST" prop_channelPipelined_ST
+  , testProperty "channelPipelined IO" prop_channelPipelined_IO
+#if !defined(mingw32_HOST_OS)
+  , testProperty "namedPipePipelined"  prop_namedPipePipelined_IO
+  , testProperty "socketPipelined"     prop_socketPipelined_IO
+#endif
   , testGroup "Codec"
     [ testProperty "codec"             prop_codec_ReqResp
     , testProperty "codec 2-splits"    prop_codec_splits2_ReqResp
@@ -186,6 +202,7 @@ prop_channel f xs = do
     server = reqRespServerPeer (reqRespServerMapAccumL
                                  (\a -> pure . f a) 0)
 
+
 prop_channel_IO :: (Int -> Int -> (Int, Int)) -> [Int] -> Property
 prop_channel_IO f xs =
     ioProperty (prop_channel f xs)
@@ -195,6 +212,97 @@ prop_channel_ST f xs =
     runSimOrThrow (prop_channel f xs)
 
 
+prop_channelPipelined :: ( MonadLabelledSTM m, MonadAsync m, MonadCatch m
+                         , MonadDelay m, MonadST m)
+                      => (Int -> Int -> (Int, Int)) -> [Int]
+                      -> m Bool
+prop_channelPipelined f xs = do
+    (c, s) <- runConnectedPeersPipelined
+                (createPipelineTestChannels 100)
+                nullTracer
+                CBOR.codecReqResp
+                client server
+    return ((s, c) == mapAccumL f 0 xs)
+  where
+    client = reqRespClientPeerPipelined (reqRespClientMapPipelined xs)
+    server = reqRespServerPeer          (reqRespServerMapAccumL
+                                            (\a -> pure . f a) 0)
+
+prop_channelPipelined_IO :: (Int -> Int -> (Int, Int)) -> [Int] -> Property
+prop_channelPipelined_IO f xs =
+    ioProperty (prop_channelPipelined f xs)
+
+prop_channelPipelined_ST :: (Int -> Int -> (Int, Int)) -> [Int] -> Property
+prop_channelPipelined_ST f xs =
+    let tr = runSimTrace (prop_channelPipelined f xs) in
+    counterexample (intercalate "\n" $ map show $ traceEvents tr)
+                 $ case traceResult True tr of
+                     Left  err -> throw err
+                     Right res -> res
+
+
+#if !defined(mingw32_HOST_OS)
+prop_namedPipePipelined_IO :: (Int -> Int -> (Int, Int)) -> [Int]
+                           -> Property
+prop_namedPipePipelined_IO f xs = ioProperty $ do
+    let client = reqRespClientPeerPipelined (reqRespClientMapPipelined xs)
+        server = reqRespServerPeer          (reqRespServerMapAccumL
+                                                (\a -> pure . f a) 0)
+    let cliPath = "client.sock"
+        srvPath = "server.sock"
+        mode = Posix.ownerModes
+
+    Posix.createNamedPipe cliPath mode
+    Posix.createNamedPipe srvPath mode
+
+    bracket   (openFile cliPath ReadWriteMode)
+              (\_ -> removeFile cliPath)
+           $ \cliHandle ->
+      bracket (openFile srvPath ReadWriteMode)
+              (\_ -> removeFile srvPath)
+           $ \srvHandle -> do
+              (c, s) <- runConnectedPeersPipelined
+                          (return ( handlesAsChannel cliHandle srvHandle
+                                  , handlesAsChannel srvHandle cliHandle
+                                  ))
+                          nullTracer
+                          CBOR.codecReqResp
+                          client server
+              return ((s, c) == mapAccumL f 0 xs)
+#endif
+
+
+#if !defined(mingw32_HOST_OS)
+prop_socketPipelined_IO :: (Int -> Int -> (Int, Int)) -> [Int]
+                        -> Property
+prop_socketPipelined_IO f xs = ioProperty $ do
+    ai : _ <- Socket.getAddrInfo Nothing (Just "127.0.0.1") Nothing
+    bracket
+      ((,) <$> Socket.openSocket ai
+           <*> Socket.openSocket ai)
+      ( \ (sock, sock') -> Socket.close sock
+                        >> Socket.close sock')
+      $ \ (sock, sock') -> do
+          Socket.bind sock (Socket.addrAddress ai)
+          addr <- Socket.getSocketName sock
+          Socket.listen sock 1
+          Socket.connect sock' addr
+          bracket (fst <$> Socket.accept sock) Socket.close
+                $ \sock'' -> do
+            let client = reqRespClientPeerPipelined (reqRespClientMapPipelined xs)
+                server = reqRespServerPeer          (reqRespServerMapAccumL
+                                                        (\a -> pure . f a) 0)
+
+            (c, s) <- runConnectedPeersPipelined
+                        (return ( socketAsChannel sock'
+                                , socketAsChannel sock''
+                                ))
+                        nullTracer
+                        CBOR.codecReqResp
+                        client server
+            return ((s, c) == mapAccumL f 0 xs)
+#endif
+
 --
 -- Codec properties
 --

typed-protocols-examples/typed-protocols-examples.cabal

@@ -89,6 +89,12 @@ test-suite test
                    , QuickCheck
                    , tasty
                    , tasty-quickcheck
+
+  if !os(windows)
+      build-depends: directory
+                   , network
+                   , unix
+
   default-language:  Haskell2010
   ghc-options:       -rtsopts
                      -Wall