eventstore-1.3.0: Database/EventStore/Internal/Operation.hs
{-# LANGUAGE DataKinds #-}
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE KindSignatures #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE Rank2Types #-}
{-# LANGUAGE StandaloneDeriving #-}
--------------------------------------------------------------------------------
-- |
-- Module : Database.EventStore.Internal.Operation
-- Copyright : (C) 2015 Yorick Laupa
-- License : (see the file LICENSE)
--
-- Maintainer : Yorick Laupa <yo.eight@gmail.com>
-- Stability : provisional
-- Portability : non-portable
--
--------------------------------------------------------------------------------
module Database.EventStore.Internal.Operation
( OpResult(..)
, OperationError(..)
, Operation
, Code
, Need(..)
, Execution(..)
, Expect(..)
, Coroutine(..)
, Payload(..)
, freshId
, failure
, retry
, send
, request
, waitForOr
, wrongVersion
, streamDeleted
, invalidTransaction
, accessDenied
, protobufDecodingError
, serverError
, invalidServerResponse
, construct
, yield
, traversing
, stop
, (<~)
, unfolding
, append
, Stream(..)
) where
--------------------------------------------------------------------------------
import Prelude (String)
import Control.Category
--------------------------------------------------------------------------------
import Data.ProtocolBuffers
import Data.Serialize
import Data.UUID
import Data.Void (Void, absurd)
import Streaming.Internal
--------------------------------------------------------------------------------
import Database.EventStore.Internal.Command
import Database.EventStore.Internal.Prelude hiding ((.), id)
import Database.EventStore.Internal.Settings
import Database.EventStore.Internal.Stream
import Database.EventStore.Internal.Types
--------------------------------------------------------------------------------
infixr 9 <~
--------------------------------------------------------------------------------
-- | Operation result sent by the server.
data OpResult
= OP_SUCCESS
| OP_PREPARE_TIMEOUT
| OP_COMMIT_TIMEOUT
| OP_FORWARD_TIMEOUT
| OP_WRONG_EXPECTED_VERSION
| OP_STREAM_DELETED
| OP_INVALID_TRANSACTION
| OP_ACCESS_DENIED
deriving (Eq, Enum, Show)
--------------------------------------------------------------------------------
-- | Operation exception that can occurs on an operation response.
data OperationError
= WrongExpectedVersion Text ExpectedVersion -- ^ Stream and Expected Version
| StreamDeleted StreamName -- ^ Stream
| InvalidTransaction
| forall t. AccessDenied (StreamId t) -- ^ Stream
| InvalidServerResponse Command Command -- ^ Expected, Found
| ProtobufDecodingError String
| ServerError (Maybe Text) -- ^ Reason
| InvalidOperation Text
| StreamNotFound StreamName
| NotAuthenticatedOp
-- ^ Invalid operation state. If happens, it's a driver bug.
| Aborted
-- ^ Occurs when the user asked to close the connection or if the
-- connection can't reconnect anymore.
deriving Typeable
--------------------------------------------------------------------------------
deriving instance Show OperationError
--------------------------------------------------------------------------------
instance Exception OperationError
--------------------------------------------------------------------------------
data Payload =
Payload
{ payloadCmd :: !Command
, payloadData :: !ByteString
, payloadCreds :: !(Maybe Credentials)
}
--------------------------------------------------------------------------------
data Coroutine k o a where
Yield :: o -> a -> Coroutine k o a
Await :: (i -> a) -> k i -> a -> Coroutine k o a
Stop :: Coroutine k o a
--------------------------------------------------------------------------------
instance Functor (Coroutine k o) where
fmap f (Yield o a) = Yield o (f a)
fmap f (Await k n a) = Await (f . k) n (f a)
fmap _ Stop = Stop
--------------------------------------------------------------------------------
data Is a b where
Same :: Is a a
--------------------------------------------------------------------------------
instance Category Is where
id = Same
Same . Same = Same
--------------------------------------------------------------------------------
data Need a where
NeedUUID :: Need UUID
NeedRemote :: Payload -> Need Package
WaitRemote :: UUID -> Need (Maybe Package)
--------------------------------------------------------------------------------
data Execution a
= Proceed a
| Retry
| Failed !OperationError
--------------------------------------------------------------------------------
instance Functor Execution where
fmap f (Proceed a) = Proceed (f a)
fmap _ Retry = Retry
fmap _ (Failed e) = Failed e
--------------------------------------------------------------------------------
instance Applicative Execution where
pure = return
(<*>) = ap
--------------------------------------------------------------------------------
instance Monad Execution where
return = Proceed
Proceed a >>= f = f a
Retry >>= _ = Retry
Failed e >>= _ = Failed e
--------------------------------------------------------------------------------
type Machine k o m r = Stream (Coroutine k o) m r
type Code output a = Machine Need output Execution a
type Operation output = Code output Void
type Process m a b = Stream (Coroutine (Is a) b) m Void
--------------------------------------------------------------------------------
awaits :: Monad m => k i -> Machine k o m i
awaits instr = Step (Await pure instr (Step Stop))
--------------------------------------------------------------------------------
await :: (Monad m, Category k) => Machine (k i) o m i
await = awaits id
--------------------------------------------------------------------------------
stop :: Machine k o m a
stop = Step Stop
--------------------------------------------------------------------------------
yield :: Monad m => o -> Machine k o m ()
yield o = Step (Yield o (pure ()))
--------------------------------------------------------------------------------
traversing :: Monad m => (a -> m b) -> Process m a b
traversing k = repeatedly $ do
a <- await
b <- lift (k a)
yield b
--------------------------------------------------------------------------------
append :: Operation o -> Operation o -> Operation o
append start right = go start
where
go cur =
case cur of
Return x -> absurd x
Effect m -> Effect (fmap go m)
Step step ->
case step of
Yield o next ->
Step $ Yield o (append next right)
Await k instr failed ->
Step $ Await (\i -> append (k i) right) instr (append failed right)
Stop ->
right
--------------------------------------------------------------------------------
stepping :: Operation a -> Code o (a, Operation a)
stepping = go
where
go cur =
case cur of
Return x -> absurd x
Effect m -> Effect (fmap go m)
Step step ->
case step of
Yield a next -> pure (a, next)
Await k instr failed -> Step $ Await (go . k) instr (go failed)
Stop -> stop
--------------------------------------------------------------------------------
unfolding :: (Maybe a -> Code o (Operation a)) -> Operation o
unfolding k = k Nothing >>= go
where
go cur = do
(a, next) <- stepping cur
newState <- k (Just a)
go (append next newState)
--------------------------------------------------------------------------------
repeatedly :: Functor m => Machine k o m x -> Machine k o m r
repeatedly start = go start
where
go (Return _) = go start
go (Effect m) = Effect (fmap go m)
go (Step step) =
case step of
Yield o next -> Step $ Yield o (go next)
Await k instr failed -> Step $ Await (go . k) instr (go failed)
Stop -> stop
--------------------------------------------------------------------------------
(<~) :: Monad m => Process m a b -> Machine k a m r -> Machine k b m r
mp <~ mb =
case mp of
Return x -> absurd x
Effect m -> Effect (fmap (<~ mb) m)
Step consumer ->
case consumer of
Yield c next -> Step $ Yield c (next <~ mb)
Stop -> stop
Await k Same failed ->
case mb of
Return _ -> failed <~ stop
Effect m -> Effect (fmap (Step consumer <~) m)
Step producer ->
case producer of
Yield b next -> k b <~ next
Await kb instr kfailed ->
Step (Await ((mp <~) . kb) instr (mp <~ kfailed))
Stop -> failed <~ stop
--------------------------------------------------------------------------------
-- | Asks for a unused 'UUID'.
freshId :: Code o UUID
freshId = Step $ Await pure NeedUUID (Step Stop)
--------------------------------------------------------------------------------
-- | Raises an 'OperationError'.
failure :: OperationError -> Code o a
failure = lift . Failed
--------------------------------------------------------------------------------
-- | Asks to resume the interpretation from the beginning.
retry :: Code o a
retry = lift Retry
--------------------------------------------------------------------------------
-- | Sends a package to the server and wait for a response.
sendRemote :: Payload -> Code o Package
sendRemote p = Step $ Await pure (NeedRemote p) (Step Stop)
--------------------------------------------------------------------------------
-- | Waits package from the server.
waitRemote :: UUID -> Code o (Maybe Package)
waitRemote c = Step $ Await pure (WaitRemote c) (Step Stop)
--------------------------------------------------------------------------------
construct :: Code o a -> Operation o
construct m = m >> stop
--------------------------------------------------------------------------------
-- | Like 'request' except it discards the correlation id of the network
-- exchange.
send :: (Encode req, Decode resp)
=> Command
-> Command
-> Maybe Credentials
-> req
-> Code o resp
send reqCmd expCmd cred req = do
let dat = runPut $ encodeMessage req
payload = Payload reqCmd dat cred
pkg <- sendRemote payload
let gotCmd = packageCmd pkg
when (gotCmd /= expCmd)
(invalidServerResponse expCmd gotCmd)
case runGet decodeMessage (packageData pkg) of
Left e -> protobufDecodingError e
Right resp -> return resp
--------------------------------------------------------------------------------
data Expect o where
Expect :: Decode resp => Command -> (UUID -> resp -> Code o ()) -> Expect o
--------------------------------------------------------------------------------
-- | Runs the first expection that matches.
runFirstMatch :: Package -> [Expect o] -> Code o ()
runFirstMatch _ [] = invalidOperation "No expectation was fulfilled"
runFirstMatch pkg (Expect cmd k:rest)
| packageCmd pkg /= cmd = runFirstMatch pkg rest
| otherwise =
case runGet decodeMessage (packageData pkg) of
Left e -> protobufDecodingError e
Right resp -> k (packageCorrelation pkg) resp
--------------------------------------------------------------------------------
-- | Sends a message to remote server. It returns the expected deserialized
-- message along with the correlation id of the network exchange.
request :: Encode req
=> Command
-> Maybe Credentials
-> req
-> [Expect o]
-> Code o ()
request reqCmd cred rq exps = do
let dat = runPut $ encodeMessage rq
payload = Payload reqCmd dat cred
pkg <- sendRemote payload
runFirstMatch pkg exps
--------------------------------------------------------------------------------
-- | @waitForElse uuid alternative expects@ Waits for a message from the server
-- at the given /uuid/. If the connection has been reset in the meantime, it
-- will use /alternative/.
waitForOr :: UUID -> Code o () -> [Expect o] -> Code o ()
waitForOr pid alt exps =
waitRemote pid >>= \case
Nothing -> alt
Just pkg ->
runFirstMatch pkg exps
--------------------------------------------------------------------------------
-- | Raises 'WrongExpectedVersion' exception.
wrongVersion :: Text -> ExpectedVersion -> Code o a
wrongVersion stream ver = failure (WrongExpectedVersion stream ver)
--------------------------------------------------------------------------------
-- | Raises 'StreamDeleted' exception.
streamDeleted :: StreamName -> Code o a
streamDeleted stream = failure (StreamDeleted stream)
--------------------------------------------------------------------------------
-- | Raises 'InvalidTransaction' exception.
invalidTransaction :: Code o a
invalidTransaction = failure InvalidTransaction
--------------------------------------------------------------------------------
-- | Raises 'AccessDenied' exception.
accessDenied :: StreamId t -> Code o a
accessDenied = failure . AccessDenied
--------------------------------------------------------------------------------
-- | Raises 'ProtobufDecodingError' exception.
protobufDecodingError :: String -> Code o a
protobufDecodingError = failure . ProtobufDecodingError
--------------------------------------------------------------------------------
-- | Raises 'ServerError' exception.
serverError :: Maybe Text -> Code o a
serverError = failure . ServerError
--------------------------------------------------------------------------------
-- | Raises 'InvalidServerResponse' exception.
invalidServerResponse :: Command -> Command -> Code o a
invalidServerResponse expe got = failure $ InvalidServerResponse expe got
--------------------------------------------------------------------------------
invalidOperation :: Text -> Code o a
invalidOperation = failure . InvalidOperation