curryer-rpc 0.3.8 → 0.4.0

raw patch · 9 files changed

+173/−53 lines, 9 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Changelog.markdown +4/−0
• bench/Basic.hs +2/−2
• curryer-rpc.cabal +1/−1
• examples/SimpleKeyValueClient.hs +1/−1
• examples/SimpleKeyValueServer.hs +1/−1
• src/Network/RPC/Curryer/Client.hs +51/−7
• src/Network/RPC/Curryer/Server.hs +51/−7
• src/Network/RPC/Curryer/StreamlyAdditions.hs +12/−16
• test/Curryer/Test/Basic.hs +50/−18

@@ -1,3 +1,7 @@+# v0.4.0 (2025-03-29)++* add ability to serve and connect to IPv6 and Unix domain sockets (#5)+ # v0.3.8 (2024-12-20)  * add support for winery 1.5 but don't require it

@@ -15,10 +15,10 @@ main = do   --start server shared across benchmarks- gather stats on round-trip requests-responses   portReadyMVar <- newEmptyMVar-  server <- async (serve benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))+  server <- async (serveIPv4 benchmarkServerRequestHandlers () localHostAddr 0 (Just portReadyMVar))   --wait for server to be ready   (SockAddrInet port _) <- takeMVar portReadyMVar-  clientConn <- connect [] localHostAddr port+  clientConn <- connectIPv4 [] localHostAddr port    _ <- syncWaitRequest clientConn 100   let syncWait ms = nfIO (syncWaitRequest clientConn ms)

@@ -1,5 +1,5 @@ Name: curryer-rpc-Version: 0.3.8+Version: 0.4.0 License: PublicDomain Build-Type: Simple Homepage: https://github.com/agentm/curryer

@@ -23,7 +23,7 @@ main = do   opts <- getRecord "SimpleKeyValueClient"   -- connect to the remote server (in this case on the localhost address)-  conn <- connect [] localHostAddr 8765+  conn <- connectIPv4 [] localHostAddr 8765   case opts of     Get k -> do       --call the remote function and validate the result

@@ -19,7 +19,7 @@ main :: IO () main = do   kvmap <- M.newIO-  void $ serve kvRequestHandlers kvmap localHostAddr 8765 Nothing+  void $ serveIPv4 kvRequestHandlers kvmap localHostAddr 8765 Nothing  -- setup incoming request handlers to operate on the server's state kvRequestHandlers :: RequestHandlers (M.Map String String)

