websockets-rpc 0.6.0 → 0.7.0
raw patch · 7 files changed
+213/−301 lines, 7 filesdep −MonadRandomdep ~basedep ~websocketsdep ~websockets-simple
Dependencies removed: MonadRandom
Dependency ranges changed: base, websockets, websockets-simple
Files
- README.md +95/−0
- example/Main-Client.hs +0/−89
- example/Main.hs +0/−79
- src/Network/WebSockets/RPC.hs +19/−21
- src/Network/WebSockets/RPC/ACKable.hs +8/−9
- src/Network/WebSockets/RPC/Trans/Client.hs +2/−2
- websockets-rpc.cabal +89/−101
+ README.md view
@@ -0,0 +1,95 @@+websockets-rpc+===============++A simple message-based streaming RPC mechanism built using WebSockets++## Usage++The idea is pretty simple:++- A client initiates the RPC call to a server with a `Subscription`+- the client may send additional data at any time with `Supply`, who can also cancel the RPC call+- the server may respond incrementally with `Reply`+- the server finishes the RPC call with a `Complete`++```+client server+ ------------------------subscribe--------------------------->+ - - - - - - - - - - - - -supply - - - - - - - - - - - - - - >+ < - - - - - - - - - - - -reply- - - - - - - - - - - - - - - -+ <-----------------------complete-----------------------------+```++if the `supply` and `reply` parts were ommitted, it would be identical to a traditional RPC mechanism.+++### Example++```haskell+import Network.WebSockets.RPC+import Data.Aeson++-- subscriptions from client to server+data MySubDSL = Foo+ deriving (FromJSON, ToJSON) -- you should figure this part out :)++-- supplies from client to server+data MySupDSL = Bar+ deriving (FromJSON, ToJSON)++-- replies from server to client+data MyRepDSL = Baz+ deriving (FromJSON, ToJSON)++-- onCompletes from server to client+data MyComDSL = Qux+ deriving (FromJSON, ToJSON)+```+++Server:++```haskell+{-# LANGUAGE NamedFieldPuns, ScopedTypeVariables #-}+++myServer :: (MonadIO m, MonadThrow m) => ServerAppT (WebSocketServerRPCT MySubDSL MySupDSL m)+myServer = rpcServer $ \RPCServerParams{reply,complete} eSubSup -> case eSubSup of+ Left Foo -> do+ forM_ [1..5] $ \_ -> do+ liftIO $ threadDelay 1000000+ reply Baz+ complete Qux+ Right Bar -> reply Baz+```++Client:++```haskell+{-# LANGUAGE NamedFieldPuns #-}+++myClient :: (MonadIO m, MonadThrow m) => ClientAppT (WebSocketClientRPCT MyRepDSL MyComDSL m) ()+myClient = rpcClient $ \dispatch ->+ -- only going to make one RPC call for this example+ dispatch RPCClient+ { subscription = Foo+ , onReply = \RPCClientParams{supply,cancel} Baz -> do+ liftIO $ threadDelay 1000000+ supply Bar+ (q :: Bool) <- liftIO getRandom+ if q then cancel else pure ()+ , onComplete = \Qux -> liftIO $ putStrLn "finished"+ }+```++> the `threadDelay` calls are just to exemplify the asynchronisity of the system, nothing to do with avoiding race conditions >.>+++To turn the `ServerAppT` and `ClientAppT` into natural [WebSockets](https://hackage.haskell.org/package/websockets)+types, use the morphisms from [Wai-Transformers](https://hackage.haskell.org/package/wai-trasformers).+++## Contributing++this is my swamp
− example/Main-Client.hs
@@ -1,89 +0,0 @@-{-# LANGUAGE- TemplateHaskell- , NamedFieldPuns- , ScopedTypeVariables- #-}--module Main where--import Network.Wai.Trans (runClientAppT)-import Network.WebSockets (runServer, runClient, ServerApp, ClientApp)-import Network.WebSockets.Simple (toClientAppT', hoistWebSocketsApp, expBackoffStrategy)-import Network.WebSockets.RPC-import Network.WebSockets.RPC.ACKable (ackableRPCClient)-import Network.WebSockets.RPC.Trans.Client (newEnv, Env, runWebSocketClientRPCT')-import Data.Aeson.TH (deriveJSON, defaultOptions, sumEncoding, SumEncoding (TwoElemArray))-import Network.Wai.Trans (ClientAppT, runClientAppT, ServerAppT, runServerAppT)-import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, link)-import Control.Monad (forM_, when, void)-import Control.Monad.Trans (lift)-import Control.Monad.IO.Class (liftIO, MonadIO)-import Control.Monad.Catch (MonadThrow, MonadCatch)-import Control.Monad.Random.Class (getRandom)----- subscriptions from client to server-data MySubDSL = Foo- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MySubDSL)---- supplies from client to server-data MySupDSL = Bar- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MySupDSL)---- replies from server to client-data MyRepDSL = Baz- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MyRepDSL)---- onCompletes from server to client-data MyComDSL = Qux- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MyComDSL)-----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- }------main :: IO ()-main = do- env <- newEnv-- let runM = id-- runWS :: WebSocketClientRPCT MyRepDSL MyComDSL IO a -> IO a- runWS = runWebSocketClientRPCT' env-- client <- runWebSocketClientRPCT' env (rpcClientSimple (\_ -> putStrLn "connection closed"- ) myClient)-- -- client <- ackableRPCClient id ("client" :: String) myClient- let myClient' :: ClientApp ()- myClient' = runClientAppT runM $ toClientAppT' $ runWebSocketClientRPCTSimple runWS client-- threadDelay 1000000- expBackoffStrategy $ runClient "127.0.0.1" 8080 "" $ runClientAppT id myClient'
− example/Main.hs
@@ -1,79 +0,0 @@-{-# LANGUAGE- TemplateHaskell- , NamedFieldPuns- , ScopedTypeVariables- #-}--module Main where--import Network.WebSockets (runServer, runClient, ServerApp, ClientApp)-import Network.WebSockets.Simple (toServerAppT, hoistWebSocketsApp)-import Network.WebSockets.RPC-import Network.WebSockets.RPC.ACKable (ackableRPCServer)-import Network.WebSockets.RPC.Trans.Server (newEnv, Env, runWebSocketServerRPCT')-import Data.Aeson.TH (deriveJSON, defaultOptions, sumEncoding, SumEncoding (TwoElemArray))-import Network.Wai.Trans (ClientAppT, runClientAppT, ServerAppT, runServerAppT)-import Control.Concurrent (threadDelay)-import Control.Concurrent.Async (async, link)-import Control.Monad (forM_, when, void)-import Control.Monad.Trans (lift)-import Control.Monad.IO.Class (liftIO, MonadIO)-import Control.Monad.Catch (MonadThrow)-import Control.Monad.Random.Class (getRandom)----- subscriptions from client to server-data MySubDSL = Foo- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MySubDSL)---- supplies from client to server-data MySupDSL = Bar- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MySupDSL)---- replies from server to client-data MyRepDSL = Baz- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MyRepDSL)---- onCompletes from server to client-data MyComDSL = Qux- deriving (Show, Eq)--$(deriveJSON defaultOptions{sumEncoding = TwoElemArray} ''MyComDSL)-----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- liftIO $ threadDelay 1000000- liftIO $ putStrLn "Replying Baz..."- reply Baz- liftIO $ putStrLn "Completing Qux..."- complete Qux- Right Bar ->- liftIO $ putStrLn "Got Bar..."-----main :: IO ()-main = do- let runM = id-- server <- execWebSocketServerRPCTSimple $ rpcServerSimple ( \_ -> putStrLn "connection closed"- ) myServer-- -- server <- ackableRPCServer runM ("server" :: String) myServer- let myServer' :: ServerApp- myServer' = runServerAppT runM $ toServerAppT server-- runServer "127.0.0.1" 8080 myServer'
src/Network/WebSockets/RPC.hs view
@@ -35,12 +35,10 @@ , RPCIdentified (..) ) import Network.WebSockets (acceptRequest, receiveDataMessage, sendDataMessage, DataMessage (Text, Binary), ConnectionException, runClient)-import Network.WebSockets.Simple (WebSocketsApp (..), hoistWebSocketsApp, WebSocketsAppParams (..))+import Network.WebSockets.Simple (WebSocketsApp (..), hoistWebSocketsApp, WebSocketsAppParams (..), CloseOrigin) import Network.Wai.Trans (ServerAppT, ClientAppT, runClientAppT) import Data.Aeson (ToJSON, FromJSON, decode, encode) import Data.IORef (newIORef, readIORef, writeIORef)-import Data.Word (Word16)-import Data.ByteString.Lazy (ByteString) import Control.Monad (forever, void) import Control.Monad.IO.Class (liftIO, MonadIO)@@ -131,11 +129,11 @@ . ( MonadBaseControl IO m , MonadIO m )- => (Maybe (Word16, ByteString) -> m ()) -- ^ see 'Network.WebSockets.Simple.onClose'+ => (CloseOrigin -> ConnectionException -> m ()) -- ^ see 'Network.WebSockets.Simple.onClose' -> RPCServer sub sup rep com m- -> WebSocketsApp (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) (WebSocketServerRPCT sub sup m)+ -> WebSocketsApp (WebSocketServerRPCT sub sup m) (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) rpcServerSimple onClose f = WebSocketsApp- { onOpen = \send -> pure ()+ { onOpen = \_ -> pure () , onReceive = \WebSocketsAppParams{send} eSubSup -> do env <- getServerEnv @@ -153,8 +151,8 @@ in runWebSocketServerRPCT' env $ send (Right c) cont :: Either sub sup -> m ()- cont eSubSup = liftBaseWith $ \runInBase -> void $ Async.async $- void $ runInBase $ f RPCServerParams{reply,complete} eSubSup+ cont eSubSup' = liftBaseWith $ \runInBase -> void $ Async.async $+ void $ runInBase $ f RPCServerParams{reply,complete} eSubSup' registerSubscribeSupply _ident cont runSubscribeSupply _ident (Left _params)@@ -168,7 +166,7 @@ case eSubSup of Left sub -> runSub sub Right sup -> runSup sup- , onClose = lift . onClose+ , onClose = \o -> lift . onClose o } @@ -260,9 +258,9 @@ rpcClientSimple :: forall sub sup rep com m . ( MonadIO m )- => (Maybe (Word16, ByteString) -> m ()) -- ^ see 'Network.WebSockets.Simple.onClose'+ => (CloseOrigin -> ConnectionException -> m ()) -- ^ see 'Network.WebSockets.Simple.onClose' -> RPCClient sub sup rep com m- -> WebSocketClientRPCT rep com m (WebSocketsApp (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) (WebSocketClientRPCT rep com m))+ -> WebSocketClientRPCT rep com m (WebSocketsApp (WebSocketClientRPCT rep com m) (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup))) rpcClientSimple onClose RPCClient{subscription,onSubscribe,onReply,onComplete} = do _ident <- freshRPCID pure WebSocketsApp@@ -281,7 +279,7 @@ registerReplyComplete _ident (onReply RPCClientParams{supply,cancel}) onComplete - , onReceive = \WebSocketsAppParams{send} eRepCom -> do+ , onReceive = \_ eRepCom -> do let runRep :: Reply rep -> WebSocketClientRPCT rep com m () runRep (Reply RPCIdentified{_ident = _ident',_params}) | _ident' == _ident = runReply _ident _params@@ -297,7 +295,7 @@ case eRepCom of Left rep -> runRep rep Right com -> runCom com- , onClose = lift . onClose+ , onClose = \o -> lift . onClose o } @@ -335,15 +333,15 @@ runWebSocketClientRPCTSimple :: ( Monad m ) => (forall a. WebSocketClientRPCT rep com m a -> m a)- -> WebSocketsApp (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) (WebSocketClientRPCT rep com m)- -> WebSocketsApp (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) m+ -> WebSocketsApp (WebSocketClientRPCT rep com m) (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup))+ -> WebSocketsApp m (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) runWebSocketClientRPCTSimple runWS = hoistWebSocketsApp runWS lift execWebSocketClientRPCTSimple :: ( MonadBaseControl IO m )- => WebSocketsApp (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) (WebSocketClientRPCT rep com m)- -> m (WebSocketsApp (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) m)+ => WebSocketsApp (WebSocketClientRPCT rep com m) (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup))+ -> m (WebSocketsApp m (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup))) execWebSocketClientRPCTSimple x = do env <- liftBaseWith $ \_ -> Client.newEnv @@ -355,15 +353,15 @@ runWebSocketServerRPCTSimple :: ( Monad m ) => (forall a. WebSocketServerRPCT sub sup m a -> m a)- -> WebSocketsApp (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) (WebSocketServerRPCT sub sup m)- -> WebSocketsApp (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) m+ -> WebSocketsApp (WebSocketServerRPCT sub sup m) (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com))+ -> WebSocketsApp m (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com)) runWebSocketServerRPCTSimple runWS = hoistWebSocketsApp runWS lift execWebSocketServerRPCTSimple :: ( MonadBaseControl IO m )- => WebSocketsApp (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) (WebSocketServerRPCT sub sup m)- -> m (WebSocketsApp (Either (Reply rep) (Complete com)) (Either (Subscribe sub) (Supply sup)) m)+ => WebSocketsApp (WebSocketServerRPCT sub sup m) (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com))+ -> m (WebSocketsApp m (Either (Subscribe sub) (Supply sup)) (Either (Reply rep) (Complete com))) execWebSocketServerRPCTSimple x = do env <- liftBaseWith $ \_ -> Server.newEnv
src/Network/WebSockets/RPC/ACKable.hs view
@@ -20,14 +20,13 @@ import Data.Hashable (Hashable) import Data.Aeson (FromJSON (..), ToJSON (..), (.:), (.=), object) import Data.Aeson.Types (typeMismatch, Value (Object))-import Control.Applicative ((<|>)) import Control.Monad (when, forever) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Concurrent (threadDelay) import Control.Concurrent.Async (async) import qualified Control.Concurrent.Async as Async import Control.Concurrent.STM (atomically)-import Control.Concurrent.STM.TVar (newTVarIO, readTVarIO, readTVar, writeTVar, modifyTVar, TVar)+import Control.Concurrent.STM.TVar (newTVarIO, readTVarIO, readTVar, writeTVar, modifyTVar) data ACKable owner a = ACKable@@ -219,12 +218,6 @@ -second = 1000000-minute = 60 * second-hour = 60 * minute-day = 24 * hour-week = 7 * day- mkBackoff :: IO a -- ^ Invoked each attempt -> IO () -- ^ on quit -> IO (Async.Async a)@@ -234,9 +227,15 @@ toWait <- readIORef spentWaiting writeIORef spentWaiting (toWait + 1) let toWait' = 2 ^ toWait- soFar = sum $ (\x -> (2 ^ x) * second) <$> [0..toWait]+ soFar = sum $ (\y -> (2 ^ y) * second) <$> [0..toWait] when (soFar > week) x threadDelay $ second * (toWait' + 10) op+ where+ second = 1000000+ minute = 60 * second+ hour = 60 * minute+ day = 24 * hour+ week = 7 * day
src/Network/WebSockets/RPC/Trans/Client.hs view
@@ -72,7 +72,7 @@ runWebSocketClientRPCT' env (WebSocketClientRPCT (ReaderT f)) = f env getClientEnv :: Applicative m => WebSocketClientRPCT rep com m (Env rep com m)-getClientEnv = WebSocketClientRPCT (ReaderT (\env -> pure env))+getClientEnv = WebSocketClientRPCT (ReaderT pure) execWebSocketClientRPCT :: MonadIO m => WebSocketClientRPCT rep com m a -> m a execWebSocketClientRPCT f = do@@ -84,7 +84,7 @@ instance MonadReader r m => MonadReader r (WebSocketClientRPCT rep com m) where ask = WebSocketClientRPCT (ReaderT (const ask))- local f (WebSocketClientRPCT (ReaderT g)) = WebSocketClientRPCT (ReaderT (\env -> local f (g env)))+ local f (WebSocketClientRPCT (ReaderT g)) = WebSocketClientRPCT (ReaderT (local f . g)) freshRPCID :: MonadIO m => WebSocketClientRPCT rep com m RPCID freshRPCID =
websockets-rpc.cabal view
@@ -1,106 +1,94 @@-Name: websockets-rpc-Version: 0.6.0-Author: Athan Clark <athan.clark@gmail.com>-Maintainer: Athan Clark <athan.clark@gmail.com>-License: BSD3-License-File: LICENSE-Synopsis: Simple streaming RPC mechanism using WebSockets--- Description:-Category: Web-Cabal-Version: >= 1.10-Build-Type: Simple--Flag Example- Description: Build the example websocket RPC test- Default: False--Flag Example-Client- Description: Build the example websocket client RPC test- Default: False+-- This file has been generated from package.yaml by hpack version 0.21.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: c8257b8fab21297c6a0bdb9cec3a4038b0367a663393b3a746bf68f0b4893453 -Library- Default-Language: Haskell2010- HS-Source-Dirs: src- GHC-Options: -Wall- Exposed-Modules: Network.WebSockets.RPC- Network.WebSockets.RPC.Types- Network.WebSockets.RPC.ACKable- Network.WebSockets.RPC.Trans.Client- Network.WebSockets.RPC.Trans.Server- Build-Depends: base >= 4.8 && < 5- , aeson- , async- , bytestring- , containers- , exceptions- , hashable- , monad-control- , mtl- , QuickCheck- , stm- , text- , transformers- , unordered-containers- , uuid- , wai-transformers- , websockets >= 0.11- , websockets-simple >= 0.0.5+name: websockets-rpc+version: 0.7.0+synopsis: Simple streaming RPC mechanism using WebSockets+description: Please see the README on Github at <https://github.com/athanclark/sparrow-server#readme>+category: Web+homepage: https://github.com/athanclark/websockets-rpc#readme+bug-reports: https://github.com/athanclark/websockets-rpc/issues+author: Athan Clark+maintainer: athan.clark@gmail.com+copyright: BSD-3+license: BSD3+license-file: LICENSE+build-type: Simple+cabal-version: >= 1.10 -Test-suite test- Type: exitcode-stdio-1.0- Default-Language: Haskell2010- Hs-Source-Dirs: test- Ghc-Options: -Wall -threaded- Main-Is: Spec.hs- Other-Modules: Network.WebSockets.RPCSpec- Build-Depends: base- , websockets-rpc- , aeson- , QuickCheck- , quickcheck-instances- , tasty- , tasty-quickcheck+extra-source-files:+ README.md -Executable example- if flag(Example)- Buildable: True- else- Buildable: False- Default-Language: Haskell2010- Hs-Source-Dirs: example- Ghc-Options: -Wall -threaded- Main-Is: Main.hs- Build-Depends: base- , websockets-rpc- , aeson- , async- , exceptions- , MonadRandom- , mtl- , wai-transformers- , websockets- , websockets-simple+source-repository head+ type: git+ location: https://github.com/athanclark/websockets-rpc -Executable example-client- if flag(Example)- Buildable: True- else- Buildable: False- Default-Language: Haskell2010- Hs-Source-Dirs: example- Ghc-Options: -Wall -threaded- Main-Is: Main-Client.hs- Build-Depends: base- , websockets-rpc- , aeson- , async- , exceptions- , MonadRandom- , mtl- , wai-transformers- , websockets- , websockets-simple+library+ exposed-modules:+ Network.WebSockets.RPC+ Network.WebSockets.RPC.ACKable+ Network.WebSockets.RPC.Trans.Client+ Network.WebSockets.RPC.Trans.Server+ Network.WebSockets.RPC.Types+ other-modules:+ Paths_websockets_rpc+ hs-source-dirs:+ src+ ghc-options: -Wall+ build-depends:+ QuickCheck+ , aeson+ , async+ , base >=4.10 && <5.0+ , bytestring+ , containers+ , exceptions+ , hashable+ , monad-control+ , mtl+ , stm+ , text+ , transformers+ , unordered-containers+ , uuid+ , wai-transformers+ , websockets >=0.12+ , websockets-simple >=0.1.0+ default-language: Haskell2010 -Source-Repository head- Type: git- Location: https://github.com/athanclark/websockets-rpc+test-suite sparrow-server-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ Network.WebSockets.RPCSpec+ Paths_websockets_rpc+ hs-source-dirs:+ test+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ QuickCheck+ , aeson+ , async+ , base+ , bytestring+ , containers+ , exceptions+ , hashable+ , monad-control+ , mtl+ , quickcheck-instances+ , stm+ , tasty+ , tasty-quickcheck+ , text+ , transformers+ , unordered-containers+ , uuid+ , wai-transformers+ , websockets >=0.12+ , websockets-rpc+ , websockets-simple >=0.1.0+ default-language: Haskell2010