diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -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. diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -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'. diff --git a/Database/EventStore/Internal/Connection.hs b/Database/EventStore/Internal/Connection.hs --- a/Database/EventStore/Internal/Connection.hs +++ b/Database/EventStore/Internal/Connection.hs @@ -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) diff --git a/Database/EventStore/Internal/Discovery.hs b/Database/EventStore/Internal/Discovery.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Discovery.hs @@ -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 +-- 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)) diff --git a/Database/EventStore/Internal/Execution/Production.hs b/Database/EventStore/Internal/Execution/Production.hs --- a/Database/EventStore/Internal/Execution/Production.hs +++ b/Database/EventStore/Internal/Execution/Production.hs @@ -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 () diff --git a/Database/EventStore/Internal/Types.hs b/Database/EventStore/Internal/Types.hs --- a/Database/EventStore/Internal/Types.hs +++ b/Database/EventStore/Internal/Types.hs @@ -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. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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 diff --git a/eventstore.cabal b/eventstore.cabal --- a/eventstore.cabal +++ b/eventstore.cabal @@ -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: diff --git a/tests/integration.hs b/tests/integration.hs --- a/tests/integration.hs +++ b/tests/integration.hs @@ -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