@@ -1,8 +1,9 @@ {-# LANGUAGE RankNTypes, ScopedTypeVariables, GADTs #-} module Network.RPC.Curryer.Client where import Network.RPC.Curryer.Server-import Network.Socket as Socket-import qualified Streamly.Network.Inet.TCP as TCP+import Network.Socket as Socket (Socket, PortNumber, SockAddr(..), close, Family(..), SocketType(..), tupleToHostAddress, tupleToHostAddress6)+import Streamly.Internal.Network.Socket (SockSpec(..))+import qualified Streamly.Internal.Network.Socket as SINS import Codec.Winery import Control.Concurrent.Async import qualified Data.UUID.V4 as UUIDBase@@ -27,15 +28,57 @@  type ClientAsyncRequestHandlers = [ClientAsyncRequestHandler] +-- | Connect to a remote server over IPv4. Wraps `connect`.+connectIPv4 ::+  ClientAsyncRequestHandlers ->+  HostAddressTuple ->+  PortNumber ->+  IO Connection+connectIPv4 asyncHandlers hostaddr portnum =+  connect asyncHandlers sockSpec sockAddr+  where+    sockSpec = SINS.SockSpec { sockFamily = AF_INET,+                               sockType = Stream,+                               sockProto = 0,+                               sockOpts = [] }+    sockAddr = SockAddrInet portnum (tupleToHostAddress hostaddr)++-- | Connect to a remote server over IPv6. Wraps `connect`.+connectIPv6 ::+  ClientAsyncRequestHandlers ->+  HostAddressTuple6 ->+  PortNumber ->+  IO Connection+connectIPv6 asyncHandlers hostaddr portnum =+  connect asyncHandlers sockSpec sockAddr  +  where+    sockSpec = SINS.SockSpec { sockFamily = AF_INET6,+                               sockType = Stream,+                               sockProto = 0,+                               sockOpts = [] }+    sockAddr = SockAddrInet6 portnum 0 (tupleToHostAddress6 hostaddr) 0++connectUnixDomain ::+  ClientAsyncRequestHandlers ->+  FilePath ->+  IO Connection+connectUnixDomain asyncHandlers socketPath =+  connect asyncHandlers sockSpec sockAddr+  where+    sockSpec = SINS.SockSpec { sockFamily = AF_UNIX,+                               sockType = Stream,+                               sockProto = 0,+                               sockOpts = [] }+    sockAddr = SockAddrUnix socketPath+ -- | Connects to a remote server with specific async callbacks registered. connect ::    ClientAsyncRequestHandlers ->-  HostAddr ->-  PortNumber ->+  SINS.SockSpec ->+  SockAddr ->   IO Connection-connect asyncHandlers hostAddr portNum = do-  sock <- TCP.connect hostAddr portNum-  Socket.setSocketOption sock NoDelay 1+connect asyncHandlers sockSpec sockAddr = do+  sock <- SINS.connect sockSpec sockAddr   syncmap <- STMMap.newIO   asyncThread <- async (clientAsync sock syncmap asyncHandlers)   sockLock <- newLock sock@@ -145,3 +188,4 @@       fprint = fingerprint msg   sendEnvelope envelope (_conn_sockLock conn)   pure (Right ())+

@@ -169,15 +169,19 @@  instance Exception TimeoutException   -type HostAddr = (Word8, Word8, Word8, Word8)- type BParser a = Parser Word8 IO a -allHostAddrs,localHostAddr :: HostAddr+type HostAddressTuple = (Word8, Word8, Word8, Word8)+type HostAddressTuple6 = (Word16, Word16, Word16, Word16, Word16, Word16, Word16, Word16)+                        +allHostAddrs,localHostAddr :: HostAddressTuple allHostAddrs = (0,0,0,0) localHostAddr = (127,0,0,1) +localHostAddr6 :: HostAddressTuple6+localHostAddr6 = (0,0,0,0,0,0,0,1) + msgTypeP :: BParser MessageType msgTypeP = (P.satisfy (== 0) *>              (RequestMessage . fromIntegral <$> word32P)) <|>@@ -261,19 +265,59 @@  type NewMessageHandler req resp = req -> IO resp +defaultSocketOptions :: [(SocketOption, Int)]+defaultSocketOptions = [(ReuseAddr, 1), (NoDelay, 1)]++-- | Listen for new connections and handle requests on an IPv4 address. Wraps `serve1.+serveIPv4 :: RequestHandlers s -> s -> HostAddressTuple -> PortNumber -> Maybe (MVar SockAddr) -> IO Bool+serveIPv4 handlers state hostaddr port mSockLock =+  serve handlers state sockSpec sockAddr mSockLock+  where+    sockAddr = SockAddrInet port (tupleToHostAddress hostaddr)+    sockSpec = SockSpec { sockFamily = AF_INET,+                          sockType = Stream,+                          sockProto = 0,+                          sockOpts = defaultSocketOptions+                        }++-- | Listen for IPv6 RPC requests. Wraps `serve`.+serveIPv6 :: RequestHandlers s -> s -> HostAddressTuple6 -> PortNumber -> Maybe (MVar SockAddr) -> IO Bool+serveIPv6 handlers state hostaddr port mSockLock =+  serve handlers state sockSpec sockAddr mSockLock+  where+    flowInfo = 0+    scopeInfo = 0+    sockAddr = SockAddrInet6 port flowInfo (tupleToHostAddress6 hostaddr) scopeInfo+    sockSpec = SockSpec { sockFamily = AF_INET6,+                          sockType = Stream,+                          sockProto = 0,+                          sockOpts = defaultSocketOptions+                          }++-- | Listen for Unix domain socket RPC requests. Wraps `serve`.+serveUnixDomain :: RequestHandlers s -> s -> FilePath -> Maybe (MVar SockAddr) -> IO Bool+serveUnixDomain handlers state socketPath mSockLock =+  serve handlers state sockSpec sockAddr mSockLock+  where+    sockSpec = SockSpec { sockFamily = AF_UNIX,+                          sockType = Stream,+                          sockProto = 0,+                          sockOpts = [] }+    sockAddr = SockAddrUnix socketPath+   -- | Listen for new connections and handle requests which are passed the server state 's'. The MVar SockAddr can be be optionally used to know when the server is ready for processing requests. serve ::           RequestHandlers s->          s ->-         HostAddr ->-         PortNumber ->+         SockSpec ->+         SockAddr ->           Maybe (MVar SockAddr) ->          IO Bool-serve userMsgHandlers serverState hostaddr port mSockLock = do+serve userMsgHandlers serverState sockSpec sockAddr mSockLock = do   let handleSock sock = do         lockingSocket <- newLock sock         drainSocketMessages sock (serverEnvelopeHandler lockingSocket userMsgHandlers serverState)-  Stream.unfold (SA.acceptorOnAddr [(ReuseAddr, 1), (NoDelay, 1)] mSockLock) (hostaddr, port) +  Stream.unfold (SA.acceptorOnSockSpec sockSpec mSockLock) sockAddr    & Stream.parMapM id handleSock    & Stream.fold FL.drain   pure True

@@ -1,32 +1,28 @@ {-# LANGUAGE CPP #-} module Network.RPC.Curryer.StreamlyAdditions where import Control.Monad.IO.Class-import Network.Socket (Socket, PortNumber, SocketOption, SockAddr(..), maxListenQueue, Family(..), SocketType(..), defaultProtocol, tupleToHostAddress, withSocketsDo, socket, setSocketOption, bind, getSocketName)+import Network.Socket (Socket, SockAddr(..), maxListenQueue, withSocketsDo, socket, setSocketOption, bind, getSocketName) import qualified Network.Socket as Net import Control.Exception (onException) import Control.Concurrent.MVar-import Data.Word import qualified Streamly.Internal.Data.Unfold as UF import Streamly.Network.Socket hiding (acceptor) import qualified Streamly.Internal.Data.Stream as D import Streamly.Internal.Data.Unfold (Unfold(..))+import Control.Monad (when)+--import Streamly.Internal.Network.Socket as INS -acceptorOnAddr+acceptorOnSockSpec     :: MonadIO m-    => [(SocketOption, Int)]+    => SockSpec     -> Maybe (MVar SockAddr)-    -> Unfold m ((Word8, Word8, Word8, Word8), PortNumber) Socket-acceptorOnAddr opts mLock = UF.lmap f (acceptor mLock)+    -> Unfold m SockAddr Socket+acceptorOnSockSpec sockSpec mLock = UF.lmap f (acceptor mLock)     where-    f (addr, port) =+    f sockAddr' =         (maxListenQueue-        , SockSpec-            { sockFamily = AF_INET-            , sockType = Stream-            , sockProto = defaultProtocol -- TCP-            , sockOpts = opts-            }-        , SockAddrInet port (tupleToHostAddress addr)+        , sockSpec+        , sockAddr'         )  acceptor :: MonadIO m => Maybe (MVar SockAddr) -> Unfold m (Int, SockSpec, SockAddr) Socket@@ -60,7 +56,7 @@     where      use sock = do-        mapM_ (uncurry (setSocketOption sock)) (sockOpts sockSpec)+        when (not (null (sockOpts sockSpec))) $ mapM_ (uncurry (setSocketOption sock)) (sockOpts sockSpec)         bind sock addr-        Net.listen sock listenQLen        +        Net.listen sock listenQLen         

@@ -30,6 +30,8 @@   ,testCase "multi-threaded client" testMultithreadedClient   ,testCase "server state" testServerState   ,testCase "request handler throws timeout" testRequestHandlerThrowTimeout+  ,testCase "ipv6 server" testIPv6Server+  ,testCase "unix domain socket server" testUnixDomainSocketServer   ]  @@ -102,10 +104,10 @@ testSimpleCall = do   readyVar <- newEmptyMVar         -  server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))   --wait for server to be ready   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs     x <- call conn (AddTwoNumbersReq 1 1)     assertEqual "server request+response" (Right (2 :: Int)) x@@ -121,9 +123,9 @@   let clientAsyncHandlers =         [ClientAsyncRequestHandler (\(AsyncHelloReq s) ->                                         putMVar receivedAsyncMessageVar s)]-  server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))+  server <- async (serveIPv4 (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))   (SockAddrInet port _) <- takeMVar portReadyVar-  conn <- connect clientAsyncHandlers localHostAddr port+  conn <- connectIPv4 clientAsyncHandlers localHostAddr port   Right () <- asyncCall conn (AsyncHelloReq "welcome")   asyncMessage <- takeMVar receivedAsyncMessageVar   assertEqual "async message" "welcome" asyncMessage@@ -140,10 +142,10 @@   portReadyVar <- newEmptyMVar   receivedAsyncMessageVar <- newEmptyMVar   -  server <- async (serve (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))+  server <- async (serveIPv4 (testServerRequestHandlers (Just receivedAsyncMessageVar)) emptyServerState localHostAddr 0 (Just portReadyVar))   (SockAddrInet port _) <- takeMVar portReadyVar -  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   --send an async message, wait for an async response to confirm receipt   Right () <- asyncCall conn (TestCallMeBackReq "hi server")   asyncMessage <- takeMVar receivedAsyncMessageVar@@ -155,10 +157,10 @@ testSyncClientCallTimeout = do   readyVar <- newEmptyMVar         -  server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))   --wait for server to be ready   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   x <- callTimeout @_ @Int (Just 500) conn (DelayMicrosecondsReq 1000)   assertEqual "client sync timeout" (Left TimeoutError) x   close conn@@ -168,10 +170,10 @@ testSyncException = do   readyVar <- newEmptyMVar         -  server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))   (SockAddrInet port _) <- takeMVar readyVar -  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   ret <- call conn ThrowServerSideExceptionReq   case ret of     Left (ExceptionError actualExc) ->@@ -187,9 +189,9 @@ testMultithreadedClient = do   readyVar <- newEmptyMVar         -  server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   let bigString = replicate (1024 * 1000) 'x'   replicateM_ 10 $ do     ret <- call conn (RoundtripStringReq bigString)@@ -207,9 +209,9 @@                                              modifyTVar (connectionServerState sState) (+ 1))                                              ]   serverState <- newTVarIO @Int 1-  server <- async (serve serverHandlers serverState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 serverHandlers serverState localHostAddr 0 (Just readyVar))   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   ret <- call conn ChangeServerState   assertEqual "server ret" (Right ()) ret   serverState' <- readTVarIO serverState@@ -226,9 +228,9 @@                                           throw TimeoutException >> pure (1 :: Int)                                           )                                              ]-  server <- async (serve serverHandlers () localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 serverHandlers () localHostAddr 0 (Just readyVar))   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   ret <- call @_ @Int conn ThrowTimeout   assertEqual "handler timeout exception" (Left TimeoutError) ret   close conn@@ -244,12 +246,42 @@ testComplexADT = do   let arg = SomeTextCon (pack "sduofhsldkfsldkfjsldkfjsdlkfjsdlfksdjlfksjdflksdjflksdjfjlsdjlfjdklskfjlsdlfk") (SomeIntCon 0 (EndCon))   readyVar <- newEmptyMVar-  server <- async (serve (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))+  server <- async (serveIPv4 (testServerRequestHandlers Nothing) emptyServerState localHostAddr 0 (Just readyVar))   --wait for server to be ready   (SockAddrInet port _) <- takeMVar readyVar-  conn <- connect [] localHostAddr port+  conn <- connectIPv4 [] localHostAddr port   replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs     res <- call conn (RoundtripSomeADTReq arg)     assertEqual "complex ADT" (Right arg) res   close conn   cancel server++testIPv6Server :: Assertion+testIPv6Server = do+  readyVar <- newEmptyMVar+        +  server <- async (serveIPv6 (testServerRequestHandlers Nothing) emptyServerState localHostAddr6 0 (Just readyVar))+  --wait for server to be ready+  (SockAddrInet6 port _ _ _) <- takeMVar readyVar+  conn <- connectIPv6 [] localHostAddr6 port+  replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs+    x <- call conn (AddTwoNumbersReq 1 1)+    assertEqual "server request+response" (Right (2 :: Int)) x+  close conn+  cancel server+  +testUnixDomainSocketServer :: Assertion+testUnixDomainSocketServer = do+  readyVar <- newEmptyMVar+  let socketPath = "/tmp/curryertest.sock"+  server <- async (serveUnixDomain (testServerRequestHandlers Nothing) emptyServerState socketPath (Just readyVar))++  --wait for server to be ready+  (SockAddrUnix socketPath') <- takeMVar readyVar+  conn <- connectUnixDomain [] socketPath'+  replicateM_ 5 $ do --make five AddTwo calls to shake out parallelism bugs+    x <- call conn (AddTwoNumbersReq 1 1)+    assertEqual "server request+response" (Right (2 :: Int)) x+  close conn+  cancel server+