remotion (empty) → 0.1.0
raw patch · 11 files changed
+1430/−0 lines, 11 filesdep +HTFdep +HUnitdep +QuickChecksetup-changed
Dependencies added: HTF, HUnit, QuickCheck, QuickCheck-GenT, async, base, bytestring, concurrent-extra, containers, directory, errors, filelock, hashable, hashtables, lifted-async, loch-th, monad-control, mtl, network, network-simple, old-locale, pipes, pipes-bytestring, pipes-cereal-plus, pipes-network, pipes-parse, placeholders, quickcheck-instances, remotion, stm, system-fileio, system-filepath, text, time, transformers-base
Files
- LICENSE +22/−0
- Setup.hs +2/−0
- htf-test-suite/HTFTestSuite.hs +8/−0
- library/Remotion/Client.hs +318/−0
- library/Remotion/Protocol.hs +81/−0
- library/Remotion/Server.hs +202/−0
- library/Remotion/Server/Connection.hs +122/−0
- library/Remotion/Session.hs +87/−0
- library/Remotion/Util/FileSystem.hs +147/−0
- library/Remotion/Util/Prelude.hs +226/−0
- remotion.cabal +215/−0
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2014, Nikita Volkov++Permission is hereby granted, free of charge, to any person+obtaining a copy of this software and associated documentation+files (the "Software"), to deal in the Software without+restriction, including without limitation the rights to use,+copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the+Software is furnished to do so, subject to the following+conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES+OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT+HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR+OTHER DEALINGS IN THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ htf-test-suite/HTFTestSuite.hs view
@@ -0,0 +1,8 @@+{-# OPTIONS_GHC -F -pgmF htfpp #-}++import Test.Framework+import HTFTestSuite.Prelude++import {-@ HTF_TESTS @-} HTFTestSuite.CommunicationTests++main = htfMain $ htf_thisModulesTests : htf_importedTests
+ library/Remotion/Client.hs view
@@ -0,0 +1,318 @@+{-# LANGUAGE CPP #-}+module Remotion.Client (+ -- * Control+ Client,+ run,+ request,+ -- * Settings+ Settings(..),+ P.UserProtocolSignature,+ URL(..),+ P.Credentials(..),+ -- * Failure+ Failure(..),+)+where+++import Remotion.Util.Prelude hiding (traceIO, traceIOWithTime, State, listen, interact)+import qualified Remotion.Util.Prelude as Prelude+import qualified Remotion.Session as S+import qualified Remotion.Protocol as P+import qualified Control.Concurrent.Async.Lifted as A+import qualified Control.Concurrent.Lock as Lock+import qualified Network+import qualified Remotion.Util.FileSystem as FS+++-- Debugging.+-------------------------+-- The following functions get enabled during debugging.++debugging = False+prefix = ("Client: " <>)+traceIO = if debugging + then Prelude.traceIO . prefix + else const $ return ()+traceIOWithTime = if debugging + then Prelude.traceIOWithTime . prefix + else const $ return ()++--------------------------------------------------------------------------------++-- |+-- A monad transformer for performing actions on client-side.+-- +-- Supports custom protocols with @i@ being the type of the client request and+-- @o@ - the server's response.+newtype Client i o m r = + Client { unClient :: ReaderT Env (EitherT Failure (S.Session m)) r }+ deriving (Functor, Applicative, Monad, MonadIO, MonadError Failure)++type Env = (KeepaliveState, KeepaliveTimeout, Lock)++type KeepaliveState = MVar (Maybe UTCTime)++type KeepaliveTimeout = Int++-- | Ensures a response to request accomodation in concurrency.+type Lock = Lock.Lock+++-- |+-- Settings of 'Client'.+type Settings = (P.UserProtocolSignature, URL)++-- |+-- Location of the server.+data URL =+ -- | Path to the socket-file.+ Socket FilePath |+ -- | Host name, port and credentials.+ Host Text Int P.Credentials++instance MonadTrans (Client i o) where+ lift = Client . lift . lift . lift++instance (MonadBase IO m) => MonadBase IO (Client i o m) where+ liftBase = Client . liftBase++instance MonadTransControl (Client i o) where+ newtype StT (Client i o) a = StT (StT S.Session (Either Failure a))+ liftWith runInM = do+ env <- Client $ ask+ Client $ lift $ lift $ liftWith $ \runSession -> runInM $ + liftM StT . runSession . runEitherT . flip runReaderT env . unClient+ restoreT m = do+ r <- Client $ lift $ lift $ do+ StT r <- lift m+ restoreT $ return $ r+ either throwError return r++instance (MonadBaseControl IO m) => MonadBaseControl IO (Client i o m) where+ newtype StM (Client i o m) a = StMT { unStMT :: ComposeSt (Client i o) m a }+ liftBaseWith = defaultLiftBaseWith StMT+ restoreM = defaultRestoreM unStMT++liftSession :: (Monad m) => S.Session m a -> Client i o m a+liftSession s = Client $ lift $ do+ r <- lift $ catchError (liftM Right $ s) (return . Left . adaptSessionFailure)+ hoistEither r+++-- |+-- Run 'Client' in the base monad.+-- +-- Requires the base monad to have a 'MonadBaseControl' instance for 'IO'.+run :: + forall i o m r.+ (Serializable IO i, Serializable IO o, MonadIO m, Applicative m,+ MonadBaseControl IO m) => + Settings -> Client i o m r -> m (Either Failure r)+run (userProtocolVersion, url) t = + runEitherT $ bracketME openSocket closeSocket $ \socket -> do+ timeout <- runHandshake socket+ lock <- liftIO $ Lock.new+ runInteraction socket timeout lock+ where+ openSocket = do+ traceIOWithTime "Opening socket"+ openURLSocketIO url |> try |> liftIO >>= \case+ Right r -> return r+ Left e -> case ioeGetErrorType e of+ NoSuchThing -> left $ UnreachableURL+ _ -> $bug $ "Unexpected IOException: " <> (packText . show) e+ closeSocket socket = do+ traceIOWithTime $ "Closing socket " <> show socket+ liftIO $ handle (const $ return () :: SomeException -> IO ()) $ hClose socket+ runHandshake socket =+ traceIOWithTime "Handshaking" >>+ S.run session settings >>= + hoistEither . fmapL adaptSessionFailure >>= + hoistEither . fmapL adaptHandshakeFailure+ where+ session = runEitherT $ do+ do+ receiveFailure+ do+ send P.version+ receiveFailure+ do+ send userProtocolVersion+ receiveFailure+ do+ send credentials+ receiveFailure+ do+ send (0::Int)+ receive+ where+ send = lift . S.send+ receive = lift S.receive+ receiveFailure = receive >>= maybe (return ()) left+ credentials = case url of+ Socket _ -> Nothing+ Host _ _ x -> x+ settings = (socket, 10^6*3)+ runInteraction socket timeout lock = do+ traceIOWithTime $ "Interacting"+ keepaliveState <- liftIO $ newMVar Nothing+ join $ fmap hoistEither $ lift $ runStack socket keepaliveState timeout lock $ do+ A.withAsync (finallyME (resetKeepalive *> t <* closeSession) stopKeepalive) $ \ta ->+ A.withAsync (keepaliveLoop) $ \ka -> do+ A.waitBoth ta ka >>= \(tr, kr) -> return tr++runStack :: + (MonadIO m) =>+ S.Socket -> KeepaliveState -> KeepaliveTimeout -> Lock -> Client i o m r -> m (Either Failure r)+runStack socket keepaliveState keepaliveTimeout lock t =+ if keepaliveTimeout < 10^3*100+ then error $ "Too small keepalive timeout setting: " <> show keepaliveTimeout+ else+ unClient t |>+ flip runReaderT (keepaliveState, keepaliveTimeout, lock) |>+ runEitherT |>+ flip S.run (socket, 10^6*30) |>+ liftM (join . fmapL adaptSessionFailure)++openURLSocketIO :: URL -> IO Handle+openURLSocketIO = \case+ Socket file -> +#if !defined(mingw32_HOST_OS) && !defined(cygwin32_HOST_OS) && !defined(_WIN32)+ Network.connectTo "" (Network.UnixSocket $ FS.encodeString file)+#else+ error "Socket used on Windows"+#endif+ Host name port _ -> + Network.connectTo (unpackText name) (Network.PortNumber $ fromIntegral port)++stopKeepalive :: (MonadIO m) => Client i o m ()+stopKeepalive = do+ traceIOWithTime "Stopping keepalive"+ (state, _, _) <- Client $ ask+ liftIO $ modifyMVar_ state $ const $ return Nothing++keepaliveLoop :: + (Applicative m, MonadIO m, Serializable IO o, Serializable IO i) => + Client i o m ()+keepaliveLoop = do+ (state, _, _) <- Client $ ask+ (liftIO $ readMVar state) >>= \case+ Nothing -> return ()+ Just nextTime -> do+ currentTime <- liftIO $ getCurrentTime+ when (currentTime >= nextTime) $ checkIn+ liftIO $ threadDelay $ 10^3 * 10+ keepaliveLoop++reduceTimeout :: Int -> Int+reduceTimeout = floor . (*10^6) . curve 1.2 1.3 . (/(10^6)) . fromIntegral+ where+ curve bending startingStraightness x = x / exp (bending / (x + startingStraightness))++resetKeepalive :: (MonadIO m) => Client i o m ()+resetKeepalive = do+ (state, timeout, _) <- Client $ ask+ liftIO $ do+ time <- getCurrentTime+ let nextTime = (microsToDiff $ toInteger $ reduceTimeout timeout) `addUTCTime` time+ modifyMVar_ state $ const $ return $ Just $ nextTime+ +interact :: + (Serializable IO o, Serializable IO i, MonadIO m, Applicative m) =>+ P.Request i -> Client i o m (Maybe o)+interact = \request -> do+ withLock $ send request >> receive >>= either (\f -> throwError $! adaptInteractionFailure f) return+ where+ withLock action = do+ (_, _, l) <- Client ask+ lock l+ finallyME action (unlock l)+ where+ lock = Client . liftIO . Lock.acquire+ unlock = Client . liftIO . Lock.release+ send r = + traceIOWithTime "Sending" *>+ (liftSession $ S.send r)+ receive = + traceIOWithTime "Receiving" *>+ liftSession S.receive++checkIn :: + (Serializable IO i, Serializable IO o, MonadIO m, Applicative m) => + Client i o m ()+checkIn = do + traceIOWithTime "Performing keepalive request"+ resetKeepalive+ interact P.Keepalive >>= maybe (return ()) ($bug "Unexpected response")++closeSession ::+ (Serializable IO i, Serializable IO o, MonadIO m, Applicative m) => + Client i o m ()+closeSession =+ traceIOWithTime "Closing session" >>+ interact P.CloseSession >>=+ maybe (return ()) ($bug "Unexpected response")++-- |+-- Send a request @i@ and receive a response @o@.+request :: + (Serializable IO i, Serializable IO o, MonadIO m, Applicative m) => + i -> Client i o m o+request a = do+ resetKeepalive+ interact (P.UserRequest a) >>= maybe ($bug "Unexpected response") return+++-- Failure+----------------------------++data Failure = + -- |+ -- Unable to connect to the provided url.+ UnreachableURL |+ -- |+ -- Server has too many connections already.+ -- It's suggested to retry later.+ ServerIsBusy |+ -- |+ -- Incorrect credentials.+ Unauthenticated |+ -- |+ -- Connection got interrupted for some reason.+ ConnectionInterrupted |+ -- |+ -- A timeout of communication with server reached.+ TimeoutReached Int |+ -- | + -- A mismatch of the internal protocol versions on client and server.+ -- First is the version on the client, second is the version on the server.+ ProtocolVersionMismatch Int Int |+ -- | + -- A mismatch of the user-supplied versions of custom protocol on client and server.+ -- First is the version on the client, second is the version on the server.+ UserProtocolSignatureMismatch ByteString ByteString |+ -- | + -- Server reports that it was unable to deserialize the request.+ -- This is only expected to happen in case of user's protocol mismatch.+ CorruptRequest Text+ deriving (Show, Read, Ord, Eq, Generic, Data, Typeable)++adaptHandshakeFailure :: P.HandshakeFailure -> Failure+adaptHandshakeFailure = \case+ P.ServerIsBusy -> ServerIsBusy+ P.ProtocolVersionMismatch c s -> ProtocolVersionMismatch c s+ P.UserProtocolSignatureMismatch c s -> UserProtocolSignatureMismatch c s+ P.Unauthenticated -> Unauthenticated++adaptInteractionFailure :: P.InteractionFailure -> Failure+adaptInteractionFailure = \case+ P.CorruptRequest t -> CorruptRequest t+ P.TimeoutReached t -> $bug $ "A connection keepalive timeout reached: " <> (packText . show) t++adaptSessionFailure :: S.Failure -> Failure+adaptSessionFailure = \case+ S.ConnectionInterrupted -> ConnectionInterrupted+ S.SendTimeoutReached t -> TimeoutReached t+ S.ReceiveTimeoutReached t -> TimeoutReached t+ S.CorruptData t -> $bug $ "Corrupt server response: " <> t
+ library/Remotion/Protocol.hs view
@@ -0,0 +1,81 @@+module Remotion.Protocol where++import Remotion.Util.Prelude+++-- Handshake+-----------------------------++-- |+-- A version of the internal protocol used for checking of server-client match.+type ProtocolVersion = Int++-- |+-- A unique identification of user's protocol version used for checking+-- of protocol versions mismatch between client and server.+-- It can be simply a user-supplied version number or +-- a hash or a serialization of the representation of a type used for protocol,+-- which can be generated using such library as +-- <http://hackage.haskell.org/package/type-structure type-structure>.+type UserProtocolSignature = ByteString++-- |+-- Either a plain ASCII password or an encoding of some data, +-- e.g. an MD5 hash of a username-password pair or just a password.+-- In more involved scenarios you can mix in serialization, +-- e.g. a serialized pair of username and a hash of just the password.+-- +-- @Nothing@ means anonymous.+type Credentials = Maybe ByteString++-- |+-- A session timeout in microseconds. +-- The period of keepalive signaling depends on that parameter.+-- If you don't want excessive requests, just make it a couple of minutes.+type Timeout = Int++data HandshakeFailure = + ServerIsBusy |+ -- | + -- A mismatch of the internal protocol versions on client and server.+ -- First is the version on the client, second is the version on the server.+ ProtocolVersionMismatch ProtocolVersion ProtocolVersion |+ UserProtocolSignatureMismatch UserProtocolSignature UserProtocolSignature |+ Unauthenticated+ deriving (Show, Generic)++instance Serializable m HandshakeFailure+++-- Interaction+-----------------------------++data Request a = + Keepalive | + CloseSession |+ UserRequest a+ deriving (Generic)++instance (Serializable m a) => Serializable m (Request a)++type Response a = Either InteractionFailure (Maybe a)++-- |+-- A failure response from server.+data InteractionFailure = + -- | + -- Server was unable to deserialize the request.+ -- This is only expected to happen in case of user's protocol mismatch.+ CorruptRequest Text | + -- |+ -- A connection keepalive timeout reached.+ TimeoutReached Int+ deriving (Show, Generic)++instance Serializable m InteractionFailure+++-----------------------------++version :: ProtocolVersion+version = 1
+ library/Remotion/Server.hs view
@@ -0,0 +1,202 @@+module Remotion.Server+ (+ -- * Control+ -- ** Monad-transformer+ Server,+ run,+ wait,+ countSlots,+ -- ** Simple+ runAndWait,+ -- * Settings+ Settings,+ P.UserProtocolSignature,+ ListeningMode(..),+ Port,+ C.Authenticate,+ P.Credentials,+ P.Timeout,+ MaxClients,+ Log,+ C.ProcessUserRequest,+ C.State,+ -- * Failure+ Failure(..),+ )+ where++import Remotion.Util.Prelude hiding (listen)+import qualified Remotion.Server.Connection as C+import qualified Remotion.Protocol as P+import qualified Remotion.Util.FileSystem as FS+import qualified Control.Concurrent.Async.Lifted as As+import qualified Network+import qualified Data.Set as Set++++-- | Settings of how to run the server.+type Settings i o s = + (P.UserProtocolSignature, ListeningMode, P.Timeout, MaxClients, Log, + C.ProcessUserRequest i o s)++-- | Defines how to listen for connections.+data ListeningMode =+ -- | + -- Listen on a port with an authentication function.+ Host Port C.Authenticate |+ -- | + -- Listen on a socket file.+ -- Since sockets are local no authentication is needed.+ -- Works only on UNIX systems.+ Socket FilePath++-- | A port to run the server on.+type Port = Int++-- | +-- A maximum amount of clients.+-- When this amount is reached the server rejects all the further connections.+type MaxClients = Int++-- |+-- A logging function.+-- If you want no logging, use @(const $ return ())@.+-- If you want to output to console use @Data.Text.IO.'Data.Text.IO.putStrLn'@.+-- If you want to somehow reformat the output, you're welcome: +-- @(Data.Text.IO.'Data.Text.IO.putStrLn' . (\"Remotion.Server: \" `<>`))@.+type Log = Text -> IO ()++--------------------------------------------------------------------------------+++-- API+------------------------++-- |+-- A monad transformer, which runs the server in the background.+newtype Server m a = + Server { unServer :: ReaderT (Wait, CountSlots) m a }+ deriving (Functor, Applicative, Monad, MonadIO, MonadTrans)++type Wait = IO ()+type CountSlots = IO Int++-- |+-- A Server failure.+data Failure =+ ListeningSocketIsBusy+ -- -- FIXME: implement the following+ -- -- | An IO exception has been thrown while accepting a connection socket.+ -- ConnectionSocketFailure IOException+ deriving (Show, Eq, Generic, Typeable)++-- |+-- Run the server, while automatically managing all related resources.+run :: + (Serializable IO i, Serializable IO o, MonadIO m) => + Settings i o s -> Server m a -> m (Either Failure a)+run (userVersion, listeningMode, timeout, maxClients, log, processRequest) m = runEitherT $ do++ let (portID, auth) = case listeningMode of+ Host port auth -> (Network.PortNumber $ fromIntegral port, auth)+ Socket path -> (Network.UnixSocket $ FS.encodeString path, const $ pure True)++ listeningSocket <- Network.listenOn portID |> try |> liftIO >>= \case+ Left e -> case ioeGetErrorType e of+ ResourceBusy -> left ListeningSocketIsBusy+ _ -> $bug $ "Unexpected IO error: " <> (packText . show) e+ Right r -> return r++ slotsVar <- liftIO $ newMVar maxClients++ activeListenerLock <- liftIO $ newMVar ()++ mainThreadID <- liftIO $ myThreadId++ liftIO $ log "Listening"++ -- Spawn all workers+ listenerAsyncs <- liftIO $ forM [1..(maxClients + 1)] $ \i -> + let + log' = log . (("Listener " <> packText (show i) <> ": ") <>)+ acquire = do+ (connectionSocket, _, _) <- do+ takeMVar activeListenerLock+ log' $ "Waiting for connection"+ Network.accept listeningSocket <* putMVar activeListenerLock ()+ modifyMVar_ slotsVar $ return . pred+ return connectionSocket+ release connectionSocket = do + log' "Releasing session's resources"+ hClose connectionSocket+ modifyMVar_ slotsVar $ return . succ+ process connectionSocket = do+ log' "Running client session"+ slots <- readMVar slotsVar+ C.runConnection connectionSocket (slots >= 0) auth timeout userVersion processRequest >>=+ either + (log' . ("Session failed: " <>) . packText . show) + (const $ log' "Session closed")+ in As.async $ forever $ do+ s <- acquire+ r <- try $ process s+ release s+ case r of+ Right r -> return r+ Left se -> if+ | Just ThreadKilled <- fromException se -> throwIO ThreadKilled+ | otherwise -> throwTo mainThreadID se+ let+ wait = do+ void $ As.waitAnyCancel listenerAsyncs+ stop = do+ log $ "Stopping server"+ forM_ listenerAsyncs As.cancel+ Network.sClose listeningSocket+ case listeningMode of+ Socket path -> FS.removeFile path+ _ -> return ()+ log $ "Stopped server"+ countSlots = readMVar slotsVar++ r <- lift $ runReaderT (unServer m) (wait, countSlots) + liftIO stop+ return r++-- | Block until the server stops (which should never happen).+wait :: (MonadIO m) => Server m ()+wait = Server $ ask >>= \(x, _) -> liftIO $ x++-- | Count the currently available slots for new connections.+countSlots :: (MonadIO m) => Server m Int+countSlots = Server $ ask >>= \(_, x) -> liftIO $ x++-- |+-- Run the server, while blocking the calling thread.+runAndWait :: (Serializable IO i, Serializable IO o) => Settings i o s -> IO (Either Failure ())+runAndWait settings = run settings $ wait+++-- "monad-control" instances+-------------------------++instance MonadBase IO m => MonadBase IO (Server m) where+ liftBase = Server . liftBase++instance MonadTransControl Server where+ newtype StT Server a = StT { unStT :: a }+ liftWith runToM = do+ wait <- Server $ ask+ Server $ lift $ runToM $ liftM StT . flip runReaderT wait . unServer+ restoreT m = do+ StT r <- Server $ lift $ m+ return r+ +instance (MonadBaseControl IO m) => MonadBaseControl IO (Server m) where+ newtype StM (Server m) a = StMT { unStMT :: ComposeSt Server m a }+ liftBaseWith = defaultLiftBaseWith StMT+ restoreM = defaultRestoreM unStMT+++
+ library/Remotion/Server/Connection.hs view
@@ -0,0 +1,122 @@+module Remotion.Server.Connection where++import Remotion.Util.Prelude hiding (State, listen, interact)+import qualified Remotion.Protocol as P+import qualified Remotion.Session as S+++runConnection :: + (MonadIO m, Applicative m, Serializable IO i, Serializable IO o) =>+ S.Socket ->+ ServerIsAvailable ->+ Authenticate ->+ P.Timeout ->+ P.UserProtocolSignature ->+ ProcessUserRequest i o s -> + m (Either ConnectionFailure ())+runConnection socket available authenticate timeout userVersion processRequest = runEitherT $ do+ do + r <- lift $ S.run (handshake available authenticate timeout userVersion) (socket, 10^6*3) + hoistEither $ join . liftM (fmapL HandshakeFailure) $ fmapL SessionFailure r+ do+ r <- lift $ S.run (interact processRequest) (socket, timeout)+ hoistEither $ fmapL SessionFailure r++data ConnectionFailure = + HandshakeFailure P.HandshakeFailure |+ SessionFailure S.Failure+ deriving (Show)+++-- Handshake+-----------------------------++-- | +-- A function, which checks the authentication data.+-- If you want to provide access to anybody, use @(const $ return True)@.+type Authenticate = P.Credentials -> IO Bool++-- |+-- +type ServerIsAvailable = Bool++handshake ::+ (MonadIO m, Applicative m) =>+ ServerIsAvailable ->+ Authenticate ->+ P.Timeout ->+ P.UserProtocolSignature ->+ S.Session m (Either P.HandshakeFailure ())+handshake available authenticate timeout userVersion = runEitherT $ do+ do+ check (not available) $ P.ServerIsBusy+ do+ cv <- receive+ check (cv /= P.version) $ P.ProtocolVersionMismatch cv P.version+ do+ cv <- receive+ check (cv /= userVersion) $ P.UserProtocolSignatureMismatch cv userVersion+ do+ credentials <- receive+ ok <- liftIO $ authenticate $ credentials+ check (not ok) $ P.Unauthenticated+ do+ 0::Int <- receive -- A workaround for otherwise unpredictable behaviour,+ -- happening in case of multiple sends.+ send $ timeout+ where+ receive = lift $ S.receive+ send = lift . S.send+ check condition failure = do+ let failureM = if condition then Just $ failure else Nothing+ send failureM+ maybe (return ()) left failureM+++-- Interaction+-----------------------------++-- | +-- A function which processes requests of type @i@ from client and +-- produces a response of type @o@,+-- while maintaining a user-defined session state of type @s@ per each client.+-- +-- This function essentially is what defines what the server actually does.+type ProcessUserRequest i o s = State s -> i -> IO o++-- |+-- A mutable state associated with particular client's connection.+-- Since we're in `IO` anyway, we use a mutable state with `IORef` wrapper.+-- You're free to extend it with whatever the data structure you want.+type State s = IORef (Maybe s)++interact :: + forall i o s m. + (MonadIO m, Serializable IO i, Serializable IO o, Applicative m) =>+ ProcessUserRequest i o s -> S.Session m ()+interact processRequest = do+ state <- liftIO $ newIORef Nothing+ let + loop = do+ i <- catchError receive $ \e -> do+ case e of+ S.ReceiveTimeoutReached t -> send $ Left $ P.TimeoutReached t+ S.SendTimeoutReached t -> send $ Left $ P.TimeoutReached t+ S.CorruptData t -> send $ Left $ P.CorruptRequest t+ _ -> return ()+ throwError e+ case i of+ P.CloseSession -> do+ send $ Right $ Nothing+ P.Keepalive -> do+ send $ Right $ Nothing+ loop+ P.UserRequest a -> do+ o <- liftIO $ processRequest state a+ send $ Right $ Just o+ loop+ loop+ where+ receive = S.receive :: S.Session m (P.Request i)+ send = S.send :: P.Response o -> S.Session m ()+
+ library/Remotion/Session.hs view
@@ -0,0 +1,87 @@+module Remotion.Session where++import Remotion.Util.Prelude+import qualified Network.Socket+import qualified Pipes.ByteString as PipesByteString+import qualified Pipes.Prelude as PipesPrelude+import qualified System.Timeout as Timeout+import qualified Control.Exception as Ex+++-- | +-- An abstraction over networking and data transmission.+-- Can be used in implementation of both the server and the client.+newtype Session m r = + Session { unSession :: ReaderT Settings (EitherT Failure m) r }+ deriving (Functor, Applicative, Monad, MonadIO, MonadReader Settings, MonadError Failure)++type Settings = (Socket, Timeout)++type Socket = Handle++-- |+-- A connection timeout in microseconds.+-- The period of keepalive signaling depends on that parameter.+-- If you don't want excessive requests, just make it a couple of minutes.+type Timeout = Int++data Failure =+ ConnectionInterrupted |+ ReceiveTimeoutReached Int |+ SendTimeoutReached Int |+ CorruptData Text+ deriving (Show)+++run :: Session m r -> Settings -> m (Either Failure r)+run (Session t) settings = runReaderT t settings |> runEitherT++adaptIOException :: IOException -> Failure+adaptIOException e = ioeGetErrorType e |> \case+ ResourceVanished -> ConnectionInterrupted+ _ -> $bug $ "Unexpected IOError: " <> (packText . show) e++adaptException :: SomeException -> Failure+adaptException e = if+ | Just ioe <- fromException e -> adaptIOException ioe+ | otherwise -> $bug $ "Unexpected exception: " <> packText (show e)++receive :: (Serializable IO i, MonadIO m) => Session m i+receive = Session $ do+ (handle, timeout) <- ask+ let pipe = PipesByteString.fromHandle handle >-> deserializingPipe+ pipe |> PipesPrelude.head |> runEitherT |> Ex.try |> Timeout.timeout timeout |> liftIO >>= \case+ Just (Right (Right (Just r))) -> return r+ Just (Right (Right Nothing)) -> throwError $ ConnectionInterrupted+ Just (Right (Left t)) -> throwError $ CorruptData t+ Just (Left e) -> throwError $ adaptIOException e+ Nothing -> throwError $ ReceiveTimeoutReached timeout+ +send :: (Serializable IO o, MonadIO m, Applicative m) => o -> Session m ()+send a = Session $ do+ (handle, timeout) <- ask+ let pipe = serializingProducer a >-> PipesByteString.toHandle handle+ lift $ do+ tr <- fmapLT adaptIOException $ tryIO $ Timeout.timeout timeout $ runEffect $ pipe+ failWith (SendTimeoutReached timeout) tr+++instance MonadTrans Session where+ lift = Session . lift . lift++instance (MonadBase IO m) => MonadBase IO (Session m) where+ liftBase = Session . liftBase++instance MonadTransControl Session where+ newtype StT Session a = StT { unStT :: Either Failure a }+ liftWith runToBase = do+ settings <- Session $ ask+ Session $ lift $ lift $ runToBase $ liftM StT . flip run settings+ restoreT base = do+ StT r <- Session $ lift $ lift $ base+ Session $ lift $ hoistEither r++instance (MonadBaseControl IO m) => MonadBaseControl IO (Session m) where+ newtype StM (Session m) a = StMT { unStMT :: ComposeSt Session m a }+ liftBaseWith = defaultLiftBaseWith StMT+ restoreM = defaultRestoreM unStMT
+ library/Remotion/Util/FileSystem.hs view
@@ -0,0 +1,147 @@+-- |+-- Utilities for dealing with 'FilePath'.+-- +module Remotion.Util.FileSystem+ ( + module Filesystem,+ module Filesystem.Path.CurrentOS,+ Status(..),+ getStatus,+ getExists,+ getTemporaryDirectory,+ remove,+ removeIfExists,+ removeTreeIfExists,+ move,+ copy,+ resolve,+ Lock,+ withLock,+ acquireLock,+ releaseLock,+ listFilesByExtension,+ )+ where++import Remotion.Util.Prelude hiding (stripPrefix, last)+import Filesystem.Path.CurrentOS+import Filesystem+import qualified System.Directory as Directory+import qualified Data.List as List+import qualified System.IO.Error as IOError+import qualified System.FileLock as Lock++++data Status = File | Directory | NotExists+ deriving (Show, Eq, Ord, Enum)++getStatus :: FilePath -> IO Status+getStatus path = do+ z <- isFile path+ if z+ then return File+ else do+ z <- isDirectory path+ if z+ then return Directory+ else return NotExists++getExists :: FilePath -> IO Bool+getExists path = getStatus path >>= return . (/= NotExists)++getTemporaryDirectory :: IO FilePath+getTemporaryDirectory = + Directory.getTemporaryDirectory >>= return . decodeString++remove :: FilePath -> IO ()+remove path = do+ status <- getStatus path+ case status of+ File -> removeFile path+ Directory -> removeTree path+ NotExists -> IOError.ioError $ IOError.mkIOError IOError.doesNotExistErrorType "" Nothing (Just $ encodeString path)++removeIfExists :: FilePath -> IO ()+removeIfExists path = do+ status <- getStatus path+ case status of+ File -> removeFile path+ Directory -> removeTree path+ NotExists -> return ()++removeTreeIfExists :: FilePath -> IO ()+removeTreeIfExists path = removeTree path `catch` \e -> case e of+ _ | IOError.isDoesNotExistError e -> return ()+ | otherwise -> throwIO e++move :: FilePath -> FilePath -> IO ()+move from to = do+ copy from to+ remove from++copy :: FilePath -> FilePath -> IO ()+copy from to = do+ isDir <- isDirectory from+ if isDir+ then do+ createTree to+ copyDirectory from to+ else do+ createTree $ directory to+ copyFile from to++copyDirectory :: FilePath -> FilePath -> IO ()+copyDirectory path path' = do+ members <- listDirectory path+ let members' = do+ member <- members+ let relative = + fromMaybe (error "Unexpectedly empty member path") $+ last member+ return $ path' <> relative+ sequence_ $ zipWith copy members members'++last :: FilePath -> Maybe FilePath+last p = case splitDirectories p of+ [] -> Nothing+ l -> Just $ List.last l++resolve :: FilePath -> IO FilePath+resolve path = case splitDirectories path of+ h:t | h == "~" -> do+ home <- getHomeDirectory+ return $ mconcat $ home : t+ _ -> return path++++type Lock = Lock.FileLock++-- | +-- Execute an IO action while using a file as an interprocess lock, +-- thus ensuring that only a single action executes across all processes +-- of the running system.+-- +-- If a file exists already it checks whether it is locked on by any running process, +-- including the current one,+-- and acquires it if it's not.+-- +-- Releases the lock in case of any failure in executed action or when the action is executed.+withLock :: FilePath -> IO a -> IO a+withLock file io = bracket (acquireLock file) releaseLock (const io)++acquireLock :: FilePath -> IO Lock+acquireLock path = + Lock.tryLockFile (encodeString path) Lock.Exclusive >>= \case+ Just lock -> return lock+ _ -> error $ "Lock `" ++ show path ++ "` is already in use"++releaseLock :: Lock -> IO ()+releaseLock = Lock.unlockFile++listFilesByExtension :: FilePath -> Text -> IO [FilePath]+listFilesByExtension dir extension = + listDirectory dir >>=+ return . filter (flip hasExtension extension)+
+ library/Remotion/Util/Prelude.hs view
@@ -0,0 +1,226 @@+module Remotion.Util.Prelude + ( + module Exports,++ LazyByteString,+ LazyText,++ (?:),+ traceM,+ traceIO,+ traceIOWithTime,+ packText,+ unpackText,+ bug,+ (|>),+ (<|),+ (|$>),+ microsToDiff,+ diffToMicros,+ bracketME,+ finallyME,+ tracingExceptions,+ )+ where++-------------+-- Control+-------------++-- base+import Prelude as Exports hiding (concat, foldr, mapM_, sequence_, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, mapM, sequence, FilePath, id, (.))+import Control.Monad as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Applicative as Exports+import Control.Arrow as Exports hiding (left, right)+import Control.Category as Exports+import Data.Monoid as Exports hiding (Any)+import Data.Foldable as Exports+import Data.Traversable as Exports hiding (for)+import Data.Maybe as Exports+import Data.List as Exports hiding (concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl')+import Data.Tuple as Exports+import Data.Ord as Exports (Down(..))+import Data.String as Exports+import Data.Int as Exports+import Data.Word as Exports+import Data.Ratio as Exports+import Data.Fixed as Exports+import Data.Ix as Exports+import Data.Data as Exports+import Text.Read as Exports (readMaybe, readEither)+import Control.Exception as Exports hiding (tryJust, assert)+import System.Mem.StableName as Exports+import System.Exit as Exports+import System.IO.Unsafe as Exports+import System.IO as Exports (Handle, hClose)+import System.IO.Error as Exports+import Unsafe.Coerce as Exports+import GHC.Exts as Exports hiding (traceEvent)+import GHC.Generics as Exports (Generic)+import GHC.IO.Exception as Exports+import Debug.Trace as Exports hiding (traceIO)+import Data.IORef as Exports+import Data.STRef as Exports+import Control.Monad.ST as Exports++-- mtl+import Control.Monad.Identity as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.State as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.Reader as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)+import Control.Monad.Writer as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM, Any)+import Control.Monad.Trans as Exports+import Control.Monad.Error as Exports hiding (mapM_, sequence_, forM_, msum, mapM, sequence, forM)++-- transformers-base+import Control.Monad.Base as Exports++-- monad-control+import Control.Monad.Trans.Control as Exports++-- errors+import Control.Error as Exports hiding ((?:))++-- placeholders+import Development.Placeholders as Exports++-- loch-th+import Debug.Trace.LocationTH as Exports++-------------+-- Data+-------------++-- time+import Data.Time.Clock as Exports++-- bytestring+import Data.ByteString as Exports (ByteString)++-- text+import Data.Text as Exports (Text)++-- containers+import Data.Map as Exports (Map)+import Data.IntMap as Exports (IntMap)+import Data.Set as Exports (Set)+import Data.IntSet as Exports (IntSet)+import Data.Sequence as Exports (Seq)+import Data.Tree as Exports (Tree)++-- hashable+import Data.Hashable as Exports (Hashable(..), hash)++-------------+-- Concurrency+-------------++-- base+import Control.Concurrent as Exports hiding (yield)++-- stm+import Control.Concurrent.STM as Exports hiding (check)++-------------+-- File-system+-------------++-- system-filepath+import Filesystem.Path as Exports (FilePath)++-------------+-- Streaming+-------------++-- pipes+import Pipes as Exports++-- pipes-cereal-plus+import PipesCerealPlus as Exports+++import qualified Data.ByteString.Lazy+import qualified Data.Text.Lazy+import qualified Data.Text+import qualified Prelude+import qualified Debug.Trace+import qualified System.Locale+import qualified Data.Time+++type LazyByteString = Data.ByteString.Lazy.ByteString+type LazyText = Data.Text.Lazy.Text+++(?:) :: Maybe a -> a -> a+maybeA ?: b = fromMaybe b maybeA+{-# INLINE (?:) #-}++traceM :: (Monad m) => String -> m ()+traceM s = trace s $ return ()++traceIO :: (MonadIO m) => String -> m ()+traceIO = liftIO . Debug.Trace.traceIO++traceIOWithTime :: (MonadIO m) => String -> m ()+traceIOWithTime s = do+ time <- liftIO $ getCurrentTime+ traceIO $ + formatTime time <> ": " <> s+ where+ formatTime = + take 15 . + Data.Time.formatTime System.Locale.defaultTimeLocale "%X.%q"++packText = Data.Text.pack+unpackText = Data.Text.unpack++bug = [e| $failure . (msg <>) . Data.Text.unpack |]+ where+ msg = "A \"remotion\" package bug: " :: String++(|>) :: a -> (a -> b) -> b+a |> aToB = aToB a+{-# INLINE (|>) #-}++(<|) :: (a -> b) -> a -> b+aToB <| a = aToB a+{-# INLINE (<|) #-}++-- | +-- The following are all the same:+-- fmap f a == f <$> a == a |> fmap f == a |$> f+-- +-- This operator accomodates the left-to-right operators: >>=, >>>, |>.+(|$>) = flip fmap+{-# INLINE (|$>) #-}++microsToDiff :: Fractional c => Integer -> c+microsToDiff = fromRational . (%(10^6))++diffToMicros :: (Real a, Integral b) => a -> b+diffToMicros = round . (*(10^6)) . toRational++bracketME :: (MonadError e m) => m a -> (a -> m b) -> (a -> m c) -> m c+bracketME acquire release apply = do+ r <- acquire+ z <- catchError (liftM Right $ apply r) (return . Left)+ release r+ either throwError return z++finallyME :: (MonadError e m) => m a -> m b -> m a+finallyME m f = do+ z <- catchError (liftM Right $ m) (return . Left)+ f+ either throwError return z++tracingExceptions :: MonadBaseControl IO m => m a -> m a+tracingExceptions m = + control $ \runInIO -> catch (runInIO m) $ \(SomeException e) -> runInIO $ do+ let rep = typeOf e+ tyCon = typeRepTyCon rep+ traceM $ + "## Uncaught exception: " ++ show e ++ "\n" +++ " Type: " ++ show rep ++ "\n" +++ " Module: " ++ tyConModule tyCon ++ "\n" +++ " Package: " ++ tyConPackage tyCon+ liftBase $ throwIO $ e
+ remotion.cabal view
@@ -0,0 +1,215 @@+name:+ remotion+version:+ 0.1.0+synopsis:+ A library for client-server applications based on custom protocols+description:+ An API abstracting over the typical tasks of client-server communication.+ It automates the authentication process, failure management and + the task of keeping the connections alive.+ It allows the user to implement protocols of any form.++ Useful for writing all kinds of services. + E.g., a <http://hackage.haskell.org/package/graph-db graph-db> + networking interface is based on this library.+category:+ Network, Service, Protocol+homepage:+ https://github.com/nikita-volkov/remotion +bug-reports:+ https://github.com/nikita-volkov/remotion/issues +author:+ Nikita Volkov <nikita.y.volkov@mail.ru>+maintainer:+ Nikita Volkov <nikita.y.volkov@mail.ru>+copyright:+ (c) 2014, Nikita Volkov+license:+ MIT+license-file:+ LICENSE+build-type:+ Simple+cabal-version:+ >=1.10+++source-repository head+ type:+ git+ location:+ git://github.com/nikita-volkov/remotion.git+++library+ hs-source-dirs:+ library+ other-modules:+ Remotion.Protocol+ Remotion.Session+ Remotion.Server.Connection+ Remotion.Util.FileSystem+ Remotion.Util.Prelude+ exposed-modules:+ Remotion.Server+ Remotion.Client+ build-depends:+ -- streaming:+ pipes >= 4.0 && < 4.2,+ pipes-bytestring == 2.0.*,+ pipes-cereal-plus >= 0.3.2 && < 0.4,+ pipes-parse > 3.0.0,+ -- networking:+ network == 2.4.*,+ network-simple == 0.3.*,+ pipes-network == 0.6.*,+ -- file-system:+ filelock == 0.1.*,+ directory,+ system-fileio,+ system-filepath,+ -- concurrency:+ lifted-async == 0.1.*,+ async == 2.0.*,+ concurrent-extra == 0.7.*,+ stm,+ -- data:+ hashtables == 1.1.*,+ time,+ old-locale,+ containers,+ hashable,+ text,+ bytestring,+ -- control:+ transformers-base == 0.4.*,+ monad-control == 0.3.*,+ loch-th == 0.2.*,+ placeholders == 0.1.*,+ errors >= 1.4.4,+ mtl,+ base >= 4.5 && < 5+ default-extensions:+ Arrows+ BangPatterns+ ConstraintKinds+ DataKinds+ DefaultSignatures+ DeriveDataTypeable+ DeriveGeneric+ EmptyDataDecls+ FlexibleContexts+ FlexibleInstances+ FunctionalDependencies+ GADTs+ GeneralizedNewtypeDeriving+ ImpredicativeTypes+ LambdaCase+ LiberalTypeSynonyms+ MultiParamTypeClasses+ MultiWayIf+ NoImplicitPrelude+ NoMonomorphismRestriction+ OverloadedStrings+ PatternGuards+ QuasiQuotes+ RankNTypes+ RecordWildCards+ ScopedTypeVariables+ StandaloneDeriving+ TemplateHaskell+ TupleSections+ TypeFamilies+ TypeOperators+ default-language:+ Haskell2010+++test-suite remotion-htf-test-suite+ ghc-options:+ -threaded+ "-with-rtsopts=-N"+ type: + exitcode-stdio-1.0+ hs-source-dirs: + htf-test-suite+ main-is: + HTFTestSuite.hs+ build-depends:+ remotion,+ -- testing:+ quickcheck-instances,+ QuickCheck-GenT == 0.1.*,+ QuickCheck,+ HUnit,+ HTF == 0.11.*,+ -- streaming:+ pipes >= 4.0 && < 4.2,+ pipes-bytestring == 2.0.*,+ pipes-cereal-plus >= 0.3.2 && < 0.4,+ pipes-parse > 3.0.0,+ -- networking:+ network == 2.4.*,+ network-simple == 0.3.*,+ pipes-network == 0.6.*,+ -- file-system:+ filelock == 0.1.*,+ directory,+ system-fileio,+ system-filepath,+ -- concurrency:+ lifted-async == 0.1.*,+ async == 2.0.*,+ concurrent-extra == 0.7.*,+ stm,+ -- data:+ hashtables == 1.1.*,+ time,+ old-locale,+ containers,+ hashable,+ text,+ bytestring,+ -- control:+ transformers-base == 0.4.*,+ monad-control == 0.3.*,+ loch-th == 0.2.*,+ placeholders == 0.1.*,+ errors >= 1.4.4,+ mtl,+ base >= 4.5 && < 5+ default-extensions:+ Arrows+ BangPatterns+ ConstraintKinds+ DataKinds+ DefaultSignatures+ DeriveDataTypeable+ DeriveGeneric+ EmptyDataDecls+ FlexibleContexts+ FlexibleInstances+ FunctionalDependencies+ GADTs+ GeneralizedNewtypeDeriving+ ImpredicativeTypes+ LambdaCase+ LiberalTypeSynonyms+ MultiParamTypeClasses+ MultiWayIf+ NoImplicitPrelude+ NoMonomorphismRestriction+ OverloadedStrings+ PatternGuards+ QuasiQuotes+ RankNTypes+ RecordWildCards+ ScopedTypeVariables+ StandaloneDeriving+ TemplateHaskell+ TupleSections+ TypeFamilies+ TypeOperators+ default-language:+ Haskell2010