eventstore 0.10.0.2 → 0.11.0.0
raw patch · 9 files changed
+657/−56 lines, 9 filesdep +arraydep +dnsdep +http-client
Dependencies added: array, dns, http-client, semigroups
Files
- CHANGELOG.markdown +5/−0
- Database/EventStore.hs +41/−7
- Database/EventStore/Internal/Connection.hs +45/−15
- Database/EventStore/Internal/Discovery.hs +518/−0
- Database/EventStore/Internal/Execution/Production.hs +5/−4
- Database/EventStore/Internal/Types.hs +16/−9
- README.md +18/−16
- eventstore.cabal +6/−1
- tests/integration.hs +3/−4
CHANGELOG.markdown view
@@ -1,3 +1,8 @@+0.11.0.0+--------+* Implement Cluster connection.+* Domain can be used to connect to a server instance.+ 0.10.0.2 -------- * Bump aeson version.
Database/EventStore.hs view
@@ -16,6 +16,7 @@ module Database.EventStore ( -- * Connection Connection+ , ConnectionType(..) , ConnectionException(..) , ServerConnectionError(..) , Credentials@@ -28,6 +29,18 @@ , connect , shutdown , waitTillClosed+ , connectionSettings+ -- * Cluster Connection+ , ClusterSettings(..)+ , DnsServer(..)+ , GossipSeed+ , gossipSeed+ , gossipSeedWithHeader+ , gossipSeedHost+ , gossipSeedHeader+ , gossipSeedPort+ , gossipSeedClusterSettings+ , dnsClusterSettings -- * Event , Event , EventData@@ -186,6 +199,8 @@ -- * Re-export , module Control.Concurrent.Async , (<>)+ , NonEmpty(..)+ , nonEmpty ) where --------------------------------------------------------------------------------@@ -193,6 +208,7 @@ import Control.Concurrent.STM import Control.Exception import Control.Monad (when)+import Data.ByteString (ByteString) import Data.Int import Data.Maybe import Data.Monoid ((<>))@@ -200,11 +216,13 @@ -------------------------------------------------------------------------------- import Control.Concurrent.Async+import Data.List.NonEmpty(NonEmpty(..), nonEmpty) import Data.Text hiding (group) import Data.UUID -------------------------------------------------------------------------------- import Database.EventStore.Internal.Connection hiding (Connection)+import Database.EventStore.Internal.Discovery import qualified Database.EventStore.Internal.Manager.Subscription as S import Database.EventStore.Internal.Manager.Subscription.Message import Database.EventStore.Internal.Operation (OperationError(..))@@ -219,11 +237,21 @@ -------------------------------------------------------------------------------- -- Connection --------------------------------------------------------------------------------+-- | Gathers every connection type handled by the client.+data ConnectionType+ = Static String Int+ -- ^ HostName and Port.+ | Cluster ClusterSettings+ | Dns ByteString (Maybe DnsServer) Int+ -- ^ Domain name, optional DNS server and port.++-------------------------------------------------------------------------------- -- | Represents a connection to a single EventStore node. data Connection = Connection { _prod :: Production , _settings :: Settings+ , _type :: ConnectionType } --------------------------------------------------------------------------------@@ -239,18 +267,24 @@ -- or a single thread can make many asynchronous requests. To get the most -- performance out of the connection it is generally recommended to use it in -- this way.-connect :: Settings- -> String -- ^ HostName- -> Int -- ^ Port- -> IO Connection-connect settings host port = do- prod <- newExecutionModel settings host port- return $ Connection prod settings+connect :: Settings -> ConnectionType -> IO Connection+connect settings tpe = do+ disc <- case tpe of+ Static host port -> return $ staticEndPointDiscovery host port+ Cluster setts -> clusterDnsEndPointDiscovery setts+ Dns dom srv port -> return $ simpleDnsEndPointDiscovery dom srv port+ prod <- newExecutionModel settings disc+ return $ Connection prod settings tpe -------------------------------------------------------------------------------- -- | Waits the 'Connection' to be closed. waitTillClosed :: Connection -> IO () waitTillClosed Connection{..} = prodWaitTillClosed _prod++--------------------------------------------------------------------------------+-- | Returns a 'Connection''s 'Settings'.+connectionSettings :: Connection -> Settings+connectionSettings = _settings -------------------------------------------------------------------------------- -- | Asynchronously closes the 'Connection'.
Database/EventStore/Internal/Connection.hs view
@@ -30,6 +30,7 @@ import Control.Concurrent.STM import Control.Exception import qualified Data.ByteString as B+import Data.IORef import Data.Typeable import System.IO @@ -39,6 +40,7 @@ import Network --------------------------------------------------------------------------------+import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Types import Database.EventStore.Logging @@ -46,9 +48,8 @@ -- | Type of connection issue that can arise during the communication with the -- server. data ConnectionException- = MaxAttemptConnectionReached HostName Int Int- -- ^ The max reconnection attempt threshold has been reached. Holds a- -- 'HostName', the port used and the given threshold.+ = MaxAttemptConnectionReached+ -- ^ The max reconnection attempt threshold has been reached. | ClosedConnection -- ^ Use of a close 'Connection'. deriving (Show, Typeable)@@ -68,8 +69,8 @@ data Connection = Connection { _var :: TMVar State- , _host :: HostName- , _port :: Int+ , _last :: IORef (Maybe EndPoint)+ , _disc :: Discovery , _setts :: Settings } @@ -81,10 +82,11 @@ -------------------------------------------------------------------------------- -- | Creates a new 'Connection'.-newConnection :: Settings -> HostName -> Int -> IO Connection-newConnection setts host port = do+newConnection :: Settings -> Discovery -> IO Connection+newConnection setts disc = do var <- newTMVarIO Offline- return $ Connection var host port setts+ ref <- newIORef Nothing+ return $ Connection var ref disc setts -------------------------------------------------------------------------------- -- | Gets current 'Connection' 'UUID'.@@ -144,7 +146,7 @@ throwIO e Right alt -> do sres <- case alt of- Nothing -> newState _setts _host _port+ Nothing -> newState _setts _last _disc Just (u, h) -> return $ Right $ Online u h case sres of Left e -> do@@ -160,25 +162,53 @@ Close -> error "impossible execute" ---------------------------------------------------------------------------------newState :: Settings -> HostName -> Int -> IO (Either ConnectionException State)-newState sett host port =+newState :: Settings+ -> IORef (Maybe EndPoint)+ -> Discovery+ -> IO (Either ConnectionException State)+newState sett ref disc = case s_retry sett of AtMost n -> let loop i = do _settingsLog sett (Info $ Connecting i)- let action = fmap Right $ connect sett host port+ let action = do+ old <- readIORef ref+ ept_opt <- runDiscovery disc old+ case ept_opt of+ Nothing -> do+ threadDelay delay+ if n <= i+ then return $+ Left MaxAttemptConnectionReached+ else loop (i + 1)+ Just ept -> do+ let host = endPointIp ept+ port = endPointPort ept+ st <- connect sett host port+ writeIORef ref (Just ept)+ return $ Right st catch action $ \(_ :: SomeException) -> do threadDelay delay if n <= i then return $- Left $- MaxAttemptConnectionReached host port n+ Left MaxAttemptConnectionReached else loop (i + 1) in loop 1 KeepRetrying -> let endlessly i = do _settingsLog sett (Info $ Connecting i)- let action = fmap Right $ connect sett host port+ let action = do+ old <- readIORef ref+ ept_opt <- runDiscovery disc old+ case ept_opt of+ Nothing -> threadDelay delay+ >> endlessly (i + 1)+ Just ept -> do+ let host = endPointIp ept+ port = endPointPort ept+ st <- connect sett host port+ writeIORef ref (Just ept)+ return $ Right st catch action $ \(_ :: SomeException) -> threadDelay delay >> endlessly (i + 1) in endlessly (1 :: Int)
+ Database/EventStore/Internal/Discovery.hs view
@@ -0,0 +1,518 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}+--------------------------------------------------------------------------------+-- |+-- Module : Database.EventStore.Internal.Discovery+-- Copyright : (C) 2016 Yorick Laupa+-- License : (see the file LICENSE)+--+-- Maintainer : Yorick Laupa <yo.eight@gmail.com>+-- Stability : provisional+-- Portability : non-portable+--+--------------------------------------------------------------------------------+module Database.EventStore.Internal.Discovery+ ( Discovery(..)+ , GossipSeed+ , DnsDiscoveryException(..)+ , ClusterSettings(..)+ , DnsServer(..)+ , EndPoint(..)+ , staticEndPointDiscovery+ , clusterDnsEndPointDiscovery+ , gossipSeedClusterSettings+ , simpleDnsEndPointDiscovery+ , dnsClusterSettings+ , gossipSeed+ , gossipSeedWithHeader+ , gossipSeedHeader+ , gossipSeedHost+ , gossipSeedPort+ ) where++--------------------------------------------------------------------------------+import Control.Applicative+import Control.Exception+import Control.Monad+import Data.Foldable (toList)+import Data.IORef+import Data.List (sortBy)+import Data.Maybe+import Data.Ord+import Data.Typeable+import GHC.Generics hiding (from, to)+import Prelude++--------------------------------------------------------------------------------+import Data.Aeson+import Data.Aeson.Types+import Data.Array.IO+import Data.ByteString (ByteString)+import Data.Int+import Data.List.NonEmpty (NonEmpty)+import Data.UUID+import Network.HTTP.Client+import Network.DNS hiding (decode)+import System.Random++--------------------------------------------------------------------------------+import Database.EventStore.Internal.TimeSpan++--------------------------------------------------------------------------------+data DnsDiscoveryException+ = MaxDiscoveryAttemptReached ByteString+ | DNSDiscoveryError DNSError+ deriving (Show, Typeable)++--------------------------------------------------------------------------------+instance Exception DnsDiscoveryException++--------------------------------------------------------------------------------+-- | Gathers both an IPv4 and a port.+data EndPoint =+ EndPoint+ { endPointIp :: !String+ , endPointPort :: !Int+ } deriving Show++--------------------------------------------------------------------------------+emptyEndPoint :: EndPoint+emptyEndPoint = EndPoint "" 0++--------------------------------------------------------------------------------+httpRequest :: EndPoint -> String -> IO Request+httpRequest (EndPoint ip p) path = parseUrl url+ where+ url = "http://" ++ ip ++ ":" ++ show p ++ path++--------------------------------------------------------------------------------+-- | Represents a source of cluster gossip.+data GossipSeed =+ GossipSeed+ { gossipEndpoint :: !EndPoint+ -- ^ The endpoint for the external HTTP endpoint of the gossip seed. The+ -- HTTP endpoint is used rather than the TCP endpoint because it is+ -- required for the client to exchange gossip with the server.+ -- standard port which should be used here in 2113.+ , gossipSeedHeader :: !String+ -- ^ The host header to be sent when requesting gossip.+ } deriving Show++--------------------------------------------------------------------------------+-- | Creates a 'GossipSeed'.+gossipSeed :: String -> Int -> GossipSeed+gossipSeed h p = GossipSeed (EndPoint h p) ""++--------------------------------------------------------------------------------+-- | Creates a 'GossipSeed' with a specific HTTP header.+gossipSeedWithHeader :: String -> Int -> String -> GossipSeed+gossipSeedWithHeader h p hd = GossipSeed (EndPoint h p) hd++--------------------------------------------------------------------------------+-- | Returns 'GossipSeed' host IP address.+gossipSeedHost :: GossipSeed -> String+gossipSeedHost = endPointIp . gossipEndpoint++--------------------------------------------------------------------------------+-- | Returns 'GossipSeed' port.+gossipSeedPort :: GossipSeed -> Int+gossipSeedPort = endPointPort . gossipEndpoint++--------------------------------------------------------------------------------+emptyGossipSeed :: GossipSeed+emptyGossipSeed = GossipSeed emptyEndPoint ""++--------------------------------------------------------------------------------+-- | Procedure used to discover an network 'EndPoint'.+newtype Discovery =+ Discovery { runDiscovery :: Maybe EndPoint -> IO (Maybe EndPoint) }++--------------------------------------------------------------------------------+staticEndPointDiscovery :: String -> Int -> Discovery+staticEndPointDiscovery host port =+ Discovery $ \_ -> return $ Just $ EndPoint host port++--------------------------------------------------------------------------------+simpleDnsEndPointDiscovery :: ByteString -> Maybe DnsServer -> Int -> Discovery+simpleDnsEndPointDiscovery domain srv port = Discovery $ \_ -> do+ let conf =+ case srv of+ Nothing -> defaultResolvConf+ Just tpe ->+ let rc =+ case tpe of+ DnsFilePath p -> RCFilePath p+ DnsHostName h -> RCHostName h+ DnsHostPort h p -> RCHostPort h (fromIntegral p)+ in defaultResolvConf { resolvInfo = rc }+ dnsSeed <- makeResolvSeed conf+ res <- withResolver dnsSeed $ \resv -> lookupA resv domain+ case res of+ Left e -> throwIO $ DNSDiscoveryError e+ Right ips -> do+ let pts = [ EndPoint (show ip) port | ip <- ips ]+ case pts of+ [] -> return Nothing+ pt:_ -> return $ Just pt++--------------------------------------------------------------------------------+-- | Tells how the DNS server should be contacted.+data DnsServer+ = DnsFilePath String+ | DnsHostName String+ | DnsHostPort String Int++--------------------------------------------------------------------------------+-- | Contains settings related to a connection to a cluster.+data ClusterSettings =+ ClusterSettings+ { clusterDns :: !ByteString+ -- ^ The DNS name to use for discovering endpoints.+ , clusterMaxDiscoverAttempts :: !Int+ -- ^ The maximum number of attempts for discovering endpoints.+ , clusterExternalGossipPort :: !Int+ -- ^ The well-known endpoint on which cluster managers are running.+ , clusterGossipSeeds :: (Maybe (NonEmpty GossipSeed))+ -- ^ Endpoints for seeding gossip if not using DNS.+ , clusterGossipTimeout :: !TimeSpan+ -- ^ Timeout for cluster gossip.+ , clusterDnsServer :: !(Maybe DnsServer)+ -- ^ Indicates a specific DNS server should be contacted.+ }++--------------------------------------------------------------------------------+-- | Configures a 'ClusterSettings' for connecting to a cluster using gossip+-- seeds.+-- clusterDns = ""+-- clusterMaxDiscoverAttempts = 10+-- clusterExternalGossipPort = 0+-- clusterGossipTimeout = 1s+gossipSeedClusterSettings :: NonEmpty GossipSeed -> ClusterSettings+gossipSeedClusterSettings xs =+ ClusterSettings+ { clusterDns = ""+ , clusterMaxDiscoverAttempts = 10+ , clusterExternalGossipPort = 0+ , clusterGossipSeeds = Just xs+ , clusterGossipTimeout = timeSpanFromSeconds 1+ , clusterDnsServer = Nothing+ }++--------------------------------------------------------------------------------+-- | Configures a 'ClusterSettings' for connecting to a cluster using DNS+-- discovery.+-- clusterMaxDiscoverAttempts = 10+-- clusterExternalGossipPort = 0+-- clusterGossipSeeds = Nothing+-- clusterGossipTimeout = 1s+dnsClusterSettings :: ByteString -> ClusterSettings+dnsClusterSettings clusterDns = ClusterSettings{..}+ where+ clusterMaxDiscoverAttempts = 10+ clusterExternalGossipPort = 0+ clusterGossipSeeds = Nothing+ clusterGossipTimeout = timeSpanFromSeconds 1+ clusterDnsServer = Nothing++--------------------------------------------------------------------------------+clusterDnsEndPointDiscovery :: ClusterSettings -> IO Discovery+clusterDnsEndPointDiscovery settings = do+ ref <- newIORef Nothing+ manager <- newManager defaultManagerSettings+ return $ Discovery $ \fend -> discoverEndPoint manager ref fend settings++--------------------------------------------------------------------------------+data VNodeState+ = Initializing+ | Unknown+ | PreReplica+ | CatchingUp+ | Clone+ | Slave+ | PreMaster+ | Master+ | Manager+ | ShuttingDown+ | Shutdown+ deriving (Eq, Ord, Generic, Show)++--------------------------------------------------------------------------------+instance FromJSON VNodeState++--------------------------------------------------------------------------------+newtype GUUID = GUUID UUID deriving Show++--------------------------------------------------------------------------------+instance FromJSON GUUID where+ parseJSON (String txt) =+ case fromText txt of+ Just uuid -> return $ GUUID uuid+ _ -> fail $ "Wrong UUID format " ++ show txt+ parseJSON invalid = typeMismatch "UUID" invalid++--------------------------------------------------------------------------------+data MemberInfo =+ MemberInfo+ { _instanceId :: !GUUID+ , _state :: !VNodeState+ , _isAlive :: !Bool+ , _internalTcpIp :: !String+ , _internalTcpPort :: !Int+ , _externalTcpIp :: !String+ , _externalTcpPort :: !Int+ , _internalHttpIp :: !String+ , _internalHttpPort :: !Int+ , _externalHttpIp :: !String+ , _externalHttpPort :: !Int+ , _lastCommitPosition :: !Int64+ , _writerCheckpoint :: !Int64+ , _chaserCheckpoint :: !Int64+ , _epochPosition :: !Int64+ , _epochNumber :: !Int+ , _epochId :: !GUUID+ , _nodePriority :: !Int+ } deriving Show++--------------------------------------------------------------------------------+instance FromJSON MemberInfo where+ parseJSON (Object m) =+ MemberInfo+ <$> m .: "instanceId"+ <*> m .: "state"+ <*> m .: "isAlive"+ <*> m .: "internalTcpIp"+ <*> m .: "internalTcpPort"+ <*> m .: "externalTcpIp"+ <*> m .: "externalTcpPort"+ <*> m .: "internalHttpIp"+ <*> m .: "internalHttpPort"+ <*> m .: "externalHttpIp"+ <*> m .: "externalHttpPort"+ <*> m .: "lastCommitPosition"+ <*> m .: "writerCheckpoint"+ <*> m .: "chaserCheckpoint"+ <*> m .: "epochPosition"+ <*> m .: "epochNumber"+ <*> m .: "epochId"+ <*> m .: "nodePriority"+ parseJSON invalid = typeMismatch "MemberInfo" invalid++--------------------------------------------------------------------------------+data ClusterInfo =+ ClusterInfo { members :: [MemberInfo] }+ deriving (Show, Generic)++--------------------------------------------------------------------------------+instance FromJSON ClusterInfo++--------------------------------------------------------------------------------+discoverEndPoint :: Manager+ -> IORef (Maybe [MemberInfo])+ -> Maybe EndPoint+ -> ClusterSettings+ -> IO (Maybe EndPoint)+discoverEndPoint mgr ref fend settings = do+ old_m <- readIORef ref+ writeIORef ref Nothing+ candidates <- case old_m of+ Nothing -> gossipCandidatesFromDns settings+ Just old -> gossipCandidatesFromOldGossip fend old+ forArrayFirst candidates $ \i -> do+ c <- readArray candidates i+ res <- tryGetGossipFrom settings mgr c+ print (i, res)+ let fin_end = do+ info <- res+ best <- tryDetermineBestNode $ members info+ return (info, best)+ case fin_end of+ Nothing -> return Nothing+ Just (info, best) -> do+ writeIORef ref (Just $ members info)+ return $ Just best++--------------------------------------------------------------------------------+tryGetGossipFrom :: ClusterSettings+ -> Manager+ -> GossipSeed+ -> IO (Maybe ClusterInfo)+tryGetGossipFrom ClusterSettings{..} mgr seed = do+ init_req <- httpRequest (gossipEndpoint seed) "/gossip?format=json"+ let timeout = truncate (timeSpanTotalMillis clusterGossipTimeout * 1000)+ req = init_req { responseTimeout = Just timeout }+ resp <- httpLbs req mgr+ return $ decode $ responseBody resp++--------------------------------------------------------------------------------+tryDetermineBestNode :: [MemberInfo] -> Maybe EndPoint+tryDetermineBestNode members = node_m+ where+ nodes = [m | m <- members+ , _isAlive m+ , allowedState $ _state m+ ]++ node_m =+ case sortOn (Down . _state) nodes of+ [] -> Nothing+ n:_ -> Just $ EndPoint (_externalTcpIp n) (_externalTcpPort n)++ allowedState Manager = False+ allowedState ShuttingDown = False+ allowedState Shutdown = False+ allowedState _ = True++--------------------------------------------------------------------------------+gossipCandidatesFromOldGossip :: Maybe EndPoint+ -> [MemberInfo]+ -> IO (IOArray Int GossipSeed)+gossipCandidatesFromOldGossip fend_m oldGossip =+ arrangeGossipCandidates candidates+ where+ candidates =+ case fend_m of+ Nothing -> oldGossip+ Just fend -> [ c | c <- oldGossip+ , _externalTcpPort c == endPointPort fend+ , _externalTcpIp c == endPointIp fend+ ]++--------------------------------------------------------------------------------+data AState = AState !Int !Int++--------------------------------------------------------------------------------+arrangeGossipCandidates :: [MemberInfo] -> IO (IOArray Int GossipSeed)+arrangeGossipCandidates members = do+ arr <- newArray (0, len) emptyGossipSeed+ AState i j <- foldM (go arr) (AState (-1) len) members++ shuffle arr 0 i -- shuffle nodes+ shuffle arr j (len - 1) -- shuffle managers++ return arr+ where+ len = length members++ go :: IOArray Int GossipSeed -> AState -> MemberInfo -> IO AState+ go arr (AState i j) m =+ case _state m of+ Manager -> do+ let new_j = j - 1+ writeArray arr new_j seed+ return (AState i new_j)+ _ -> do+ let new_i = i + 1+ writeArray arr new_i seed+ return (AState new_i j)+ where+ end = EndPoint (_externalHttpIp m) (_externalHttpPort m)+ seed = GossipSeed end ""++--------------------------------------------------------------------------------+gossipCandidatesFromDns :: ClusterSettings -> IO (IOArray Int GossipSeed)+gossipCandidatesFromDns settings@ClusterSettings{..} = do+ arr <- endpoints+ shuffleAll arr+ return arr+ where+ endpoints =+ case clusterGossipSeeds of+ Nothing -> resolveDns settings+ Just ss -> let ls = toList ss+ len = length ls+ in newListArray (0, len - 1) ls++--------------------------------------------------------------------------------+resolveDns :: ClusterSettings -> IO (IOArray Int GossipSeed)+resolveDns ClusterSettings{..} = do+ let timeoutMicros = timeSpanTotalMillis clusterGossipTimeout * 1000+ conf =+ case clusterDnsServer of+ Nothing -> defaultResolvConf+ Just tpe ->+ let rc =+ case tpe of+ DnsFilePath p -> RCFilePath p+ DnsHostName h -> RCHostName h+ DnsHostPort h p -> RCHostPort h (fromIntegral p)+ in defaultResolvConf { resolvInfo = rc }+ dnsSeed <- makeResolvSeed conf+ { resolvTimeout = truncate timeoutMicros+ , resolvRetry = clusterMaxDiscoverAttempts+ }+ withResolver dnsSeed $ \resv -> do+ result <- lookupA resv clusterDns+ case result of+ Left e -> throwIO $ DNSDiscoveryError e+ Right ips -> do+ let len = length ips - 1+ arr <- newArray_ (0, len)+ forM_ (zip [0..] ips) $ \(idx, ip) -> do+ let end = EndPoint (show ip) clusterExternalGossipPort+ seed = GossipSeed end ""+ writeArray arr idx seed+ return arr++--------------------------------------------------------------------------------+shuffleAll :: IOArray Int a -> IO ()+shuffleAll arr = do+ (low, hig) <- getBounds arr+ shuffle arr low hig++--------------------------------------------------------------------------------+shuffle :: IOArray Int a -> Int -> Int -> IO ()+shuffle arr from to = forRange_ from to $ \cur -> do+ idx <- randomRIO (cur, to)+ tmp <- readArray arr idx+ value <- readArray arr cur+ writeArray arr idx value+ writeArray arr cur tmp++--------------------------------------------------------------------------------+forRange_ :: Int -> Int -> (Int -> IO ()) -> IO ()+forRange_ from to k = do+ when (from <= to) $ loop (to + 1) from+ where+ loop len cur+ | len == cur = return ()+ | otherwise = do+ k cur+ loop len (cur + 1)++--------------------------------------------------------------------------------+forArrayFirst :: IOArray Int a+ -> (Int -> IO (Maybe b))+ -> IO (Maybe b)+forArrayFirst arr k = do+ (low, hig) <- getBounds arr+ forRangeFirst low hig k++--------------------------------------------------------------------------------+forRangeFirst :: Int+ -> Int+ -> (Int -> IO (Maybe b))+ -> IO (Maybe b)+forRangeFirst from to k = do+ if from <= to then loop (to + 1) from else return Nothing+ where+ loop len cur+ | len == cur = return Nothing+ | otherwise = do+ res <- k cur+ if isJust res then return res else loop len (cur + 1)++--------------------------------------------------------------------------------+-- | Taken from base >= 4.8 because prior base don't have it.+-- Sort a list by comparing the results of a key function applied to each+-- element. @sortOn f@ is equivalent to @sortBy . comparing f@, but has the+-- performance advantage of only evaluating @f@ once for each element in the+-- input list. This is called the decorate-sort-undecorate paradigm, or+-- Schwartzian transform.+--+-- @since 4.8.0.0+sortOn :: Ord b => (a -> b) -> [a] -> [a]+sortOn f =+ map snd . sortBy (comparing fst) . map (\x -> let y = f x in y `seq` (y, x))
Database/EventStore/Internal/Execution/Production.hs view
@@ -60,6 +60,7 @@ -------------------------------------------------------------------------------- import Database.EventStore.Internal.Connection+import Database.EventStore.Internal.Discovery import Database.EventStore.Internal.Generator import Database.EventStore.Internal.Manager.Subscription hiding ( submitPackage@@ -602,13 +603,13 @@ -------------------------------------------------------------------------------- -- | Main Production execution model entry point.-newExecutionModel :: Settings -> HostName -> Int -> IO Production-newExecutionModel setts host port = do+newExecutionModel :: Settings -> Discovery -> IO Production+newExecutionModel setts disc = do gen <- newGenerator queue <- newCycleQueue pkg_queue <- newCycleQueue job_queue <- newCycleQueue- conn <- newConnection setts host port+ conn <- newConnection setts disc conn_ref <- newIORef conn var <- newTVarIO $ emptyState setts gen nxt_sub <- newTVarIO (atomically . writeCycleQueue queue)@@ -623,7 +624,7 @@ Just (_ :: ConnectionException) -> atomically $ do writeTVar nxt_sub (raiseException e) putTMVar disposed ()- _ -> do new_conn <- newConnection setts host port+ _ -> do new_conn <- newConnection setts disc writeIORef conn_ref new_conn _ <- forkFinally (bootstrap env) handler return ()
Database/EventStore/Internal/Types.hs view
@@ -525,16 +525,23 @@ -------------------------------------------------------------------------------- -- | Default global settings.+-- s_heartbeatInterval = 750 ms+-- s_heartbeatTimeout = 1500 ms+-- s_requireMaster = True+-- s_credentials = Nothing+-- s_retry = 'atMost' 3+-- s_reconnect_delay_secs = 3+-- s_logger = Nothing defaultSettings :: Settings-defaultSettings = Settings- { s_heartbeatInterval = msDiffTime 750 -- 750ms- , s_heartbeatTimeout = msDiffTime 1500 -- 1500ms- , s_requireMaster = True- , s_credentials = Nothing- , s_retry = atMost 3- , s_reconnect_delay_secs = 3- , s_logger = Nothing- }+defaultSettings = Settings+ { s_heartbeatInterval = msDiffTime 750 -- 750ms+ , s_heartbeatTimeout = msDiffTime 1500 -- 1500ms+ , s_requireMaster = True+ , s_credentials = Nothing+ , s_retry = atMost 3+ , s_reconnect_delay_secs = 3+ , s_logger = Nothing+ } -------------------------------------------------------------------------------- -- | Triggers the logger callback if it has been set.
README.md view
@@ -16,6 +16,7 @@ * Authenticated communication with EventStore server. * Read stream metadata (ACL and custom properties). * Write stream metadata (ACL and custom properties).+ * Cluster Connection. Not implemented yet ===================@@ -41,7 +42,7 @@ $ git clone https://github.com/YoEight/eventstore.git $ cd eventstore $ cabal install --only-dependencies-$ cabal configure +$ cabal configure $ cabal install ``` @@ -60,40 +61,41 @@ ```haskell {-# LANGUAGE OverloadedStrings #-} -- That library uses `Text` pervasively. This pragma permits to use -- String literal when a Text is needed.-module Main where +module Main where import Data.Aeson -- It requires to have `aeson` package installed. Note that EventStore doesn't constraint you to JSON -- format but putting common use aside, by doing so you'll be able to use some interesting EventStore -- features like its Complex Event Processing (CEP) capabality.- + import Database.EventStore -- Note that import also re-exports 'Control.Concurrent.Async' module, allowing the use of 'wait'--- function for instance.+-- function for instance. There are also 'NonEmpty' data constructor and 'nonEmpty' function from+-- 'Data.List.NonEmpty'. main :: IO () main = do- -- A common pattern with an EventStore connection is to create a single instance only and pass it - -- wherever you need it (it's threadsafe). It's very important to not consider an EventStore connection like + -- A common pattern with an EventStore connection is to create a single instance only and pass it+ -- wherever you need it (it's threadsafe). It's very important to not consider an EventStore connection like -- its regular SQL counterpart. An EventStore connection will try its best to reconnect -- automatically to the server if the connection dropped. Of course that behavior can be tuned -- through some settings.- conn <- connect defaultSettings "127.0.0.1" 1113+ conn <- connect defaultSettings (Static "127.0.0.1" 1113) let js = "isHaskellTheBest" .= True -- (.=) comes from Data.Aeson module. evt = createEvent "programming" Nothing (withJson js)- - -- Appends an event to a stream named `languages`. ++ -- Appends an event to a stream named `languages`. as <- sendEvent conn "languages" anyVersion evt- - -- EventStore interactions are fundamentally asynchronous. Nothing requires you to wait ++ -- EventStore interactions are fundamentally asynchronous. Nothing requires you to wait -- for the completion of an operation, but it's good to know if something went wrong. _ <- wait as- - -- Again, if you decide to `shutdown` an EventStore connection, it means your application is ++ -- Again, if you decide to `shutdown` an EventStore connection, it means your application is -- about to terminate. shutdown conn- - -- Make sure the EventStore connection completes every ongoing operation. For instance, if ++ -- Make sure the EventStore connection completes every ongoing operation. For instance, if -- at the moment we call `shutdown` and some operations (or subscriptions) were still pending, -- the connection aborted all of them. waitTillClosed conn@@ -104,6 +106,6 @@ Contributions and bug reports are welcome! -BSD3 License +BSD3 License -Yorick Laupa
eventstore.cabal view
@@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version: 0.10.0.2+version: 0.11.0.0 tested-with: GHC >= 7.8.3 && < 7.11 @@ -63,6 +63,7 @@ Database.EventStore.Logging -- Modules included in this library but not exported. other-modules: Database.EventStore.Internal.Connection+ Database.EventStore.Internal.Discovery Database.EventStore.Internal.Execution.Production Database.EventStore.Internal.Generator Database.EventStore.Internal.Operation@@ -113,6 +114,10 @@ , uuid ==1.3.* , unordered-containers , stm+ , semigroups >=0.5+ , dns+ , array+ , http-client -- Directories containing source files. -- hs-source-dirs:
tests/integration.hs view
@@ -30,15 +30,14 @@ let setts = defaultSettings { s_credentials = Just $ credentials "admin" "changeit" , s_reconnect_delay_secs = 1- -- , s_logger = Just logger }- conn <- connect setts "127.0.0.1" 1113+ conn <- connect setts (Static "127.0.0.1" 1113) let tree = tests conn defaultMainWithIngredients [consoleTestReporter] tree ---------------------------------------------------------------------------------logger :: Log -> IO ()-logger l = do+_logger :: Log -> IO ()+_logger l = do t <- getCurrentTime putStr "[" putStr $ show t