diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2017, Athan Clark + +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 Athan Clark 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. diff --git a/example/Main-Client.hs b/example/Main-Client.hs new file mode 100644 --- /dev/null +++ b/example/Main-Client.hs @@ -0,0 +1,77 @@ +{-# LANGUAGE + TemplateHaskell + , NamedFieldPuns + , ScopedTypeVariables + #-} + +module Main where + +import Network.WebSockets (runServer, runClient, ServerApp, ClientApp) +import Network.WebSockets.RPC +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.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) + + + + +myClient :: (MonadIO m, MonadThrow 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 + , 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 + } + + + + + +main :: IO () +main = do + let myClient' :: ClientApp () + myClient' = runClientAppT execWebSocketClientRPCT myClient + + threadDelay 1000000 + runClient "127.0.0.1" 8080 "" myClient' diff --git a/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -0,0 +1,71 @@ +{-# LANGUAGE + TemplateHaskell + , NamedFieldPuns + , ScopedTypeVariables + #-} + +module Main where + +import Network.WebSockets (runServer, runClient, ServerApp, ClientApp) +import Network.WebSockets.RPC +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.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) => ServerAppT (WebSocketServerRPCT MySubDSL MySupDSL m) +myServer = rpcServer $ \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 -> do + liftIO $ print Bar + liftIO $ putStrLn "Replying Baz..." + reply Baz + + + + +main :: IO () +main = do + let myServer' :: ServerApp + myServer' = runServerAppT execWebSocketServerRPCT myServer + + runServer "127.0.0.1" 8080 myServer' diff --git a/src/Network/WebSockets/RPC.hs b/src/Network/WebSockets/RPC.hs new file mode 100644 --- /dev/null +++ b/src/Network/WebSockets/RPC.hs @@ -0,0 +1,183 @@ +{-# LANGUAGE + RankNTypes + , ScopedTypeVariables + , NamedFieldPuns + #-} + +module Network.WebSockets.RPC + ( -- * Server + RPCServerParams (..), RPCServer, rpcServer + , -- * Client + RPCClientParams (..), RPCClient (..), rpcClient + , -- * Re-Exports + WebSocketServerRPCT, execWebSocketServerRPCT, WebSocketClientRPCT, execWebSocketClientRPCT + , WebSocketRPCException (..) + ) where + +import Network.WebSockets.RPC.Trans.Server ( WebSocketServerRPCT, runWebSocketServerRPCT', getServerEnv, execWebSocketServerRPCT + , registerSubscribeSupply, runSubscribeSupply, unregisterSubscribeSupply) +import Network.WebSockets.RPC.Trans.Client ( WebSocketClientRPCT, runWebSocketClientRPCT', getClientEnv, execWebSocketClientRPCT + , registerReplyComplete, runReply, runComplete, unregisterReplyComplete + , freshRPCID + ) +import Network.WebSockets.RPC.Types ( WebSocketRPCException (..), Subscribe (..), Supply (..), Reply (..), Complete (..) + , ClientToServer (Sub, Sup), ServerToClient (Rep, Com) + , RPCIdentified (..) + ) +import Network.WebSockets (acceptRequest, receiveDataMessage, sendDataMessage, DataMessage (Text, Binary)) +import Network.Wai.Trans (ServerAppT, ClientAppT) +import Data.Aeson (ToJSON, FromJSON, decode, encode) + +import Control.Monad (forever) +import Control.Monad.IO.Class (liftIO, MonadIO) +import Control.Monad.Catch (MonadThrow, throwM) + + +-- * Server + +data RPCServerParams rep com m = RPCServerParams + { reply :: rep -> m () -- ^ dispatch a reply + , complete :: com -> m () -- ^ dispatch a onComplete + } + +type RPCServer sub sup rep com m + = RPCServerParams rep com m + -> Either sub sup -- ^ handle incoming message + -> m () + +-- | May throw a 'WebSocketRPCParseFailure' if the opposing party sends bad data +rpcServer :: forall sub sup rep com m + . ( ToJSON rep + , ToJSON com + , FromJSON sub + , FromJSON sup + , MonadIO m + , MonadThrow m + ) + => RPCServer sub sup rep com m + -> ServerAppT (WebSocketServerRPCT sub sup m) +rpcServer f pendingConn = do + conn <- liftIO (acceptRequest pendingConn) + let runSub :: Subscribe sub -> WebSocketServerRPCT sub sup m () + runSub (Subscribe RPCIdentified{_ident,_params}) = do + env <- getServerEnv + + let reply :: rep -> m () + reply rep = + let r = Reply RPCIdentified{_ident, _params = rep} + in liftIO (sendDataMessage conn (Text (encode r))) + + complete :: com -> m () + complete com = + let c = Complete RPCIdentified{_ident, _params = com} + in do liftIO (sendDataMessage conn (Text (encode c))) + runWebSocketServerRPCT' env (unregisterSubscribeSupply _ident) + + cont :: Either sub sup -> m () + cont = f RPCServerParams{reply,complete} + + registerSubscribeSupply _ident cont + runSubscribeSupply _ident (Left _params) + + runSup :: Supply sup -> WebSocketServerRPCT sub sup m () + runSup (Supply RPCIdentified{_ident,_params}) = + case _params of + Nothing -> unregisterSubscribeSupply _ident -- FIXME this could bork the server if I `async` a routine thread + Just params -> runSubscribeSupply _ident (Right params) + + forever $ do + data' <- liftIO (receiveDataMessage conn) + case data' of + Text xs -> case decode xs of + Nothing -> + throwM (WebSocketRPCParseFailure ["server","text"] xs) + Just x -> case x of + Sub sub -> runSub sub + Sup sup -> runSup sup + Binary xs -> case decode xs of + Nothing -> + throwM (WebSocketRPCParseFailure ["server","binary"] xs) + Just x -> case x of + Sub sub -> runSub sub + Sup sup -> runSup sup + + + + +-- * Client + + +data RPCClientParams sup m = RPCClientParams + { supply :: sup -> m () -- ^ dispatch a supply + , cancel :: m () -- ^ cancel the RPC call + } + +data RPCClient sub sup rep com m = RPCClient + { subscription :: !sub + , onReply :: RPCClientParams sup m + -> rep + -> m () -- ^ handle incoming reply + , onComplete :: com -> m () -- ^ handle finalized onComplete + } + + +-- | May throw a 'WebSocketRPCParseFailure' if the opposing party sends bad data +rpcClient :: forall sub sup rep com m + . ( ToJSON sub + , ToJSON sup + , FromJSON rep + , FromJSON com + , MonadIO m + , MonadThrow m + ) + => ((RPCClient sub sup rep com m -> WebSocketClientRPCT rep com m ()) -> WebSocketClientRPCT rep com m ()) + -> ClientAppT (WebSocketClientRPCT rep com m) () +rpcClient userGo conn = + let go RPCClient{subscription,onReply,onComplete} = do + _ident <- freshRPCID + + liftIO (sendDataMessage conn (Text (encode (Subscribe RPCIdentified{_ident, _params = subscription})))) + + env <- getClientEnv + + let supply :: sup -> m () + supply sup = liftIO (sendDataMessage conn (Text (encode (Supply RPCIdentified{_ident, _params = Just sup})))) + + cancel :: m () + cancel = do + liftIO (sendDataMessage conn (Text (encode (Supply RPCIdentified{_ident, _params = Nothing :: Maybe ()})))) + runWebSocketClientRPCT' env (unregisterReplyComplete _ident) + + registerReplyComplete _ident (onReply RPCClientParams{supply,cancel}) onComplete + + let runRep :: Reply rep -> WebSocketClientRPCT rep com m () + runRep (Reply RPCIdentified{_ident = _ident',_params}) + | _ident' == _ident = runReply _ident _params + | otherwise = pure () -- FIXME fail somehow legibly?? + + runCom :: Complete com -> WebSocketClientRPCT rep com m () + runCom (Complete RPCIdentified{_ident = _ident', _params}) + | _ident' == _ident = do + runComplete _ident' _params -- NOTE don't use onComplete here because it might not exist later + unregisterReplyComplete _ident' + | otherwise = pure () + + forever $ do + data' <- liftIO (receiveDataMessage conn) + case data' of + Text xs -> + case decode xs of + Nothing -> + throwM (WebSocketRPCParseFailure ["client","text"] xs) + Just x -> case x of + Rep rep -> runRep rep + Com com -> runCom com + Binary xs -> + case decode xs of + Nothing -> + throwM (WebSocketRPCParseFailure ["client","binary"] xs) + Just x -> case x of + Rep rep -> runRep rep + Com com -> runCom com + + in userGo go diff --git a/src/Network/WebSockets/RPC/Trans/Client.hs b/src/Network/WebSockets/RPC/Trans/Client.hs new file mode 100644 --- /dev/null +++ b/src/Network/WebSockets/RPC/Trans/Client.hs @@ -0,0 +1,127 @@ +{-# LANGUAGE + NamedFieldPuns + , DeriveGeneric + , DeriveDataTypeable + , DeriveFunctor + , GeneralizedNewtypeDeriving + , FlexibleInstances + , MultiParamTypeClasses + , UndecidableInstances + #-} + +module Network.WebSockets.RPC.Trans.Client + ( WebSocketClientRPCT + , runWebSocketClientRPCT' + , getClientEnv + , execWebSocketClientRPCT + , -- * Utilities + freshRPCID + , registerReplyComplete + , unregisterReplyComplete + , runReply + , runComplete + ) where + +import GHC.Generics (Generic) +import Data.Data (Typeable) + +import Network.WebSockets.RPC.Types (RPCID, getRPCID) + +import Control.Concurrent.STM.TVar (newTVarIO, readTVar, readTVarIO, writeTVar, TVar) +import Control.Concurrent.STM (atomically) + +import Control.Monad.State.Class (MonadState) +import Control.Monad.Writer.Class (MonadWriter) +import Control.Monad.IO.Class (MonadIO (liftIO)) +import Control.Monad.Trans (MonadTrans (lift)) +import Control.Monad.Reader.Class (MonadReader (ask, local)) +import Control.Monad.Reader (ReaderT (ReaderT)) +import Control.Monad.Catch (MonadThrow, MonadCatch, MonadMask) + +import Data.IntMap.Lazy (IntMap) +import qualified Data.IntMap.Lazy as IntMap + + +data Conts rep com m = Conts + { reply :: rep -> m () + , complete :: com -> m () + } deriving (Generic, Typeable) + +data Env rep com m = Env + { rpcIdentVar :: TVar RPCID + , rpcContsVar :: TVar (IntMap (Conts rep com m)) + } + + +newEnv :: IO (Env rep com m) +newEnv = do + rpcIdentVar <- newTVarIO minBound + rpcContsVar <- newTVarIO mempty + pure Env{rpcIdentVar, rpcContsVar} + + +newtype WebSocketClientRPCT rep com m a = WebSocketClientRPCT + { runWebSocketClientRPCT :: ReaderT (Env rep com m) m a + } deriving ( Generic, Typeable, Functor, Applicative, Monad + , MonadState s, MonadWriter w, MonadIO, MonadThrow, MonadCatch, MonadMask + ) + +runWebSocketClientRPCT' :: Env rep com m -> WebSocketClientRPCT rep com m a -> m a +runWebSocketClientRPCT' env (WebSocketClientRPCT (ReaderT f)) = f env + +getClientEnv :: Applicative m => WebSocketClientRPCT rep com m (Env rep com m) +getClientEnv = WebSocketClientRPCT (ReaderT (\env -> pure env)) + +execWebSocketClientRPCT :: MonadIO m => WebSocketClientRPCT rep com m a -> m a +execWebSocketClientRPCT f = do + env <- liftIO newEnv + runWebSocketClientRPCT' env f + +instance MonadTrans (WebSocketClientRPCT rep com) where + lift x = WebSocketClientRPCT (ReaderT (const x)) + +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))) + +freshRPCID :: MonadIO m => WebSocketClientRPCT rep com m RPCID +freshRPCID = + let f Env{rpcIdentVar} = + let go = do rpcId <- readTVar rpcIdentVar + writeTVar rpcIdentVar (succ rpcId) + pure rpcId + in liftIO (atomically go) + in WebSocketClientRPCT (ReaderT f) + +registerReplyComplete :: MonadIO m => RPCID -> (rep -> m ()) -> (com -> m ()) -> WebSocketClientRPCT rep com m () +registerReplyComplete rpcId reply complete = + let f Env{rpcContsVar} = liftIO $ do + conts <- readTVarIO rpcContsVar + atomically (writeTVar rpcContsVar (IntMap.insert (getRPCID rpcId) Conts{reply,complete} conts)) + in WebSocketClientRPCT (ReaderT f) + +unregisterReplyComplete :: MonadIO m => RPCID -> WebSocketClientRPCT rep com m () +unregisterReplyComplete rpcId = + let f Env{rpcContsVar} = liftIO $ do + conts <- readTVarIO rpcContsVar + atomically (writeTVar rpcContsVar (IntMap.delete (getRPCID rpcId) conts)) + in WebSocketClientRPCT (ReaderT f) + + +runReply :: MonadIO m => RPCID -> rep -> WebSocketClientRPCT rep com m () +runReply rpcId rep = + let f Env{rpcContsVar} = do + conts <- liftIO (readTVarIO rpcContsVar) + case IntMap.lookup (getRPCID rpcId) conts of + Nothing -> pure () -- TODO throw notfound warning? + Just Conts{reply} -> reply rep + in WebSocketClientRPCT (ReaderT f) + +runComplete :: MonadIO m => RPCID -> com -> WebSocketClientRPCT rep com m () +runComplete rpcId rep = + let f Env{rpcContsVar} = do + conts <- liftIO (readTVarIO rpcContsVar) + case IntMap.lookup (getRPCID rpcId) conts of + Nothing -> pure () -- TODO throw notfound warning? + Just Conts{complete} -> complete rep + in WebSocketClientRPCT (ReaderT f) diff --git a/src/Network/WebSockets/RPC/Trans/Server.hs b/src/Network/WebSockets/RPC/Trans/Server.hs new file mode 100644 --- /dev/null +++ b/src/Network/WebSockets/RPC/Trans/Server.hs @@ -0,0 +1,103 @@ +{-# LANGUAGE + NamedFieldPuns + , DeriveGeneric + , DeriveDataTypeable + , DeriveFunctor + , GeneralizedNewtypeDeriving + , FlexibleInstances + , MultiParamTypeClasses + , UndecidableInstances + #-} + +module Network.WebSockets.RPC.Trans.Server + ( WebSocketServerRPCT + , runWebSocketServerRPCT' + , getServerEnv + , execWebSocketServerRPCT + , -- * Utilities + registerSubscribeSupply + , unregisterSubscribeSupply + , runSubscribeSupply + ) where + +import GHC.Generics (Generic) +import Data.Data (Typeable) + +import Network.WebSockets.RPC.Types (RPCID, getRPCID) + +import Control.Concurrent.STM.TVar (newTVarIO, readTVarIO, writeTVar, TVar) +import Control.Concurrent.STM (atomically) + +import Control.Monad.State.Class (MonadState) +import Control.Monad.Writer.Class (MonadWriter) +import Control.Monad.IO.Class (MonadIO (liftIO)) +import Control.Monad.Trans (MonadTrans (lift)) +import Control.Monad.Reader.Class (MonadReader (ask, local)) +import Control.Monad.Catch (MonadThrow, MonadCatch, MonadMask) + +import Control.Monad.Reader (ReaderT (ReaderT)) + +import Data.IntMap.Lazy (IntMap) +import qualified Data.IntMap.Lazy as IntMap + + +newtype Cont sub sup m = Cont + { getCont :: Either sub sup -> m () + } + +data Env sub sup m = Env + { rpcContsVar :: TVar (IntMap (Cont sub sup m)) + } + +newEnv :: IO (Env sub sup m) +newEnv = Env <$> newTVarIO mempty + + +newtype WebSocketServerRPCT sub sup m a = WebSocketServerRPCT + { runWebSocketServerRPCT :: ReaderT (Env sub sup m) m a + } deriving ( Generic, Typeable, Functor, Applicative, Monad + , MonadState s, MonadWriter w, MonadIO, MonadThrow, MonadCatch, MonadMask + ) + +runWebSocketServerRPCT' :: Env sub sup m -> WebSocketServerRPCT sub sup m a -> m a +runWebSocketServerRPCT' env (WebSocketServerRPCT (ReaderT f)) = f env + +getServerEnv :: Applicative m => WebSocketServerRPCT sub sup m (Env sub sup m) +getServerEnv = WebSocketServerRPCT (ReaderT (\env -> pure env)) + +execWebSocketServerRPCT :: MonadIO m => WebSocketServerRPCT sub sup m a -> m a +execWebSocketServerRPCT f = do + env <- liftIO newEnv + runWebSocketServerRPCT' env f + +instance MonadTrans (WebSocketServerRPCT sub sup) where + lift x = WebSocketServerRPCT (ReaderT (const x)) + +instance MonadReader r m => MonadReader r (WebSocketServerRPCT sub sup m) where + ask = WebSocketServerRPCT (ReaderT (const ask)) + local f (WebSocketServerRPCT (ReaderT g)) = WebSocketServerRPCT (ReaderT (\env -> local f (g env))) + + +registerSubscribeSupply :: MonadIO m => RPCID -> (Either sub sup -> m ()) -> WebSocketServerRPCT sub sup m () +registerSubscribeSupply rpcId getCont = + let f Env{rpcContsVar} = liftIO $ do + conts <- readTVarIO rpcContsVar + atomically (writeTVar rpcContsVar (IntMap.insert (getRPCID rpcId) Cont{getCont} conts)) + in WebSocketServerRPCT (ReaderT f) + +unregisterSubscribeSupply :: MonadIO m => RPCID -> WebSocketServerRPCT sub sup m () +unregisterSubscribeSupply rpcId = + let f Env{rpcContsVar} = liftIO $ do + conts <- readTVarIO rpcContsVar + atomically (writeTVar rpcContsVar (IntMap.delete (getRPCID rpcId) conts)) + in WebSocketServerRPCT (ReaderT f) + + +runSubscribeSupply :: MonadIO m => RPCID -> Either sub sup -> WebSocketServerRPCT sub sup m () +runSubscribeSupply rpcId x = + let f Env{rpcContsVar} = do + conts <- liftIO (readTVarIO rpcContsVar) + case IntMap.lookup (getRPCID rpcId) conts of + Nothing -> pure () -- TODO throw notfound warning? + Just Cont{getCont} -> getCont x + in WebSocketServerRPCT (ReaderT f) diff --git a/src/Network/WebSockets/RPC/Types.hs b/src/Network/WebSockets/RPC/Types.hs new file mode 100644 --- /dev/null +++ b/src/Network/WebSockets/RPC/Types.hs @@ -0,0 +1,168 @@ +{-# LANGUAGE + OverloadedStrings + , DeriveGeneric + , DeriveDataTypeable + , GeneralizedNewtypeDeriving + , NamedFieldPuns + #-} + +module Network.WebSockets.RPC.Types + ( RPCID, getRPCID + , -- * RPC Methods + RPCIdentified (..), Subscribe (Subscribe), Supply (..), Reply (Reply), Complete (Complete) + , -- ** Categorized + ClientToServer (..), ServerToClient (..) + , -- * Exceptions + WebSocketRPCException (..) + ) where + +import Data.Data (Data, Typeable) +import GHC.Generics (Generic) +import Data.Aeson (FromJSON (..), ToJSON (..), (.:), (.=), object) +import Data.Aeson.Types (typeMismatch, Value (Object, String)) +import qualified Data.HashMap.Lazy as HM +import Data.Text (Text) +import Data.ByteString.Lazy (ByteString) + +import Control.Applicative ((<|>)) +import Control.Monad.Catch (Exception) + +import Test.QuickCheck (Arbitrary (..), CoArbitrary) + + +-- | Unique identifier for an RPC session +newtype RPCID = RPCID {getRPCID :: Int} + deriving (Show, Read, Num, Eq, Ord, Enum, Bounded, Generic, Data, Typeable, FromJSON, ToJSON, Arbitrary, CoArbitrary) + + +data RPCIdentified a = RPCIdentified + { _ident :: {-# UNPACK #-} !RPCID + , _params :: !a + } deriving (Show, Read, Eq, Generic, Data, Typeable) + +instance ToJSON a => ToJSON (RPCIdentified a) where + toJSON RPCIdentified {_ident,_params} = object + [ "ident" .= _ident + , "params" .= _params + ] + +instance FromJSON a => FromJSON (RPCIdentified a) where + parseJSON (Object o) = RPCIdentified <$> o .: "ident" <*> o .: "params" + parseJSON x = typeMismatch "RPCIdentified" x + +instance Arbitrary a => Arbitrary (RPCIdentified a) where + arbitrary = RPCIdentified <$> arbitrary <*> arbitrary + shrink RPCIdentified{_ident,_params} = RPCIdentified <$> shrink _ident <*> shrink _params + + + +newtype Subscribe a = Subscribe {getSubscribe :: RPCIdentified a} + deriving (Show, Read, Eq, Generic, Data, Typeable, Arbitrary) + + +instance ToJSON a => ToJSON (Subscribe a) where + toJSON (Subscribe x) = case toJSON x of + Object xs -> Object (HM.insert "type" (String "sub") xs) + _ -> error "inconceivable!" + +instance FromJSON a => FromJSON (Subscribe a) where + parseJSON x@(Object o) = do + t <- o .: "type" + if t == ("sub" :: Text) + then Subscribe <$> parseJSON x + else fail "Not a subscription" + parseJSON x = typeMismatch "Subscribe" x + + +-- | @Nothing@ means the RPC is canceled +newtype Supply a = Supply { getSupply :: RPCIdentified (Maybe a) } + deriving (Show, Read, Eq, Generic, Data, Typeable, Arbitrary) + +instance ToJSON a => ToJSON (Supply a) where + toJSON Supply {getSupply} = case toJSON getSupply of + Object xs -> Object (HM.insert "type" (String "sup") xs) + _ -> error "inconceivable!" + +instance FromJSON a => FromJSON (Supply a) where + parseJSON x@(Object o) = do + t <- o .: "type" + if t == ("sup" :: Text) + then Supply <$> parseJSON x + else fail "Not a supply" + parseJSON x = typeMismatch "Supply" x + + +newtype Reply a = Reply {getReply :: RPCIdentified a} + deriving (Show, Read, Eq, Generic, Data, Typeable, Arbitrary) + +instance ToJSON a => ToJSON (Reply a) where + toJSON (Reply x) = case toJSON x of + Object xs -> Object (HM.insert "type" (String "rep") xs) + _ -> error "inconceivable!" + +instance FromJSON a => FromJSON (Reply a) where + parseJSON x@(Object o) = do + t <- o .: "type" + if t == ("rep" :: Text) + then Reply <$> parseJSON x + else fail "Not a reply" + parseJSON x = typeMismatch "Reply" x + + +newtype Complete a = Complete {getComplete :: RPCIdentified a} + deriving (Show, Read, Eq, Generic, Data, Typeable, Arbitrary) + +instance ToJSON a => ToJSON (Complete a) where + toJSON (Complete x) = case toJSON x of + Object xs -> Object (HM.insert "type" (String "com") xs) + _ -> error "inconceivable!" + +instance FromJSON a => FromJSON (Complete a) where + parseJSON x@(Object o) = do + t <- o .: "type" + if t == ("com" :: Text) + then Complete <$> parseJSON x + else fail "Not a complete" + parseJSON x = typeMismatch "Complete" x + + +data ClientToServer sub sup + = Sub (Subscribe sub) + | Sup (Supply sup) + deriving (Show, Read, Eq, Generic, Data, Typeable) + +instance (Arbitrary sub, Arbitrary sup) => Arbitrary (ClientToServer sub sup) where + arbitrary = do + q <- arbitrary + if q then Sub <$> arbitrary else Sup <$> arbitrary + +instance (ToJSON sub, ToJSON sup) => ToJSON (ClientToServer sub sup) where + toJSON (Sub x) = toJSON x + toJSON (Sup x) = toJSON x + +instance (FromJSON sub, FromJSON sup) => FromJSON (ClientToServer sub sup) where + parseJSON x = (Sub <$> parseJSON x) <|> (Sup <$> parseJSON x) + +data ServerToClient rep com + = Rep (Reply rep) + | Com (Complete com) + deriving (Show, Read, Eq, Generic, Data, Typeable) + +instance (Arbitrary sub, Arbitrary sup) => Arbitrary (ServerToClient sub sup) where + arbitrary = do + q <- arbitrary + if q then Rep <$> arbitrary else Com <$> arbitrary + +instance (ToJSON rep, ToJSON com) => ToJSON (ServerToClient rep com) where + toJSON (Rep x) = toJSON x + toJSON (Com x) = toJSON x + +instance (FromJSON rep, FromJSON com) => FromJSON (ServerToClient rep com) where + parseJSON x = (Rep <$> parseJSON x) <|> (Com <$> parseJSON x) + + +data WebSocketRPCException + = WebSocketRPCParseFailure [String] ByteString + deriving (Show, Generic) + +instance Exception WebSocketRPCException diff --git a/test/Network/WebSockets/RPCSpec.hs b/test/Network/WebSockets/RPCSpec.hs new file mode 100644 --- /dev/null +++ b/test/Network/WebSockets/RPCSpec.hs @@ -0,0 +1,37 @@ +{-# LANGUAGE + ScopedTypeVariables + #-} + +module Network.WebSockets.RPCSpec (spec) where + +import Network.WebSockets.RPC +import Network.WebSockets.RPC.Types (RPCIdentified, Subscribe, Supply, Reply, Complete, ClientToServer, ServerToClient) +import Data.Aeson (encode, decode, FromJSON, ToJSON) + +import Test.Tasty +import Test.Tasty.QuickCheck as QC +import Test.QuickCheck +import Test.QuickCheck.Instances + + +spec :: TestTree +spec = testGroup "Network.WebSockets.RPC" + [ QC.testProperty "RPCIdentified parse/print iso" + (\(x :: RPCIdentified ()) -> invariantJSON x) + , QC.testProperty "Subscribe parse/print iso" + (\(x :: Subscribe ()) -> invariantJSON x) + , QC.testProperty "Supply parse/print iso" + (\(x :: Supply ()) -> invariantJSON x) + , QC.testProperty "Reply parse/print iso" + (\(x :: Reply ()) -> invariantJSON x) + , QC.testProperty "Complete parse/print iso" + (\(x :: Complete ()) -> invariantJSON x) + , QC.testProperty "ClientToServer parse/print iso" + (\(x :: ClientToServer () ()) -> invariantJSON x) + , QC.testProperty "ServerToClient parse/print iso" + (\(x :: ServerToClient () ()) -> invariantJSON x) + ] + + +invariantJSON :: (FromJSON a, ToJSON a, Eq a, Show a) => a -> Property +invariantJSON x = decode (encode x) === Just x diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,13 @@ +module Main where + +import Network.WebSockets.RPCSpec + +import Test.Tasty + + +main :: IO () +main = defaultMain tests + +tests :: TestTree +tests = testGroup "Testing..." + [spec] diff --git a/websockets-rpc.cabal b/websockets-rpc.cabal new file mode 100644 --- /dev/null +++ b/websockets-rpc.cabal @@ -0,0 +1,97 @@ +Name: websockets-rpc +Version: 0.0.0 +Author: Athan Clark +Maintainer: Athan Clark +License: BSD3 +License-File: LICENSE +Synopsis: Simple streaming RPC mechanism using WebSockets +-- Description: +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 + +Library + Default-Language: Haskell2010 + HS-Source-Dirs: src + GHC-Options: -Wall + Exposed-Modules: Network.WebSockets.RPC + Network.WebSockets.RPC.Types + Network.WebSockets.RPC.Trans.Client + Network.WebSockets.RPC.Trans.Server + -- Other-Modules: Network.WebSockets.RPC.Internal + Build-Depends: base >= 4.8 && < 5 + , aeson + , bytestring + , containers + , exceptions + , mtl + , QuickCheck + , stm + , text + , unordered-containers + , wai-transformers + , websockets + +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 + +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 + +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 + +Source-Repository head + Type: git + Location: https://github.com/athanclark/websockets-rpc