packages feed

sockets (empty) → 0.1.0.0

raw patch · 13 files changed

+1300/−0 lines, 13 filesdep +asyncdep +basedep +bytestringsetup-changed

Dependencies added: async, base, bytestring, fast-logger, ip, posix-api, primitive, sockets, tasty, tasty-hunit

Files

+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for stream-sockets++## 0.1.0.0 -- YYYY-mm-dd++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2019, Andrew Martin++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Andrew Martin nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Macro.hs view
@@ -0,0 +1,144 @@+{-# language BangPatterns #-}+{-# language ScopedTypeVariables #-}++import Control.Concurrent.Async (concurrently)+import Control.Exception (Exception)+import Control.Exception (throwIO)+import Control.Monad.ST (runST)+import Data.Primitive (ByteArray)+import Data.Word (Word16,Word8)+import GHC.Exts (RealWorld)+import Test.Tasty+import Test.Tasty.HUnit++import qualified Socket.Datagram.IPv4.Undestined as DIU+import qualified Socket.Stream.IPv4 as SI+import qualified GHC.Exts as E+import qualified Data.Primitive as PM+import qualified Data.Primitive.MVar as PM+import qualified Net.IPv4 as IPv4++main :: IO ()+main = do+  [duration] <- getArgs+  newIORef True+  complete <- newPrimArray 1+  replicateM_ (take+  ++participants :: Int+participants = 128++totalReceives :: Int+totalReceives = 1000000++-- The PrimArray must be of length @participants@.+worker :: +     Int -- ^ Worker identifier+  -> MutablePrimArray RealWorld Int -- ^ Counter of opened sockets, singleton array+  -> MutablePrimArray RealWorld Int -- ^ Counter of total sends, singleton array+  -> MutablePrimArray RealWorld Word16 -- ^ Ports used by local team+  -> MVar (PrimArray Word16) -- ^ MVar for ports used by local team+  -> MVar (PrimArray Word16) -- ^ MVar for ports used by remote team+  -> IO ()+worker !ident !counter !locals !mlocals !mremotes = do+  unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock port -> do+    PM.writePrimArray locals ident port+    increment counter >>= \case+      True -> PM.unsafeFreezePrimArray locals >>= putMVar mlocals+      False -> pure ()+    remotes <- readMVar mremotes+    act ++act ::+     DIU.Socket -- Socket+  -> MutableByteArray RealWorld -- Buffer for receives+  -> PrimArray Word16 -- Ports used by remote team+  -> IO ()+act !sock !buf !remotes = case act of+  DIU.send sock _ _ _ _+  DIU.send sock _ _ _ _++-- Returns true if the value of the counter reached the total+-- number of participants.+incrementWorkerCounter :: MutablePrimArray RealWorld Int -> IO Bool+incrementWorkerCounter (MutablePrimArray arr) = IO $ \s0 -> case fetchAddIntArray arr 0# 1# s0 of+  (# s1, i #) -> (# s1, I# i == participants - 1 #)++incrementReceiveCounter :: MutablePrimArray RealWorld Int -> IO Bool+incrementReceiveCounter (MutablePrimArray arr) = IO $ \s0 -> case fetchAddIntArray arr 0# 1# s0 of+  (# s1, i #) -> (# s1, I# i == totalReceives - 1 #)++tests :: TestTree+tests = testGroup "socket"+  [ testGroup "datagram"+    [ testGroup "ipv4"+      [ testGroup "undestined"+        [ testCase "A" testDatagramUndestinedA+        ]+      ]+    ]+  , testGroup "stream"+    [ testGroup "ipv4"+      [ testCase "A" testStreamA+      ]+    ]+  ]++unhandled :: Exception e => IO (Either e a) -> IO a+unhandled action = action >>= either throwIO pure++testDatagramUndestinedA :: Assertion+testDatagramUndestinedA = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  (port,received) <- concurrently (sender m) (receiver m)+  received @=? (DIU.Endpoint IPv4.loopback port, message)+  where+  message = E.fromList [0,1,2,3] :: ByteArray+  sz = PM.sizeofByteArray message+  sender :: PM.MVar RealWorld Word16 -> IO Word16+  sender m = unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock srcPort -> do+    dstPort <- PM.takeMVar m+    unhandled $ DIU.send sock (DIU.Endpoint IPv4.loopback dstPort) message 0 sz+    pure srcPort+  receiver :: PM.MVar RealWorld Word16 -> IO (DIU.Endpoint,ByteArray)+  receiver m = unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock port -> do+    PM.putMVar m port+    unhandled $ DIU.receive sock sz++-- This test involves a made up protocol that goes like this:+-- The sender always starts by sending the length of the rest+-- of the payload as a native-endian encoded machine-sized int.+-- (This could only ever work for a machine that is communicating+-- with itself). Then, it sends a bytearray of that specified+-- length. Then, both ends are expected to shutdown their sides+-- of the connection.+testStreamA :: Assertion+testStreamA = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  ((),received) <- concurrently (sender m) (receiver m)+  received @=? message+  where+  message = E.fromList (enumFromTo 0 (100 :: Word8)) :: ByteArray+  sz = PM.sizeofByteArray message+  szb = runST $ do+    marr <- PM.newByteArray (PM.sizeOf (undefined :: Int))+    PM.writeByteArray marr 0 sz+    PM.unsafeFreezeByteArray marr+  sender :: PM.MVar RealWorld Word16 -> IO ()+  sender m = do+    dstPort <- PM.takeMVar m+    unhandled $ SI.withConnection (DIU.Endpoint IPv4.loopback dstPort) $ \conn -> do+      unhandled $ SI.sendByteArray conn szb+      unhandled $ SI.sendByteArray conn message+  receiver :: PM.MVar RealWorld Word16 -> IO ByteArray+  receiver m = unhandled $ SI.withListener (SI.Endpoint IPv4.loopback 0) $ \listener port -> do+    PM.putMVar m port+    unhandled $ SI.withAccepted listener $ \conn _ -> do+      serializedSize <- unhandled $ SI.receiveByteArray conn (PM.sizeOf (undefined :: Int))+      let theSize = PM.indexByteArray serializedSize 0 :: Int+      result <- unhandled $ SI.receiveByteArray conn theSize+      pure result+++
+ example/Main.hs view
@@ -0,0 +1,76 @@+{-# language BangPatterns #-}+{-# language LambdaCase #-}+{-# language OverloadedStrings #-}++import Prelude hiding (log)++import Control.Exception (Exception,throwIO)+import System.Environment (getArgs)+import Data.Primitive (ByteArray(..))+import Control.Monad (replicateM_)++import qualified Socket.Datagram.IPv4.Undestined as DIU+import qualified Socket.Stream.IPv4 as SI+import qualified GHC.Exts as E+import qualified Data.Primitive as PM+import qualified Data.Primitive.MVar as PM+import qualified Net.IPv4 as IPv4+import qualified Data.ByteString as B+import qualified Data.ByteString.Short.Internal as SB+import qualified Data.ByteString.Char8 as BC+import qualified System.Log.FastLogger as FL++main :: IO ()+main = udpStdoutServer++gettysburgClient :: IO ()+gettysburgClient = do+  [port] <- getArgs+  unhandled $ SI.withConnection (SI.Endpoint IPv4.loopback (read port)) $ \conn -> do+    unhandled $ SI.sendByteArray conn gettysburgAddress++-- This waits until a single connection is established. It then dumps out+-- anything it receives to stdout. When the remote application shuts down+-- its end, this shuts down its end in return and then terminates.+tcpStdoutServer :: IO ()+tcpStdoutServer = do+  FL.withFastLogger (FL.LogStdout 2048) $ \log -> do+    unhandled $ SI.withListener (SI.Endpoint IPv4.loopback 0) $ \listener port -> do+      log (FL.toLogStr ("Listening on 127.0.0.1:" <> BC.pack (show port) <> "\n============================\n"))+      unhandled $ SI.withAccepted listener $ \conn _ -> do+        let go = SI.receiveBoundedByteArray conn 512 >>= \case+              Left (SI.SocketException SI.Receive SI.RemoteShutdown) -> pure ()+              Left e -> throwIO e+              Right (PM.ByteArray arr) -> do+                log (FL.toLogStr (SB.fromShort (SB.SBS arr)))+                go+        go++-- Print every UDP packet that we receive. This terminates, closing the+-- socket, after receiving ten packets.+udpStdoutServer :: IO ()+udpStdoutServer = do+  unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock port -> do+    BC.putStrLn ("Receiving datagrams on 127.0.0.1:" <> BC.pack (show port))+    replicateM_ 10 $ do+      (remote,ByteArray payload) <- unhandled (DIU.receive sock 1024)+      BC.putStrLn ("Datagram from " <> BC.pack (show remote))+      BC.putStr (SB.fromShort (SB.SBS payload))++unhandled :: Exception e => IO (Either e a) -> IO a+unhandled action = action >>= either throwIO pure++gettysburgAddress :: PM.ByteArray+gettysburgAddress = PM.ByteArray arr+  where+  !(SB.SBS arr) = SB.toShort $ mconcat+    [ "Four score and seven years ago our fathers brought forth on this "+    , "continent, a new nation, conceived in Liberty, and dedicated to "+    , "the proposition that all men are created equal. Now we are engaged "+    , "in a great civil war, testing whether that nation, or any nation so "+    , "conceived and so dedicated, can long endure. We are met on a great "+    , "battle-field of that war. We have come to dedicate a portion of that "+    , "field, as a final resting place for those who here gave their lives "+    , "that that nation might live. It is altogether fitting and proper that "+    , "we should do this."+    ]
+ sockets.cabal view
@@ -0,0 +1,88 @@+cabal-version: 2.2+name: sockets+version: 0.1.0.0+synopsis: High-level network sockets+description: High-level abstraction for network sockets+homepage: https://github.com/andrewthad/sockets+bug-reports: https://github.com/andrewthad/sockets/issues+license: BSD-3-Clause+license-file: LICENSE+author: Andrew Martin+maintainer: andrew.thaddeus@gmail.com+copyright: 2019 Andrew Martin+category: Network+extra-source-files: CHANGELOG.md++flag debug+  manual: True+  description: Print debug output +  default: False++flag example+  manual: True+  description: Build example executables+  default: False++library+  exposed-modules:+    Socket.Datagram.IPv4.Undestined+    Socket.Stream.IPv4+  other-modules:+    Socket.Debug+    Socket.IPv4+    Socket+  build-depends:+    , base >= 4.11.1.0 && < 5+    , posix-api >= 0.2+    , primitive >= 0.6.4+    , ip >= 1.4.1+  hs-source-dirs: src+  if flag(debug)+    hs-source-dirs: src-debug+  else+    hs-source-dirs: src-production+  default-language: Haskell2010+  ghc-options: -O2 -Wall++test-suite test+  type: exitcode-stdio-1.0+  hs-source-dirs: test+  main-is: Main.hs+  build-depends:+    , base >= 4.11.1.0 && < 5+    , sockets+    , tasty+    , tasty-hunit+    , ip >= 1.4.1+    , primitive >= 0.6.4+    , async+  ghc-options: -Wall -O2 -threaded+  default-language: Haskell2010++benchmark macro+  type: exitcode-stdio-1.0+  build-depends:+    , base >= 4.11.1.0 && < 5+    , sockets+    , ip >= 1.4.1+    , primitive >= 0.6.4+  ghc-options: -Wall -O2 -threaded+  default-language: Haskell2010+  hs-source-dirs: bench+  main-is: Macro.hs++executable sockets-example+  if flag(example)+    build-depends:+      , base >= 4.11.1.0 && < 5+      , sockets+      , ip >= 1.4.1+      , primitive >= 0.6.4+      , bytestring >= 0.10.8.2+      , fast-logger >= 2.4.13+  else+    buildable: False+  hs-source-dirs: example+  main-is: Main.hs+  ghc-options: -Wall -O2 -threaded+  default-language: Haskell2010
+ src-debug/Socket/Debug.hs view
@@ -0,0 +1,6 @@+module Socket.Debug+  ( debug+  ) where++debug :: String -> IO ()+debug = putStrLn
+ src-production/Socket/Debug.hs view
@@ -0,0 +1,6 @@+module Socket.Debug+  ( debug+  ) where++debug :: String -> IO ()+debug _ = pure ()
+ src/Socket.hs view
@@ -0,0 +1,74 @@+{-# language BangPatterns #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}++module Socket+  ( SocketException(..)+  , Context(..)+  , Reason(..)+  ) where++import Control.Exception (Exception)+import Foreign.C.Types (CInt)++-- | Represents any unexpected behaviors that a function working on a+--   socket, connection, or listener can exhibit.+data SocketException = SocketException+  { context :: Context+  , reason :: Reason+  }+  deriving stock (Eq,Show)+  deriving anyclass (Exception)++-- | The function that behaved unexpectedly.+data Context+  = Accept+  | Bind+  | Close+  | Connect+  | GetName+  | Listen+  | Open+  | Option+  | Receive+  | Send+  | Shutdown+  deriving stock (Eq,Show)++-- | A description of the unexpected behavior.+data Reason+  = MessageTruncated !Int !Int+    -- ^ The datagram did not fit in the buffer. This can happen while+    --   sending or receiving. Fields: buffer size, datagram size.+  | SocketAddressSize+    -- ^ The socket address was not the expected size. This exception+    --   indicates a bug in this library or (less likely) in the+    --   operating system.+  | SocketAddressFamily+    -- ^ The socket address had an unexpected family. This exception+    --   indicates a bug in this library or (less likely) in the+    --   operating system.+  | OptionValueSize+    -- ^ The option value was not the expected size. This exception+    --   indicates a bug in this library or (less likely) in the+    --   operating system.+  | NegativeBytesRequested+    -- ^ The user requested a negative number of bytes in a call+    --   to a receive function.+  | RemoteNotShutdown+    -- ^ The remote end sent more data when it was expected to send+    --   a shutdown.+  | RemoteShutdown+    -- ^ The remote end has shutdown its side of the full-duplex+    --   connection. This can happen @receive@ is called on a+    --   stream socket. This is not necessarily a bad thing. Many+    --   protocols use shutdown to indicate that no more data+    --   is available. These protocols can be contrasted with+    --   protocols that send a length representing a number of+    --   expected bytes.+  | ErrorCode !CInt+    -- ^ Any error code from the operating system that this library does+    --   not expect or recognize. Consult your operating system manual+    --   for details about the error code.+  deriving stock (Eq,Show)
+ src/Socket/Datagram/IPv4/Undestined.hs view
@@ -0,0 +1,263 @@+{-# language BangPatterns #-}+{-# language DeriveAnyClass #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}+{-# language LambdaCase #-}+{-# language MagicHash #-}+{-# language NamedFieldPuns #-}++module Socket.Datagram.IPv4.Undestined+  ( -- * Types+    Socket(..)+  , Endpoint(..)+    -- * Establish+  , withSocket+    -- * Communicate+  , send+  , receive+  , receiveMutableByteArraySlice_+    -- * Exceptions+  , SocketException(..)+  , Context(..)+  , Reason(..)+    -- * Examples+    -- $examples+  ) where++import Control.Concurrent (threadWaitWrite,threadWaitRead)+import Control.Exception (mask,onException)+import Data.Primitive (ByteArray,MutableByteArray(..))+import Data.Word (Word16)+import Foreign.C.Error (Errno(..),eWOULDBLOCK,eAGAIN)+import Foreign.C.Types (CInt,CSize)+import GHC.Exts (Int(I#),RealWorld,shrinkMutableByteArray#)+import Net.Types (IPv4(..))+import Socket (SocketException(..),Context(..),Reason(..))+import Socket.Debug (debug)+import Socket.IPv4 (Endpoint(..))+import System.Posix.Types (Fd)++import qualified Control.Monad.Primitive as PM+import qualified Data.Primitive as PM+import qualified Linux.Socket as L+import qualified Posix.Socket as S++-- | A connectionless datagram socket that may communicate with many different+-- endpoints on a datagram-by-datagram basis.+newtype Socket = Socket Fd+  deriving (Eq,Ord)++-- | Open a socket and run the supplied callback on it. This closes the socket+-- when the callback finishes or when an exception is thrown. Do not return +-- the socket from the callback. This leads to undefined behavior. If the+-- address @0.0.0.0@ is used, the socket receives on all network interfaces.+-- If the port 0 is used, an unused port is chosen by the operating system.+-- The callback provides the chosen port (or if the user specified a non-zero+-- port, the chosen port will be that value).+withSocket ::+     Endpoint -- ^ Address and port to use+  -> (Socket -> Word16 -> IO a) -- ^ Callback providing the socket and the chosen port+  -> IO (Either SocketException a)+withSocket endpoint@Endpoint{port = specifiedPort} f = mask $ \restore -> do+  debug ("withSocket: opening socket " ++ show endpoint)+  e1 <- S.uninterruptibleSocket S.internet+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.datagram)+    S.defaultProtocol+  debug ("withSocket: opened socket " ++ show endpoint)+  case e1 of+    Left err -> pure (Left (errorCode Open err))+    Right fd -> do+      e2 <- S.uninterruptibleBind fd+        (S.encodeSocketAddressInternet (endpointToSocketAddressInternet endpoint))+      debug ("withSocket: requested binding for " ++ show endpoint)+      case e2 of+        Left err -> do+          -- We intentionally discard any exceptions thrown by close. There is+          -- simply nothing that can be done with them.+          S.uninterruptibleErrorlessClose fd+          pure (Left (errorCode Bind err))+        Right _ -> do+          eactualPort <- if specifiedPort == 0+            then S.uninterruptibleGetSocketName fd S.sizeofSocketAddressInternet >>= \case+              Left err -> do+                S.uninterruptibleErrorlessClose fd+                pure (Left (errorCode GetName err))+              Right (sockAddrRequiredSz,sockAddr) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet+                then case S.decodeSocketAddressInternet sockAddr of+                  Just S.SocketAddressInternet{port = actualPort} -> do+                    let cleanPort = S.networkToHostShort actualPort+                    debug ("withSocket: successfully bound " ++ show endpoint ++ " and got port " ++ show cleanPort)+                    pure (Right cleanPort)+                  Nothing -> do+                    S.uninterruptibleErrorlessClose fd+                    pure (Left (exception GetName SocketAddressFamily))+                else do+                  S.uninterruptibleErrorlessClose fd+                  pure (Left (exception GetName SocketAddressSize))+            else pure (Right specifiedPort)+          case eactualPort of+            Left err -> pure (Left err)+            Right actualPort -> do+              a <- onException (restore (f (Socket fd) actualPort)) (S.uninterruptibleErrorlessClose fd)+              S.uninterruptibleClose fd >>= \case+                Left err -> pure (Left (errorCode Close err))+                Right _ -> pure (Right a)++-- | Send a slice of a bytearray to the specified endpoint.+send ::+     Socket -- ^ Socket+  -> Endpoint -- ^ Remote IPv4 address and port+  -> ByteArray -- ^ Buffer (will be sliced)+  -> Int -- ^ Offset into payload+  -> Int -- ^ Lenth of slice into buffer+  -> IO (Either SocketException ())+send (Socket !s) !remote !payload !off !len = do+  debug ("send: about to send to " ++ show remote)+  e1 <- S.uninterruptibleSendToByteArray s payload+    (intToCInt off)+    (intToCSize len)+    mempty+    (S.encodeSocketAddressInternet (endpointToSocketAddressInternet remote))+  debug ("send: just sent to " ++ show remote)+  case e1 of+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+      then do+        debug ("send: waiting to for write ready to send to " ++ show remote)+        threadWaitWrite s+        e2 <- S.uninterruptibleSendToByteArray s payload+          (intToCInt off)+          (intToCSize len)+          mempty+          (S.encodeSocketAddressInternet (endpointToSocketAddressInternet remote))+        case e2 of+          Left err2 -> do+            debug ("send: encountered error after sending")+            pure (Left (errorCode Send err2))+          Right sz -> if csizeToInt sz == len+            then pure (Right ())+            else pure (Left (exception Send (MessageTruncated (csizeToInt sz) len)))+      else pure (Left (errorCode Send err1))+    Right sz -> if csizeToInt sz == len+      then do+        debug ("send: success")+        pure (Right ())+      else pure (Left (exception Send (MessageTruncated (csizeToInt sz) len)))++-- | Receive a datagram into a freshly allocated bytearray.+receive ::+     Socket -- ^ Socket+  -> Int -- ^ Maximum size of datagram to receive+  -> IO (Either SocketException (Endpoint,ByteArray))+receive (Socket !fd) !maxSz = do+  debug "receive: about to wait"+  threadWaitRead fd+  debug "receive: socket is now readable"+  marr <- PM.newByteArray maxSz+  -- We use MSG_TRUNC so that we are able to figure out whether+  -- or not bytes were discarded. If bytes were discarded+  -- (meaning that the buffer was too small), we return an+  -- exception.+  e <- S.uninterruptibleReceiveFromMutableByteArray fd marr 0+    (intToCSize maxSz) (L.truncate) S.sizeofSocketAddressInternet+  debug "receive: finished reading from socket"+  case e of+    Left err -> pure (Left (errorCode Receive err))+    Right (sockAddrRequiredSz,sockAddr,recvSz) -> if csizeToInt recvSz <= maxSz+      then if sockAddrRequiredSz == S.sizeofSocketAddressInternet+        then case S.decodeSocketAddressInternet sockAddr of+          Just sockAddrInet -> do+            shrinkMutableByteArray marr (csizeToInt recvSz)+            arr <- PM.unsafeFreezeByteArray marr+            pure $ Right+              ( socketAddressInternetToEndpoint sockAddrInet+              , arr+              )+          Nothing -> pure (Left (exception Receive SocketAddressFamily))+        else pure (Left (exception Receive SocketAddressSize))+      else pure (Left (exception Receive (MessageTruncated maxSz (csizeToInt recvSz))))++-- | Receive a datagram into a mutable byte array, ignoring information about+--   the remote endpoint. Returns the actual number of bytes present in the+--   datagram. Precondition: @buffer_length - offset >= max_datagram_length@.+receiveMutableByteArraySlice_ ::+     Socket -- ^ Socket+  -> MutableByteArray RealWorld -- ^ Buffer+  -> Int -- ^ Offset into buffer+  -> Int -- ^ Maximum size of datagram to receive+  -> IO (Either SocketException Int)+receiveMutableByteArraySlice_ (Socket !fd) !buf !off !maxSz = do+  threadWaitRead fd+  -- We use MSG_TRUNC so that we are able to figure out whether+  -- or not bytes were discarded. If bytes were discarded+  -- (meaning that the buffer was too small), we return an+  -- exception.+  e <- S.uninterruptibleReceiveFromMutableByteArray_ fd buf (intToCInt off) (intToCSize maxSz) (L.truncate)+  case e of+    Left err -> pure (Left (errorCode Receive err))+    Right recvSz -> if csizeToInt recvSz <= maxSz+      then pure (Right (csizeToInt recvSz))+      else pure (Left (exception Receive (MessageTruncated maxSz (csizeToInt recvSz))))++-- TODO: add receiveTimeout+-- receiveTimeout ::+--      Socket -- ^ Socket+--   -> Int -- ^ Maximum size of datagram to receive+--   -> Int -- ^ Microseconds to wait before giving up+--   -> IO (Maybe (IPv4,ByteArray))+-- receiveTimeout = error "uhoetuhntoehu"++endpointToSocketAddressInternet :: Endpoint -> S.SocketAddressInternet+endpointToSocketAddressInternet (Endpoint {address, port}) = S.SocketAddressInternet+  { port = S.hostToNetworkShort port+  , address = S.hostToNetworkLong (getIPv4 address)+  }++socketAddressInternetToEndpoint :: S.SocketAddressInternet -> Endpoint+socketAddressInternetToEndpoint (S.SocketAddressInternet {address,port}) = Endpoint+  { address = IPv4 (S.networkToHostLong address)+  , port = S.networkToHostShort port+  }++intToCInt :: Int -> CInt+intToCInt = fromIntegral++intToCSize :: Int -> CSize+intToCSize = fromIntegral++csizeToInt :: CSize -> Int+csizeToInt = fromIntegral++errorCode :: Context -> Errno -> SocketException+errorCode func (Errno x) = SocketException func (ErrorCode x)++exception :: Context -> Reason -> SocketException+exception func reason = SocketException func reason++shrinkMutableByteArray :: MutableByteArray RealWorld -> Int -> IO ()+shrinkMutableByteArray (MutableByteArray arr) (I# sz) =+  PM.primitive_ (shrinkMutableByteArray# arr sz)++{- $examples+ +Print every UDP packet that we receive. This terminates, closing the+socket, after receiving ten packets. This code throws any exception that+happens. This is commonly a useful behavior since most exceptions cannot+be handled gracefully.++> import qualified Data.ByteString.Char8 as BC+> import Control.Monad (replicateM_)+> import qualified Data.ByteString.Short.Internal as SB+> +> udpStdoutServer :: IO ()+> udpStdoutServer = do+>   unhandled $ withSocket (Endpoint IPv4.loopback 0) $ \sock port -> do+>     BC.putStrLn ("Receiving datagrams on 127.0.0.1:" <> BC.pack (show port))+>     replicateM_ 10 $ do+>       (remote,ByteArray payload) <- unhandled (receive sock 1024)+>       BC.putStrLn ("Datagram from " <> BC.pack (show remote))+>       BC.putStr (SB.fromShort (SB.SBS payload))+> +> unhandled :: Exception e => IO (Either e a) -> IO a+> unhandled action = action >>= either throwIO pure++-}
+ src/Socket/IPv4.hs view
@@ -0,0 +1,18 @@+{-# language BangPatterns #-}+{-# language DerivingStrategies #-}+{-# language DuplicateRecordFields #-}++module Socket.IPv4+  ( Endpoint(..)+  ) where++import Data.Word (Word16)+import Net.Types (IPv4(..))++-- | An endpoint for an IPv4 socket, connection, or listener.+--   Everything is in host byte order, and the user is not+--   responisble for performing any conversions.+data Endpoint = Endpoint+  { address :: !IPv4+  , port :: !Word16+  } deriving stock (Eq,Show)
+ src/Socket/Stream/IPv4.hs view
@@ -0,0 +1,493 @@+{-# language BangPatterns #-}+{-# language RankNTypes #-}+{-# language DuplicateRecordFields #-}+{-# language LambdaCase #-}+{-# language NamedFieldPuns #-}+{-# language MagicHash #-}++module Socket.Stream.IPv4+  ( -- * Types+    Listener+  , Connection+  , Endpoint(..)+    -- * Bracketed+  , withListener+  , withAccepted+  , withConnection+  , forkAccepted+  , forkAcceptedUnmasked+    -- * Communicate+  , sendByteArray+  , sendByteArraySlice+  , sendMutableByteArray+  , sendMutableByteArraySlice+  , receiveByteArray+  , receiveBoundedByteArray+  , receiveMutableByteArray+    -- * Exceptions+  , SocketException(..)+  , Context(..)+  , Reason(..)+  ) where++import Control.Concurrent (ThreadId,threadWaitWrite,threadWaitRead)+import Control.Concurrent (forkIO,forkIOWithUnmask)+import Control.Exception (mask,onException)+import Data.Bifunctor (bimap)+import Data.Primitive (ByteArray,MutableByteArray(..))+import Data.Word (Word16)+import Foreign.C.Error (Errno(..),eAGAIN,eWOULDBLOCK,eINPROGRESS)+import Foreign.C.Types (CInt,CSize)+import GHC.Exts (RealWorld,Int(I#),shrinkMutableByteArray#)+import Socket (SocketException(..),Context(..),Reason(..))+import Socket.Debug (debug)+import Socket.IPv4 (Endpoint(..))+import System.Posix.Types (Fd)+import Net.Types (IPv4(..))++import qualified Control.Monad.Primitive as PM+import qualified Data.Primitive as PM+import qualified Linux.Socket as L+import qualified Posix.Socket as S++-- | A socket that listens for incomming connections.+newtype Listener = Listener Fd++-- | A connection-oriented stream socket.+newtype Connection = Connection Fd++withListener ::+     Endpoint+  -> (Listener -> Word16 -> IO a)+  -> IO (Either SocketException a)+withListener endpoint@Endpoint{port = specifiedPort} f = mask $ \restore -> do+  debug ("withSocket: opening listener " ++ show endpoint)+  e1 <- S.uninterruptibleSocket S.internet+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream)+    S.defaultProtocol+  debug ("withSocket: opened listener " ++ show endpoint)+  case e1 of+    Left err -> pure (Left (errorCode Open err))+    Right fd -> do+      e2 <- S.uninterruptibleBind fd+        (S.encodeSocketAddressInternet (endpointToSocketAddressInternet endpoint))+      debug ("withSocket: requested binding for listener " ++ show endpoint)+      case e2 of+        Left err -> do+          _ <- S.uninterruptibleClose fd+          pure (Left (errorCode Bind err))+        Right _ -> S.uninterruptibleListen fd 16 >>= \case+          -- We hardcode the listen backlog to 16. The author is unfamiliar+          -- with use cases where gains are realized from tuning this parameter.+          -- Open an issue if this causes problems for anyone.+          Left err -> do+            _ <- S.uninterruptibleClose fd+            debug "withSocket: listen failed with error code"+            pure (Left (errorCode Listen err))+          Right _ -> do +            -- The getsockname is copied from code in Socket.Datagram.IPv4.Undestined.+            -- Consider factoring this out.+            eactualPort <- if specifiedPort == 0+              then S.uninterruptibleGetSocketName fd S.sizeofSocketAddressInternet >>= \case+                Left err -> do+                  _ <- S.uninterruptibleClose fd+                  pure (Left (errorCode GetName err))+                Right (sockAddrRequiredSz,sockAddr) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet+                  then case S.decodeSocketAddressInternet sockAddr of+                    Just S.SocketAddressInternet{port = actualPort} -> do+                      let cleanActualPort = S.networkToHostShort actualPort+                      debug ("withSocket: successfully bound listener " ++ show endpoint ++ " and got port " ++ show cleanActualPort)+                      pure (Right cleanActualPort)+                    Nothing -> do+                      _ <- S.uninterruptibleClose fd+                      pure (Left (exception GetName SocketAddressFamily))+                  else do+                    _ <- S.uninterruptibleClose fd+                    pure (Left (exception GetName SocketAddressSize))+              else pure (Right specifiedPort)+            case eactualPort of+              Left err -> pure (Left err)+              Right actualPort -> do+                a <- onException (restore (f (Listener fd) actualPort)) (S.uninterruptibleClose fd)+                S.uninterruptibleClose fd >>= \case+                  Left err -> pure (Left (errorCode Close err))+                  Right _ -> pure (Right a)++-- | Accept a connection on the listener and run the supplied callback+-- on it. This closes the connection when the callback finishes or if+-- an exception is thrown. Since this function blocks the thread until+-- the callback finishes, it is only suitable for stream socket clients+-- that handle one connection at a time. The variant 'forkAcceptedUnmasked'+-- is preferrable for servers that need to handle connections concurrently+-- (most use cases).+withAccepted ::+     Listener+  -> (Connection -> Endpoint -> IO a)+  -> IO (Either SocketException a)+withAccepted lst cb = internalAccepted+  ( \restore action -> do+    action restore+  ) lst cb++internalAccepted ::+     ((forall x. IO x -> IO x) -> ((IO a -> IO b) -> IO (Either SocketException b)) -> IO (Either SocketException c))+  -> Listener+  -> (Connection -> Endpoint -> IO a)+  -> IO (Either SocketException c)+internalAccepted wrap (Listener !lst) f = do+  threadWaitRead lst+  mask $ \restore -> do+    S.uninterruptibleAccept lst S.sizeofSocketAddressInternet >>= \case+      Left err -> pure (Left (errorCode Accept err))+      Right (sockAddrRequiredSz,sockAddr,acpt) -> if sockAddrRequiredSz == S.sizeofSocketAddressInternet+        then case S.decodeSocketAddressInternet sockAddr of+          Just sockAddrInet -> do+            let acceptedEndpoint = socketAddressInternetToEndpoint sockAddrInet+            debug ("withAccepted: successfully accepted connection from " ++ show acceptedEndpoint)+            wrap restore $ \restore' -> do+              a <- onException (restore' (f (Connection acpt) acceptedEndpoint)) (S.uninterruptibleClose acpt)+              gracefulClose acpt a+          Nothing -> do+            _ <- S.uninterruptibleClose acpt+            pure (Left (exception GetName SocketAddressFamily))+        else do+          _ <- S.uninterruptibleClose acpt+          pure (Left (exception GetName SocketAddressSize))++gracefulClose :: Fd -> a -> IO (Either SocketException a)+gracefulClose fd a = S.uninterruptibleShutdown fd S.write >>= \case+  Left err -> do+    _ <- S.uninterruptibleClose fd+    pure (Left (errorCode Shutdown err))+  Right _ -> do+    buf <- PM.newByteArray 1+    S.uninterruptibleReceiveMutableByteArray fd buf 0 1 mempty >>= \case+      Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+        then do+          threadWaitRead fd+          S.uninterruptibleReceiveMutableByteArray fd buf 0 1 mempty >>= \case+            Left err -> do+              _ <- S.uninterruptibleClose fd+              pure (Left (errorCode Shutdown err))+            Right sz -> if sz == 0+              then fmap (bimap (errorCode Close) (const a)) (S.uninterruptibleClose fd)+              else do+                debug ("Socket.Stream.IPv4.gracefulClose: remote not shutdown A")+                _ <- S.uninterruptibleClose fd+                pure (Left (exception Shutdown RemoteNotShutdown))+        else do+          _ <- S.uninterruptibleClose fd+          -- Is this the right error context? It's a call+          -- to recv, but it happens while shutting down+          -- the socket.+          pure (Left (errorCode Shutdown err1))+      Right sz -> if sz == 0+        then fmap (bimap (errorCode Close) (const a)) (S.uninterruptibleClose fd)+        else do+          debug ("Socket.Stream.IPv4.gracefulClose: remote not shutdown B")+          _ <- S.uninterruptibleClose fd+          pure (Left (exception Shutdown RemoteNotShutdown))++-- | Accept a connection on the listener and run the supplied callback in+-- a new thread. Prefer 'forkAcceptedUnmasked' unless the masking state+-- needs to be preserved for the callback. Such a situation seems unlikely+-- to the author.+forkAccepted ::+     Listener+  -> (Either SocketException a -> IO ())+  -> (Connection -> Endpoint -> IO a)+  -> IO (Either SocketException ThreadId)+forkAccepted lst consumeException cb = internalAccepted+  ( \restore action -> do+    tid <- forkIO $ do+      x <- action restore+      restore (consumeException x)+    pure (Right tid)+  ) lst cb++-- | Accept a connection on the listener and run the supplied callback in+-- a new thread. The masking state is set to @Unmasked@ when running the+-- callback.+forkAcceptedUnmasked ::+     Listener+  -> (Either SocketException a -> IO ())+  -> (Connection -> Endpoint -> IO a)+  -> IO (Either SocketException ThreadId)+forkAcceptedUnmasked lst consumeException cb = internalAccepted+  ( \_ action -> do+    tid <- forkIOWithUnmask $ \unmask -> do+      x <- action unmask+      unmask (consumeException x)+    pure (Right tid)+  ) lst cb++-- | Establish a connection to a server.+withConnection ::+     Endpoint -- ^ Remote endpoint+  -> (Connection -> IO a) -- ^ Callback to consume connection+  -> IO (Either SocketException a)+withConnection !remote f = mask $ \restore -> do+  debug ("withSocket: opening connection " ++ show remote)+  e1 <- S.uninterruptibleSocket S.internet+    (L.applySocketFlags (L.closeOnExec <> L.nonblocking) S.stream)+    S.defaultProtocol+  debug ("withSocket: opened connection " ++ show remote)+  case e1 of+    Left err1 -> pure (Left (errorCode Open err1))+    Right fd -> do+      let sockAddr = id+            $ S.encodeSocketAddressInternet+            $ endpointToSocketAddressInternet+            $ remote+      merr <- S.uninterruptibleConnect fd sockAddr >>= \case+        Left err2 -> if err2 == eINPROGRESS+          then do+            threadWaitWrite fd+            pure Nothing+          else pure (Just (errorCode Connect err2))+        Right _ -> pure Nothing+      case merr of+        Just err -> do+          _ <- S.uninterruptibleClose fd+          pure (Left err)+        Nothing -> do+          e <- S.uninterruptibleGetSocketOption fd+            S.levelSocket S.optionError (intToCInt (PM.sizeOf (undefined :: CInt)))+          case e of+            Left err -> do+              _ <- S.uninterruptibleClose fd+              pure (Left (errorCode Option err))+            Right (sz,S.OptionValue val) -> if sz == intToCInt (PM.sizeOf (undefined :: CInt))+              then+                let err = PM.indexByteArray val 0 :: CInt in+                if err == 0+                  then do+                    a <- onException (restore (f (Connection fd))) (S.uninterruptibleClose fd)+                    gracefulClose fd a+                  else do+                    _ <- S.uninterruptibleClose fd+                    pure (Left (errorCode Connect (Errno err)))+              else do+                _ <- S.uninterruptibleClose fd+                pure (Left (exception Option OptionValueSize))++sendByteArray ::+     Connection -- ^ Connection+  -> ByteArray -- ^ Buffer (will be sliced)+  -> IO (Either SocketException ())+sendByteArray conn arr =+  sendByteArraySlice conn arr 0 (PM.sizeofByteArray arr)++sendByteArraySlice ::+     Connection -- ^ Connection+  -> ByteArray -- ^ Buffer (will be sliced)+  -> Int -- ^ Offset into payload+  -> Int -- ^ Lenth of slice into buffer+  -> IO (Either SocketException ())+sendByteArraySlice !conn !payload !off0 !len0 = go off0 len0+  where+  go !off !len = if len > 0+    then internalSend conn payload off len >>= \case+      Left e -> pure (Left e)+      Right sz' -> do+        let sz = csizeToInt sz'+        go (off + sz) (len - sz)+    else pure (Right ())++sendMutableByteArray ::+     Connection -- ^ Connection+  -> MutableByteArray RealWorld -- ^ Buffer (will be sliced)+  -> IO (Either SocketException ())+sendMutableByteArray conn arr =+  sendMutableByteArraySlice conn arr 0 =<< PM.getSizeofMutableByteArray arr++sendMutableByteArraySlice ::+     Connection -- ^ Connection+  -> MutableByteArray RealWorld -- ^ Buffer (will be sliced)+  -> Int -- ^ Offset into payload+  -> Int -- ^ Lenth of slice into buffer+  -> IO (Either SocketException ())+sendMutableByteArraySlice !conn !payload !off0 !len0 = go off0 len0+  where+  go !off !len = if len > 0+    then internalSendMutable conn payload off len >>= \case+      Left e -> pure (Left e)+      Right sz' -> do+        let sz = csizeToInt sz'+        go (off + sz) (len - sz)+    else pure (Right ())++-- The length must be greater than zero.+internalSendMutable :: +     Connection -- ^ Connection+  -> MutableByteArray RealWorld -- ^ Buffer (will be sliced)+  -> Int -- ^ Offset into payload+  -> Int -- ^ Length of slice into buffer+  -> IO (Either SocketException CSize)+internalSendMutable (Connection !s) !payload !off !len = do+  e1 <- S.uninterruptibleSendMutableByteArray s payload+    (intToCInt off)+    (intToCSize len)+    mempty+  case e1 of+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+      then do+        threadWaitWrite s+        e2 <- S.uninterruptibleSendMutableByteArray s payload+          (intToCInt off)+          (intToCSize len)+          mempty+        case e2 of+          Left err2 -> pure (Left (errorCode Send err2))+          Right sz -> pure (Right sz)+      else pure (Left (errorCode Send err1))+    Right sz -> pure (Right sz)++-- The length must be greater than zero.+internalSend ::+     Connection -- ^ Connection+  -> ByteArray -- ^ Buffer (will be sliced)+  -> Int -- ^ Offset into payload+  -> Int -- ^ Length of slice into buffer+  -> IO (Either SocketException CSize)+internalSend (Connection !s) !payload !off !len = do+  debug ("send: about to send chunk on stream socket, offset " ++ show off ++ " and length " ++ show len)+  e1 <- S.uninterruptibleSendByteArray s payload+    (intToCInt off)+    (intToCSize len)+    mempty+  debug "send: just sent chunk on stream socket"+  case e1 of+    Left err1 -> if err1 == eWOULDBLOCK || err1 == eAGAIN+      then do+        debug "send: waiting to for write ready on stream socket"+        threadWaitWrite s+        e2 <- S.uninterruptibleSendByteArray s payload+          (intToCInt off)+          (intToCSize len)+          mempty+        case e2 of+          Left err2 -> do+            debug "send: encountered error after sending chunk on stream socket"+            pure (Left (errorCode Send err2))+          Right sz -> pure (Right sz)+      else pure (Left (errorCode Send err1))+    Right sz -> pure (Right sz)++-- The maximum number of bytes to receive must be greater than zero.+-- The operating system guarantees us that the returned actual number+-- of bytes is less than or equal to the requested number of bytes.+-- This function does not validate that the result size is greater+-- than zero. Functions calling this must perform that check. This+-- also does not trim the buffer. The caller must do that if it is+-- necessary.+internalReceiveMaximally ::+     Connection -- ^ Connection+  -> Int -- ^ Maximum number of bytes to receive+  -> MutableByteArray RealWorld -- ^ Receive buffer+  -> Int -- ^ Offset into buffer+  -> IO (Either SocketException Int)+internalReceiveMaximally (Connection !fd) !maxSz !buf !off = do+  debug "receive: stream socket about to wait"+  threadWaitRead fd+  debug ("receive: stream socket is now readable, receiving up to " ++ show maxSz ++ " bytes at offset " ++ show off)+  e <- S.uninterruptibleReceiveMutableByteArray fd buf (intToCInt off) (intToCSize maxSz) mempty+  debug "receive: finished reading from stream socket"+  case e of+    Left err -> pure (Left (errorCode Receive err))+    Right recvSz -> pure (Right (csizeToInt recvSz))++-- | Receive exactly the given number of bytes. If the remote application+--   shuts down its end of the connection before sending the required+--   number of bytes, this returns+--   @'Left' ('SocketException' 'Receive' 'RemoteShutdown')@.+receiveByteArray ::+     Connection -- ^ Connection+  -> Int -- ^ Number of bytes to receive+  -> IO (Either SocketException ByteArray)+receiveByteArray !conn0 !total = do+  marr <- PM.newByteArray total+  go conn0 marr 0 total+  where+  go !conn !marr !off !remaining = case compare remaining 0 of+    GT -> internalReceiveMaximally conn remaining marr off >>= \case+      Left err -> pure (Left err)+      Right sz -> if sz /= 0+        then go conn marr (off + sz) (remaining - sz)+        else pure (Left (exception Receive RemoteShutdown))+    EQ -> do+      arr <- PM.unsafeFreezeByteArray marr+      pure (Right arr)+    LT -> pure (Left (exception Receive NegativeBytesRequested))++-- | Receive a number of bytes exactly equal to the size of the mutable+--   byte array. If the remote application shuts down its end of the+--   connection before sending the required number of bytes, this returns+--   @'Left' ('SocketException' 'Receive' 'RemoteShutdown')@.+receiveMutableByteArray ::+     Connection+  -> MutableByteArray RealWorld+  -> IO (Either SocketException ())+receiveMutableByteArray !conn0 !marr0 = do+  total <- PM.getSizeofMutableByteArray marr0+  go conn0 marr0 0 total+  where+  go !conn !marr !off !remaining = if remaining > 0+    then internalReceiveMaximally conn remaining marr off >>= \case+      Left err -> pure (Left err)+      Right sz -> if sz /= 0+        then go conn marr (off + sz) (remaining - sz)+        else pure (Left (exception Receive RemoteShutdown))+    else pure (Right ())++-- | Receive up to the given number of bytes. If the remote application+--   shuts down its end of the connection instead of sending any bytes,+--   this returns+--   @'Left' ('SocketException' 'Receive' 'RemoteShutdown')@.+receiveBoundedByteArray :: +     Connection -- ^ Connection+  -> Int -- ^ Maximum number of bytes to receive+  -> IO (Either SocketException ByteArray)+receiveBoundedByteArray !conn !total+  | total > 0 = do+      m <- PM.newByteArray total+      internalReceiveMaximally conn total m 0 >>= \case+        Left err -> pure (Left err) +        Right sz -> if sz /= 0+          then do+            shrinkMutableByteArray m sz+            fmap Right (PM.unsafeFreezeByteArray m)+          else pure (Left (exception Receive RemoteShutdown))+  | total == 0 = pure (Right mempty)+  | otherwise = pure (Left (exception Receive NegativeBytesRequested))++endpointToSocketAddressInternet :: Endpoint -> S.SocketAddressInternet+endpointToSocketAddressInternet (Endpoint {address, port}) = S.SocketAddressInternet+  { port = S.hostToNetworkShort port+  , address = S.hostToNetworkLong (getIPv4 address)+  }++socketAddressInternetToEndpoint :: S.SocketAddressInternet -> Endpoint+socketAddressInternetToEndpoint (S.SocketAddressInternet {address,port}) = Endpoint+  { address = IPv4 (S.networkToHostLong address)+  , port = S.networkToHostShort port+  }++errorCode :: Context -> Errno -> SocketException+errorCode func (Errno x) = SocketException func (ErrorCode x)++exception :: Context -> Reason -> SocketException+exception func reason = SocketException func reason++intToCInt :: Int -> CInt+intToCInt = fromIntegral++intToCSize :: Int -> CSize+intToCSize = fromIntegral++csizeToInt :: CSize -> Int+csizeToInt = fromIntegral++shrinkMutableByteArray :: MutableByteArray RealWorld -> Int -> IO ()+shrinkMutableByteArray (MutableByteArray arr) (I# sz) =+  PM.primitive_ (shrinkMutableByteArray# arr sz)
+ test/Main.hs view
@@ -0,0 +1,95 @@+{-# language BangPatterns #-}+{-# language ScopedTypeVariables #-}++import Control.Concurrent.Async (concurrently)+import Control.Exception (Exception)+import Control.Exception (throwIO)+import Control.Monad.ST (runST)+import Data.Primitive (ByteArray)+import Data.Word (Word16,Word8)+import GHC.Exts (RealWorld)+import Test.Tasty+import Test.Tasty.HUnit++import qualified Socket.Datagram.IPv4.Undestined as DIU+import qualified Socket.Stream.IPv4 as SI+import qualified GHC.Exts as E+import qualified Data.Primitive as PM+import qualified Data.Primitive.MVar as PM+import qualified Net.IPv4 as IPv4++main :: IO ()+main = defaultMain tests++tests :: TestTree+tests = testGroup "socket"+  [ testGroup "datagram"+    [ testGroup "ipv4"+      [ testGroup "undestined"+        [ testCase "A" testDatagramUndestinedA+        ]+      ]+    ]+  , testGroup "stream"+    [ testGroup "ipv4"+      [ testCase "A" testStreamA+      ]+    ]+  ]++unhandled :: Exception e => IO (Either e a) -> IO a+unhandled action = action >>= either throwIO pure++testDatagramUndestinedA :: Assertion+testDatagramUndestinedA = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  (port,received) <- concurrently (sender m) (receiver m)+  received @=? (DIU.Endpoint IPv4.loopback port, message)+  where+  message = E.fromList [0,1,2,3] :: ByteArray+  sz = PM.sizeofByteArray message+  sender :: PM.MVar RealWorld Word16 -> IO Word16+  sender m = unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock srcPort -> do+    dstPort <- PM.takeMVar m+    unhandled $ DIU.send sock (DIU.Endpoint IPv4.loopback dstPort) message 0 sz+    pure srcPort+  receiver :: PM.MVar RealWorld Word16 -> IO (DIU.Endpoint,ByteArray)+  receiver m = unhandled $ DIU.withSocket (DIU.Endpoint IPv4.loopback 0) $ \sock port -> do+    PM.putMVar m port+    unhandled $ DIU.receive sock sz++-- This test involves a made up protocol that goes like this:+-- The sender always starts by sending the length of the rest+-- of the payload as a native-endian encoded machine-sized int.+-- (This could only ever work for a machine that is communicating+-- with itself). Then, it sends a bytearray of that specified+-- length. Then, both ends are expected to shutdown their sides+-- of the connection.+testStreamA :: Assertion+testStreamA = do+  (m :: PM.MVar RealWorld Word16) <- PM.newEmptyMVar+  ((),received) <- concurrently (sender m) (receiver m)+  received @=? message+  where+  message = E.fromList (enumFromTo 0 (100 :: Word8)) :: ByteArray+  sz = PM.sizeofByteArray message+  szb = runST $ do+    marr <- PM.newByteArray (PM.sizeOf (undefined :: Int))+    PM.writeByteArray marr 0 sz+    PM.unsafeFreezeByteArray marr+  sender :: PM.MVar RealWorld Word16 -> IO ()+  sender m = do+    dstPort <- PM.takeMVar m+    unhandled $ SI.withConnection (DIU.Endpoint IPv4.loopback dstPort) $ \conn -> do+      unhandled $ SI.sendByteArray conn szb+      unhandled $ SI.sendByteArray conn message+  receiver :: PM.MVar RealWorld Word16 -> IO ByteArray+  receiver m = unhandled $ SI.withListener (SI.Endpoint IPv4.loopback 0) $ \listener port -> do+    PM.putMVar m port+    unhandled $ SI.withAccepted listener $ \conn _ -> do+      serializedSize <- unhandled $ SI.receiveByteArray conn (PM.sizeOf (undefined :: Int))+      let theSize = PM.indexByteArray serializedSize 0 :: Int+      result <- unhandled $ SI.receiveByteArray conn theSize+      pure result++