websockets-rpc 0.2.0 → 0.3.0
raw patch · 5 files changed
+137/−107 lines, 5 files
Files
- example/Main-Client.hs +19/−22
- example/Main.hs +8/−7
- src/Network/WebSockets/RPC.hs +4/−3
- src/Network/WebSockets/RPC/ACKable.hs +105/−74
- websockets-rpc.cabal +1/−1
example/Main-Client.hs view
@@ -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'
example/Main.hs view
@@ -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'
src/Network/WebSockets/RPC.hs view
@@ -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)
src/Network/WebSockets/RPC/ACKable.hs view
@@ -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
websockets-rpc.cabal view
@@ -1,5 +1,5 @@ Name: websockets-rpc-Version: 0.2.0+Version: 0.3.0 Author: Athan Clark <athan.clark@gmail.com> Maintainer: Athan Clark <athan.clark@gmail.com> License: BSD3