module Control.Distributed.Process.FSM.Internal.Types
( apply
, applyTransitions
, State(..)
, Transition(..)
, Event(..)
, Stopping(..)
, resolveEvent
, Step(..)
, FSM(..)
, runFSM
, lift
, liftIO
, currentState
, currentInput
, currentMessage
, stateData
, addTransition
, baseErr
, decodeToEvent
) where
import Control.Distributed.Process
( Process
, unwrapMessage
, handleMessage
, handleMessageIf
, wrapMessage
, die
)
import qualified Control.Distributed.Process as P
( liftIO
, Message
)
import Control.Distributed.Process.Extras (ExitReason(..))
import Control.Distributed.Process.ManagedProcess
( Action
, GenProcess
, continue
, stopWith
, setProcessState
, processState
)
import qualified Control.Distributed.Process.ManagedProcess.Internal.GenProcess as Gen (enqueue, push)
import Control.Distributed.Process.ManagedProcess.Internal.Types
( Priority(..)
)
import qualified Control.Distributed.Process.ManagedProcess.Internal.Types as MP
( lift
)
import Control.Distributed.Process.ManagedProcess.Server.Priority (act)
import Control.Distributed.Process.Serializable (Serializable)
import Control.Monad.Fix (MonadFix)
import Control.Monad.IO.Class (MonadIO)
import qualified Control.Monad.State.Strict as ST
( MonadState
, StateT
, get
, modify
, lift
, runStateT
)
import Data.Binary
import Data.Maybe (fromJust, isJust)
import Data.Sequence
( Seq
, ViewR(..)
, (<|)
, (|>)
, viewr
)
import qualified Data.Sequence as Q (null)
import Data.Typeable (Typeable, typeOf)
import Data.Tuple (swap, uncurry)
import GHC.Generics
data State s d = (Show s, Eq s) =>
State { stName :: s
, stData :: d
, stProg :: Step s d
, stInput :: Maybe P.Message
, stReply :: (P.Message -> Process ())
, stTrans :: Seq (Transition s d)
, stQueue :: Seq P.Message
}
instance forall s d . (Show s) => Show (State s d) where
show State{..} = "State{stName=" ++ (show stName)
++ ", stTrans=" ++ (show stTrans) ++ "}"
data Transition s d = Remain
| PutBack
| Push P.Message
| Enqueue P.Message
| Postpone
| Enter s
| Stop ExitReason
| Eval (GenProcess (State s d) ())
instance forall s d . (Show s) => Show (Transition s d) where
show Remain = "Remain"
show PutBack = "PutBack"
show Postpone = "Postpone"
show (Push m) = "Push " ++ (show m)
show (Enqueue m) = "Enqueue " ++ (show m)
show (Enter s) = "Enter " ++ (show s)
show (Stop er) = "Stop " ++ (show er)
show (Eval _) = "Eval"
data Event m where
Wait :: (Serializable m) => Event m
WaitP :: (Serializable m) => Priority () -> Event m
Event :: (Serializable m) => m -> Event m
data Stopping = Stopping { reason :: ExitReason
, errored :: Bool
} deriving (Typeable, Generic, Show)
instance Binary Stopping where
resolveEvent :: forall s d m . (Serializable m)
=> Event m
-> P.Message
-> State s d
-> m
-> Process (Int, P.Message)
resolveEvent ev m _ _
| WaitP p <- ev = return (getPrio p, m)
| otherwise = return (0, m)
instance forall m . (Typeable m) => Show (Event m) where
show ev@Wait = show $ "Wait::" ++ (show $ typeOf ev)
show ev@(WaitP _) = show $ "WaitP::" ++ (show $ typeOf ev)
show ev = show $ typeOf ev
data Step s d where
Init :: Step s d -> Step s d -> Step s d
Yield :: s -> d -> Step s d
SafeWait :: (Serializable m) => Event m -> Step s d -> Step s d
Await :: (Serializable m) => Event m -> Step s d -> Step s d
Always :: (Serializable m) => (m -> FSM s d (Transition s d)) -> Step s d
Perhaps :: (Eq s) => s -> FSM s d (Transition s d) -> Step s d
Matching :: (Serializable m) => (m -> Bool) -> (m -> FSM s d (Transition s d)) -> Step s d
Sequence :: Step s d -> Step s d -> Step s d
Alternate :: Step s d -> Step s d -> Step s d
Reply :: (Serializable r) => FSM s d r -> Step s d
instance forall s d . (Show s) => Show (Step s d) where
show st
| Init _ _ <- st = "Init"
| Yield _ _ <- st = "Yield"
| Await _ s <- st = "Await (_ " ++ (show s) ++ ")"
| SafeWait _ s <- st = "SafeWait (_ " ++ (show s) ++ ")"
| Always _ <- st = "Always _"
| Perhaps s _ <- st = "Perhaps (" ++ (show s) ++ ")"
| Matching _ _ <- st = "Matching _ _"
| Sequence a b <- st = "Sequence [" ++ (show a) ++ " |> " ++ (show b) ++ "]"
| Alternate a b <- st = "Alternate [" ++ (show a) ++ " .| " ++ (show b) ++ "]"
| Reply _ <- st = "Reply"
newtype FSM s d o = FSM {
unFSM :: ST.StateT (State s d) Process o
}
deriving ( Functor
, Monad
, ST.MonadState (State s d)
, MonadIO
, MonadFix
, Typeable
, Applicative
)
runFSM :: State s d -> FSM s d o -> Process (o, State s d)
runFSM state proc = ST.runStateT (unFSM proc) state
lift :: Process a -> FSM s d a
lift p = FSM $ ST.lift p
liftIO :: IO a -> FSM s d a
liftIO = lift . P.liftIO
currentState :: FSM s d s
currentState = ST.get >>= return . stName
stateData :: FSM s d d
stateData = ST.get >>= return . stData
currentMessage :: forall s d . FSM s d P.Message
currentMessage = ST.get >>= return . fromJust . stInput
currentInput :: forall s d m . (Serializable m) => FSM s d (Maybe m)
currentInput = currentMessage >>= \m -> lift (unwrapMessage m :: Process (Maybe m))
addTransition :: Transition s d -> FSM s d ()
addTransition t = ST.modify (\s -> fromJust $ enqueue s (Just t) )
seqEnqueue :: Seq a -> a -> Seq a
seqEnqueue s a = a <| s
seqPush :: Seq a -> a -> Seq a
seqPush s a = s |> a
seqPop :: Seq a -> Maybe (a, Seq a)
seqPop s = maybe Nothing (\(s' :> a) -> Just (a, s')) $ getR s
getR :: Seq a -> Maybe (ViewR a)
getR s =
case (viewr s) of
EmptyR -> Nothing
a -> Just a
enqueue :: State s d -> Maybe (Transition s d) -> Maybe (State s d)
enqueue st@State{..} trans
| isJust trans = Just $ st { stTrans = seqPush stTrans (fromJust trans) }
| otherwise = Nothing
apply :: (Show s) => State s d -> P.Message -> Step s d -> Process (Maybe (State s d))
apply st msg step
| Init is ns <- step = do
st' <- apply st msg is
case st' of
Just s -> apply (s { stProg = ns }) msg ns
Nothing -> die $ ExitOther $ baseErr ++ ":InitFailed"
| Yield sn sd <- step = do
return $ Just $ st { stName = sn, stData = sd }
| SafeWait evt act' <- step = do
let ev = decodeToEvent evt msg
if isJust (ev) then apply st msg act'
else
return Nothing
| Await evt act' <- step = do
let ev = decodeToEvent evt msg
if isJust (ev) then apply st msg act'
else
return Nothing
| Always fsm <- step = do
runFSM st (handleMessage msg fsm) >>= mstash
| Perhaps eqn act' <- step = do
if eqn == (stName st) then runFSM st act' >>= stash
else return Nothing
| Matching chk fsm <- step = do
runFSM st (handleMessageIf msg chk fsm) >>= mstash
| Sequence ac1 ac2 <- step = do s <- apply st msg ac1
if isJust s then apply (fromJust s) msg ac2
else return Nothing
| Alternate al1 al2 <- step = do s <- apply st msg al1
if isJust s then return s
else apply st msg al2
| Reply rply <- step = do
let ev = Eval $ do fSt <- processState
s' <- MP.lift $ do (r, s) <- runFSM fSt rply
(stReply s) $ wrapMessage r
return s
setProcessState s'
return $ enqueue st (Just ev)
| otherwise = error $ baseErr ++ ".Internal.Types.apply:InvalidStep"
where
mstash = return . uncurry enqueue . swap
stash (o, s) = return $ enqueue s (Just o)
applyTransitions :: forall s d. (Show s)
=> State s d
-> [GenProcess (State s d) ()]
-> Action (State s d)
applyTransitions st@State{..} evals
| Q.null stTrans, [] <- evals = continue $ st
| Q.null stTrans = act $ do setProcessState st
mapM_ id evals
| (tr, st2) <- next
, PutBack <- tr = applyTransitions st2 ((Gen.enqueue $ fromJust stInput) : evals)
| isJust stInput
, input <- fromJust stInput
, (tr, st2) <- next
, Postpone <- tr = applyTransitions (st2 { stQueue = seqEnqueue stQueue input }) evals
| (tr, st2) <- next
, Enqueue m <- tr = applyTransitions st2 ((Gen.enqueue m):evals)
| (tr, st2) <- next
, Push m <- tr = applyTransitions st2 ((Gen.push m):evals)
| (tr, st2) <- next
, Eval proc <- tr = applyTransitions st2 (proc:evals)
| (tr, st2) <- next
, Remain <- tr = applyTransitions st2 evals
| (tr, _) <- next
, Stop er <- tr = stopWith st er
| (tr, st2) <- next
, Enter s <- tr =
if s == stName then applyTransitions st2 evals
else do let st' = st2 { stName = s }
let evals' = if Q.null stQueue then evals
else (mapM_ Gen.push stQueue) : evals
applyTransitions st' evals'
| otherwise = error $ baseErr ++ ".Internal.Process.applyTransitions:InvalidTransition"
where
next = let (t, q) = fromJust $ seqPop stTrans
in (t, st { stTrans = q })
baseErr :: String
baseErr = "Control.Distributed.Process.FSM"
decodeToEvent :: Serializable m => Event m -> P.Message -> Maybe (Event m)
decodeToEvent Wait msg = unwrapMessage msg >>= fmap Event
decodeToEvent (WaitP _) msg = unwrapMessage msg >>= fmap Event
decodeToEvent ev@(Event _) _ = Just ev