diff --git a/example/Main-Client.hs b/example/Main-Client.hs --- a/example/Main-Client.hs +++ b/example/Main-Client.hs @@ -8,6 +8,7 @@ import Network.WebSockets (runServer, runClient, ServerApp, ClientApp) import Network.WebSockets.RPC +import Network.WebSockets.RPC.ACKable (ackableRPCClient) import Data.Aeson.TH (deriveJSON, defaultOptions, sumEncoding, SumEncoding (TwoElemArray)) import Network.Wai.Trans (ClientAppT, runClientAppT, ServerAppT, runServerAppT) import Control.Concurrent (threadDelay) @@ -45,27 +46,20 @@ -myClient :: (MonadIO m, MonadThrow m, MonadCatch m) => ClientAppT (WebSocketClientRPCT MyRepDSL MyComDSL m) () -myClient = rpcClient $ \dispatch -> do - -- only going to make one RPC call for this example - liftIO $ putStrLn "Subscribing Foo..." - dispatch RPCClient - { subscription = Foo - , onSubscribe = \RPCClientParams{supply,cancel} -> do - liftIO $ putStrLn "Supplying Bar..." - supply Bar - , onReply = \RPCClientParams{supply,cancel} Baz -> do - liftIO $ print Baz - liftIO $ threadDelay 1000000 - liftIO $ putStrLn "Supplying Bar..." - supply Bar - (q :: Int) <- (`mod` 10) <$> liftIO getRandom - when (q == 0) $ do - liftIO $ putStrLn "Canceling..." - cancel - , onComplete = \Qux -> - liftIO $ print Qux - } +myClient :: (MonadIO m, MonadThrow m, MonadCatch m) => RPCClient MySubDSL MySupDSL MyRepDSL MyComDSL m +myClient = RPCClient + { subscription = Foo + , onSubscribe = \RPCClientParams{supply,cancel} -> do + liftIO $ putStrLn "Supplying Bar..." + supply Bar + , onReply = \RPCClientParams{supply,cancel} Baz -> do + liftIO $ print Baz + liftIO $ threadDelay 1000000 + liftIO $ putStrLn "Supplying Bar..." + supply Bar + , onComplete = \Qux -> + liftIO $ print Qux + } @@ -73,8 +67,11 @@ main :: IO () main = do + client <- ackableRPCClient id ("client" :: String) myClient let myClient' :: ClientApp () - myClient' = runClientAppT execWebSocketClientRPCT myClient + myClient' = runClientAppT execWebSocketClientRPCT $ rpcClient $ \dispatch -> do + liftIO $ putStrLn "Subscribing Foo..." + dispatch client threadDelay 1000000 runClientAppTBackingOff id "127.0.0.1" 8080 "" myClient' diff --git a/example/Main.hs b/example/Main.hs --- a/example/Main.hs +++ b/example/Main.hs @@ -8,6 +8,7 @@ import Network.WebSockets (runServer, runClient, ServerApp, ClientApp) import Network.WebSockets.RPC +import Network.WebSockets.RPC.ACKable (ackableRPCServer) import Data.Aeson.TH (deriveJSON, defaultOptions, sumEncoding, SumEncoding (TwoElemArray)) import Network.Wai.Trans (ClientAppT, runClientAppT, ServerAppT, runServerAppT) import Control.Concurrent (threadDelay) @@ -45,8 +46,8 @@ -myServer :: (MonadIO m, MonadThrow m) => ServerAppT (WebSocketServerRPCT MySubDSL MySupDSL m) -myServer = rpcServer $ \RPCServerParams{reply,complete} eSubSup -> case eSubSup of +myServer :: (MonadIO m, MonadThrow m) => RPCServer MySubDSL MySupDSL MyRepDSL MyComDSL m +myServer RPCServerParams{reply,complete} eSubSup = case eSubSup of Left Foo -> do liftIO $ print Foo forM_ [1..5] $ \_ -> do @@ -55,17 +56,17 @@ reply Baz liftIO $ putStrLn "Completing Qux..." complete Qux - Right Bar -> do - liftIO $ print Bar - liftIO $ putStrLn "Replying Baz..." - reply Baz + Right Bar -> + liftIO $ putStrLn "Got Bar..." main :: IO () main = do + let runM = id + server <- ackableRPCServer runM ("server" :: String) myServer let myServer' :: ServerApp - myServer' = runServerAppT execWebSocketServerRPCT myServer + myServer' = runServerAppT execWebSocketServerRPCT $ rpcServer runM server runServer "127.0.0.1" 8080 myServer' diff --git a/src/Network/WebSockets/RPC.hs b/src/Network/WebSockets/RPC.hs --- a/src/Network/WebSockets/RPC.hs +++ b/src/Network/WebSockets/RPC.hs @@ -59,9 +59,10 @@ , MonadIO m , MonadThrow m ) - => RPCServer sub sup rep com m + => (forall a. m a -> IO a) + -> RPCServer sub sup rep com m -> ServerAppT (WebSocketServerRPCT sub sup m) -rpcServer f pendingConn = do +rpcServer runM f pendingConn = do conn <- liftIO (acceptRequest pendingConn) void $ liftIO $ Async.async $ forever $ do sendDataMessage conn (Text (encode (Pong :: ServerToClient () ()))) @@ -81,7 +82,7 @@ in liftIO (sendDataMessage conn (Text (encode c))) cont :: Either sub sup -> m () - cont = f RPCServerParams{reply,complete} + cont eSubSup = liftIO $ void $ Async.async $ runM $ f RPCServerParams{reply,complete} eSubSup registerSubscribeSupply _ident cont runSubscribeSupply _ident (Left _params) diff --git a/src/Network/WebSockets/RPC/ACKable.hs b/src/Network/WebSockets/RPC/ACKable.hs --- a/src/Network/WebSockets/RPC/ACKable.hs +++ b/src/Network/WebSockets/RPC/ACKable.hs @@ -2,6 +2,7 @@ RankNTypes , ScopedTypeVariables , NamedFieldPuns + , OverloadedStrings #-} module Network.WebSockets.RPC.ACKable @@ -11,12 +12,14 @@ ) where import Network.WebSockets.RPC -import Data.UUID (UUID) +import Data.UUID (UUID, toString, fromString) import Data.UUID.V4 (nextRandom) import qualified Data.HashMap.Lazy as HM import qualified Data.HashSet as HS import Data.IORef (newIORef, readIORef, writeIORef) import Data.Hashable (Hashable) +import Data.Aeson (FromJSON (..), ToJSON (..), (.:), (.=), object) +import Data.Aeson.Types (typeMismatch, Value (Object)) import Control.Applicative ((<|>)) import Control.Monad (when) import Control.Monad.IO.Class (MonadIO, liftIO) @@ -24,7 +27,7 @@ import Control.Concurrent.Async (async) import qualified Control.Concurrent.Async as Async import Control.Concurrent.STM (atomically) -import Control.Concurrent.STM.TVar (newTVarIO, readTVarIO, writeTVar, modifyTVar, TVar) +import Control.Concurrent.STM.TVar (newTVarIO, readTVarIO, readTVar, writeTVar, modifyTVar, TVar) data ACKable owner a = ACKable @@ -33,7 +36,30 @@ , ackableData :: Maybe a -- ^ 'Data.Maybe.Nothing' represents an ACK } +instance (ToJSON owner, ToJSON a) => ToJSON (ACKable owner a) where + toJSON ACKable{ackableID,ackableOwner,ackableData} = object + [ "id" .= toString ackableID + , "owner" .= ackableOwner + , "data" .= ackableData + ] +instance (FromJSON owner, FromJSON a) => FromJSON (ACKable owner a) where + parseJSON (Object o) = do + id' <- o .: "id" + case fromString id' of + Nothing -> fail "can't parse id" + Just ackableID -> do + ackableOwner <- o .: "owner" + ackableData <- o .: "data" + pure ACKable + { ackableID + , ackableOwner + , ackableData + } + parseJSON x = typeMismatch "ACKable" x + + + ack :: UUID -> owner -> ACKable owner a ack ackableID ackableOwner = ACKable { ackableID @@ -52,61 +78,68 @@ => (forall a. m a -> IO a) -> owner -> RPCServer sub sup rep com m - -> RPCServer (ACKable owner sub) (ACKable owner sup) (ACKable owner rep) com m -ackableRPCServer runM serverOwner rpc RPCServerParams{reply,complete} eSubSup = do + -> m (RPCServer (ACKable owner sub) (ACKable owner sup) (ACKable owner rep) com m) +ackableRPCServer runM serverOwner rpc = do replyMailbox <- liftIO $ newTVarIO HM.empty ownerPending <- liftIO $ newTVarIO (HM.empty :: HM.HashMap owner (HS.HashSet UUID)) - let params :: owner -> RPCServerParams rep com m - params clientOwner = RPCServerParams - { reply = \r -> do - ackableID <- liftIO nextRandom - let op = - reply ACKable - { ackableID - , ackableOwner = serverOwner - , ackableData = Just r - } - liftIO $ do - expBackoff <- mkBackoff (runM op) $ do - replies <- readTVarIO replyMailbox + pure $ \RPCServerParams{reply,complete} eSubSup -> do + let params :: owner -> RPCServerParams rep com m + params clientOwner = RPCServerParams + { reply = \r -> do + ackableID <- liftIO nextRandom + let op = do + liftIO $ putStrLn $ "Replying: " ++ show ackableID + reply ACKable + { ackableID + , ackableOwner = serverOwner + , ackableData = Just r + } + liftIO $ do + expBackoff <- mkBackoff (runM op) $ do + replies <- readTVarIO replyMailbox + putStrLn $ "Deleting: " ++ show ackableID + atomically $ do + modifyTVar replyMailbox $ HM.delete ackableID + modifyTVar ownerPending $ HM.delete clientOwner + case HM.lookup ackableID replies of + Nothing -> pure () + Just (_,expBackoff) -> Async.cancel expBackoff atomically $ do - modifyTVar replyMailbox $ HM.delete ackableID - modifyTVar ownerPending $ HM.delete clientOwner - case HM.lookup ackableID replies of - Nothing -> pure () - Just (_,expBackoff) -> Async.cancel expBackoff - atomically $ do - modifyTVar replyMailbox $ HM.insert ackableID (r, expBackoff) - modifyTVar ownerPending $ HM.insertWith HS.union clientOwner (HS.singleton ackableID) - op - , complete - } + modifyTVar replyMailbox $ HM.insert ackableID (r, expBackoff) + modifyTVar ownerPending $ HM.insertWith HS.union clientOwner (HS.singleton ackableID) + keys <- HM.keys <$> readTVarIO replyMailbox + print keys + op + , complete + } - case eSubSup of - Left ACKable{ackableID,ackableOwner,ackableData} -> - case ackableData of - Nothing -> - liftIO $ putStrLn "Somehow received an ACK from a Sub on a server" - Just sub -> do - reply (ack ackableID serverOwner) - rpc (params ackableOwner) (Left sub) - Right ACKable{ackableID,ackableOwner,ackableData} -> - case ackableData of - Nothing -> do - replies <- liftIO $ readTVarIO replyMailbox - owners <- liftIO $ readTVarIO ownerPending - case HM.lookup ackableID replies of - Nothing -> liftIO $ putStrLn $ "Somehow received an ACK that doesn't exist: " ++ show ackableID - Just (_,expBackoff) -> liftIO $ do - Async.cancel expBackoff - atomically $ do - writeTVar replyMailbox $ HM.delete ackableID replies - writeTVar ownerPending $ HM.adjust (HS.delete ackableID) ackableOwner owners + case eSubSup of + Left ACKable{ackableID,ackableOwner,ackableData} -> case ackableData of + Nothing -> liftIO $ putStrLn $ "Somehow was provided a Sub ACK: " ++ show ackableID + Just sub -> rpc (params ackableOwner) (Left sub) + Right ACKable{ackableID,ackableOwner,ackableData} -> + case ackableData of + Nothing -> liftIO $ do + mExpBackoff <- atomically $ do + replies <- readTVar replyMailbox + owners <- readTVar ownerPending + case HM.lookup ackableID replies of + Nothing -> pure Nothing + Just (_,expBackoff) -> do + writeTVar replyMailbox $ HM.delete ackableID replies + writeTVar ownerPending $ HM.adjust (HS.delete ackableID) ackableOwner owners + pure (Just expBackoff) + keys <- HM.keys <$> readTVarIO replyMailbox + putStrLn $ "ACK keys: " ++ show keys + case mExpBackoff of + Nothing -> putStrLn $ "Somehow received an ACK that doesn't exist: " ++ show ackableID + Just expBackoff -> Async.cancel expBackoff - Just sup -> do - reply (ack ackableID serverOwner) - rpc (params ackableOwner) (Right sup) + Just sup -> do + liftIO $ putStrLn $ "Acknowleding: " ++ show ackableID + reply (ack ackableID serverOwner) + rpc (params ackableOwner) (Right sup) @@ -120,17 +153,16 @@ -> RPCClient sub sup rep com m -> m (RPCClient (ACKable owner sub) (ACKable owner sup) (ACKable owner rep) com m) ackableRPCClient runM clientOwner RPCClient{subscription,onSubscribe,onReply,onComplete} = do - subscriptionMailbox <- liftIO $ newTVarIO HM.empty supplyMailbox <- liftIO $ newTVarIO HM.empty ownerPending <- liftIO $ newTVarIO (HM.empty :: HM.HashMap owner (HS.HashSet UUID)) - ackableID <- liftIO nextRandom let ackParams :: Maybe owner -> RPCClientParams (ACKable owner sup) m -> RPCClientParams sup m ackParams mOwner RPCClientParams{supply,cancel} = RPCClientParams { supply = \s -> do ackableID <- liftIO nextRandom - let op = + let op = do + liftIO $ putStrLn $ "Supplying: " ++ show ackableID supply ACKable { ackableID , ackableOwner = clientOwner @@ -138,18 +170,15 @@ } liftIO $ do expBackoff <- mkBackoff (runM op) $ do - subscriptions <- readTVarIO subscriptionMailbox supplies <- readTVarIO supplyMailbox atomically $ do - modifyTVar subscriptionMailbox $ HM.delete ackableID modifyTVar supplyMailbox $ HM.delete ackableID case mOwner of Nothing -> pure () Just serverOwner -> modifyTVar ownerPending $ HM.delete serverOwner - case Left <$> HM.lookup ackableID subscriptions <|> Right <$> HM.lookup ackableID supplies of - Nothing -> pure () - Just (Left (_,expBackoff)) -> Async.cancel expBackoff - Just (Right (_,expBackoff)) -> Async.cancel expBackoff + case HM.lookup ackableID supplies of + Nothing -> pure () -- deleting self + Just (_,expBackoff) -> Async.cancel expBackoff atomically $ do modifyTVar supplyMailbox $ HM.insert ackableID (s,expBackoff) case mOwner of @@ -158,6 +187,7 @@ op , cancel } + ackableID <- liftIO nextRandom pure RPCClient { subscription = ACKable { ackableID @@ -166,23 +196,21 @@ } , onSubscribe = onSubscribe . ackParams Nothing , onReply = \params ACKable{ackableID,ackableData,ackableOwner} -> case ackableData of - Nothing -> do - subscriptions <- liftIO $ readTVarIO subscriptionMailbox - supplies <- liftIO $ readTVarIO supplyMailbox - owners <- liftIO $ readTVarIO ownerPending - case Left <$> HM.lookup ackableID subscriptions <|> Right <$> HM.lookup ackableID supplies of - Nothing -> pure () - Just (Left (_,expBackoff)) -> liftIO $ do - atomically $ do - writeTVar subscriptionMailbox $ HM.delete ackableID subscriptions - writeTVar ownerPending $ HM.adjust (HS.delete ackableID) ackableOwner owners - Async.cancel expBackoff - Just (Right (_,expBackoff)) -> liftIO $ do - atomically $ do + Nothing -> liftIO $ do + mExpBackoff <- atomically $ do + supplies <- readTVar supplyMailbox + owners <- readTVar ownerPending + case HM.lookup ackableID supplies of + Nothing -> pure Nothing + Just (_,expBackoff) -> do writeTVar supplyMailbox $ HM.delete ackableID supplies writeTVar ownerPending $ HM.adjust (HS.delete ackableID) ackableOwner owners - Async.cancel expBackoff + pure (Just expBackoff) + case mExpBackoff of + Nothing -> putStrLn $ "Somehow received an ACK that doesn't exist: " ++ show ackableID + Just expBackoff -> Async.cancel expBackoff Just rep -> do + liftIO $ putStrLn $ "Acknowledging: " ++ show ackableID (supply params) (ack ackableID clientOwner) onReply (ackParams (Just ackableOwner) params) rep , onComplete @@ -197,6 +225,9 @@ day = 24 * hour week = 7 * day +mkBackoff :: IO a -- ^ Invoked each attempt + -> IO () -- ^ on quit + -> IO (Async.Async a) mkBackoff op x = do spentWaiting <- newIORef (0 :: Int) async $ do diff --git a/websockets-rpc.cabal b/websockets-rpc.cabal --- a/websockets-rpc.cabal +++ b/websockets-rpc.cabal @@ -1,5 +1,5 @@ Name: websockets-rpc -Version: 0.2.0 +Version: 0.3.0 Author: Athan Clark Maintainer: Athan Clark License: BSD3