diff --git a/CHANGELOG.markdown b/CHANGELOG.markdown --- a/CHANGELOG.markdown +++ b/CHANGELOG.markdown @@ -1,3 +1,8 @@ +0.7.1.0 +------- +* Internal connection changes +* Allow creating an event with an existing ID + 0.7.0.1 ------- * Tight package channnel to connection instance in order to prevent loss on connection drops. diff --git a/Database/EventStore.hs b/Database/EventStore.hs --- a/Database/EventStore.hs +++ b/Database/EventStore.hs @@ -20,6 +20,7 @@ , withJsonAndMetadata -- * Connection , Connection + , ConnectionException(..) , Credentials , Settings(..) , Retry diff --git a/Database/EventStore/Internal/Connection.hs b/Database/EventStore/Internal/Connection.hs new file mode 100644 --- /dev/null +++ b/Database/EventStore/Internal/Connection.hs @@ -0,0 +1,101 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE ScopedTypeVariables #-} +-------------------------------------------------------------------------------- +-- | +-- Module : Database.EventStore.Internal.Connection +-- Copyright : (C) 2015 Yorick Laupa +-- License : (see the file LICENSE) +-- +-- Maintainer : Yorick Laupa +-- Stability : provisional +-- Portability : non-portable +-- +-------------------------------------------------------------------------------- +module Database.EventStore.Internal.Connection + ( Connection + , ConnectionException(..) + , connUUID + , connClose + , connFlush + , connSend + , connRecv + , newConnection + ) where + +-------------------------------------------------------------------------------- +import Control.Concurrent +import Control.Exception +import qualified Data.ByteString as B +import Data.Typeable +import System.IO +import Text.Printf + +-------------------------------------------------------------------------------- +import Data.UUID +import Network +import System.Random + +-------------------------------------------------------------------------------- +import Database.EventStore.Internal.Types + +-------------------------------------------------------------------------------- +data ConnectionException = + MaxAttempt HostName Int Int -- ^ HostName Port MaxAttempt's value + deriving (Show, Typeable) + +-------------------------------------------------------------------------------- +instance Exception ConnectionException + +-------------------------------------------------------------------------------- +data Connection = + Connection + { connUUID :: UUID + , connClose :: IO () + , connFlush :: IO () + , connSend :: B.ByteString -> IO () + , connRecv :: Int -> IO B.ByteString + } + +-------------------------------------------------------------------------------- +newConnection :: Settings -> HostName -> Int -> IO Connection +newConnection sett host port = + case s_retry sett of + AtMost n -> + let loop i = do + printf "Connecting...Attempt %d\n" i + catch (connect sett host port) $ \(_ :: SomeException) -> do + threadDelay delay + if n <= i then throwIO $ MaxAttempt host port n + else loop (i + 1) in + loop 1 + KeepRetrying -> + let endlessly i = do + printf "Connecting...Attempt %d\n" i + catch (connect sett host port) $ \(_ :: SomeException) -> do + threadDelay delay >> endlessly (i + 1) in + endlessly (1 :: Int) + where + delay = (s_reconnect_delay_secs sett) * secs + +-------------------------------------------------------------------------------- +secs :: Int +secs = 1000000 + +-------------------------------------------------------------------------------- +connect :: Settings -> HostName -> Int -> IO Connection +connect _ host port = do + hdl <- connectTo host (PortNumber $ fromIntegral port) + hSetBuffering hdl NoBuffering + uuid <- randomIO + return $ regularConnection hdl uuid + +-------------------------------------------------------------------------------- +regularConnection :: Handle -> UUID -> Connection +regularConnection h uuid = + Connection + { connUUID = uuid + , connClose = hClose h + , connFlush = hFlush h + , connSend = B.hPut h + , connRecv = B.hGet h + } diff --git a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs --- a/Database/EventStore/Internal/Operation/TransactionStartOperation.hs +++ b/Database/EventStore/Internal/Operation/TransactionStartOperation.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE RecordWildCards #-} -------------------------------------------------------------------------------- -- | @@ -19,10 +18,8 @@ import Control.Exception import Data.Int import Data.Maybe -#if MIN_VERSION_base(4,8,0) -#else import Data.Traversable -#endif +import Prelude -------------------------------------------------------------------------------- import Control.Concurrent.Async @@ -262,12 +259,14 @@ eventToNewEvent :: Event -> IO NewEvent eventToNewEvent evt = newEvent evt_type + evt_id evt_data_type evt_metadata_type evt_data_bytes evt_metadata_bytes where evt_type = eventType evt + evt_id = eventId evt evt_data_bytes = eventDataBytes $ eventData evt evt_data_type = eventDataType $ eventData evt evt_metadata_bytes = eventMetadataBytes $ eventData evt diff --git a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs --- a/Database/EventStore/Internal/Operation/WriteEventsOperation.hs +++ b/Database/EventStore/Internal/Operation/WriteEventsOperation.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE CPP #-} {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE DataKinds #-} -------------------------------------------------------------------------------- @@ -19,11 +18,9 @@ import Control.Concurrent import Data.Int import Data.Maybe -#if MIN_VERSION_base(4,8,0) -#else import Data.Traversable -#endif import GHC.Generics (Generic) +import Prelude -------------------------------------------------------------------------------- import Data.ProtocolBuffers @@ -150,12 +147,14 @@ eventToNewEvent :: Event -> IO NewEvent eventToNewEvent evt = newEvent evt_type + evt_id evt_data_type evt_metadata_type evt_data_bytes evt_metadata_bytes where evt_type = eventType evt + evt_id = eventId evt evt_data_bytes = eventDataBytes $ eventData evt evt_data_type = eventDataType $ eventData evt evt_metadata_bytes = eventMetadataBytes $ eventData evt diff --git a/Database/EventStore/Internal/Processor.hs b/Database/EventStore/Internal/Processor.hs --- a/Database/EventStore/Internal/Processor.hs +++ b/Database/EventStore/Internal/Processor.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE RecordWildCards #-} {-# LANGUAGE ScopedTypeVariables #-} -------------------------------------------------------------------------------- @@ -13,7 +12,8 @@ -- -------------------------------------------------------------------------------- module Database.EventStore.Internal.Processor - ( InternalException(..) + ( ConnectionException(..) + , InternalException(..) , Processor(..) , DropReason(..) , NewSubscriptionCB @@ -33,19 +33,16 @@ import Control.Exception import Data.Functor (void) import Data.Monoid ((<>)) -import Data.Typeable import Data.Word -import System.IO import Text.Printf -------------------------------------------------------------------------------- -import Control.Concurrent.Async import Data.UUID import FRP.Sodium import Network -import System.Random -------------------------------------------------------------------------------- +import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Manager.Operation import Database.EventStore.Internal.Manager.Subscription import Database.EventStore.Internal.Packages @@ -70,16 +67,6 @@ newProcessor sett = sync . network sett =<< newChan -------------------------------------------------------------------------------- --- Exception --------------------------------------------------------------------------------- -data ProcessorException - = MaxAttempt HostName Int Int -- ^ Hostname Port MaxAttempt value - deriving (Show, Typeable) - --------------------------------------------------------------------------------- -instance Exception ProcessorException - --------------------------------------------------------------------------------- -- State -------------------------------------------------------------------------------- data State @@ -101,7 +88,6 @@ data Reconnect = Reconnect data Reconnected = Reconnected UUID (IO ()) - -------------------------------------------------------------------------------- heartbeatRequestCmd :: Word8 heartbeatRequestCmd = 0x01 @@ -154,7 +140,6 @@ push_recv_io (push_con_io host port) push_reco_io - onSend host port @@ -164,7 +149,6 @@ push_recv_io push_recod_io push_reco_io - onSend host port @@ -184,6 +168,8 @@ , processorNewSubcription = push_sub } + _ <- listen onSend (writeChan chan) + return processor -------------------------------------------------------------------------------- @@ -204,55 +190,23 @@ } -------------------------------------------------------------------------------- -secs :: Int -secs = 1000000 - --------------------------------------------------------------------------------- connection :: Settings -> Chan Package -> (Package -> IO ()) -> (UUID -> IO () -> IO ()) -> IO () - -> Event Package -> HostName -> Int -> IO () -connection sett chan push_pkg push_con push_reco evt_pkg host port = go - where - go = - case s_retry sett of - AtMost n -> - let loop i = - catch (doConnect i) $ \(_ :: SomeException) -> do - threadDelay delay - if n <= i then throwIO $ MaxAttempt host port n - else loop (i + 1) in - loop 1 - KeepRetrying -> - let endlessly i = - catch (doConnect i) $ \(_ :: SomeException) -> - threadDelay delay >> endlessly (i + 1) in - endlessly (1 :: Int) - - delay = (s_reconnect_delay_secs sett) * secs - - doConnect att = do - printf "Connecting...Attempt %d\n" att - hdl <- connectTo host (PortNumber $ fromIntegral port) - hSetBuffering hdl NoBuffering - - uuid <- randomIO - as_rl <- async $ sync $ listen evt_pkg (writeChan chan) - rid <- forkFinally (readerThread push_pkg hdl) (recovering push_reco) - wid <- forkFinally (writerThread chan hdl) (recovering push_reco) - - push_con uuid $ do - throwTo rid Stopped - throwTo wid Stopped - hClose hdl - rel_w <- wait as_rl - rel_w - printf "Disconnected %s\n" (toString uuid) +connection sett chan push_pkg push_con push_reco host port = do + conn <- newConnection sett host port + rid <- forkFinally (readerThread push_pkg conn) (recovering push_reco) + wid <- forkFinally (writerThread chan conn) (recovering push_reco) + push_con (connUUID conn) $ do + throwTo rid Stopped + throwTo wid Stopped + connClose conn + printf "Disconnected %s\n" (toString $ connUUID conn) -------------------------------------------------------------------------------- recovering :: IO () -> Either SomeException () -> IO () diff --git a/Database/EventStore/Internal/Reader.hs b/Database/EventStore/Internal/Reader.hs --- a/Database/EventStore/Internal/Reader.hs +++ b/Database/EventStore/Internal/Reader.hs @@ -12,28 +12,27 @@ module Database.EventStore.Internal.Reader (readerThread) where -------------------------------------------------------------------------------- -import Prelude hiding (take) -import Control.Monad -import qualified Data.ByteString as B -import System.IO -import Text.Printf +import Prelude hiding (take) +import Control.Monad +import Text.Printf -------------------------------------------------------------------------------- import Data.Serialize.Get import Data.UUID -------------------------------------------------------------------------------- +import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -readerThread :: (Package -> IO ()) -> Handle -> IO () -readerThread push_p h = forever $ do - header_bs <- B.hGet h 4 +readerThread :: (Package -> IO ()) -> Connection -> IO () +readerThread push_p c = forever $ do + header_bs <- connRecv c 4 case runGet getLengthPrefix header_bs of Left _ -> error "Wrong package framing" Right length_prefix - -> B.hGet h length_prefix >>= parsePackage + -> connRecv c length_prefix >>= parsePackage where parsePackage bs = case runGet getPackage bs of 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 @@ -72,12 +72,14 @@ data Event = Event { eventType :: !Text + , eventId :: !(Maybe UUID) , eventData :: !EventData } -------------------------------------------------------------------------------- -createEvent :: Text -- ^ Event type - -> EventData -- ^ Event data +createEvent :: Text -- ^ Event type + -> Maybe UUID -- ^ Event ID, generated if 'Nothing' + -> EventData -- ^ Event data -> Event createEvent = Event @@ -193,13 +195,14 @@ -------------------------------------------------------------------------------- newEvent :: Text -- ^ Event type + -> Maybe UUID -- ^ Event ID -> Int32 -- ^ Data content type -> Int32 -- ^ Metadata content type -> ByteString -- ^ Event data -> Maybe ByteString -- ^ Metadata -> IO NewEvent -newEvent evt_type data_type meta_type evt_data evt_meta = do - new_uuid <- randomIO +newEvent evt_type evt_id data_type meta_type evt_data evt_meta = do + new_uuid <- maybe randomIO return evt_id let uuid_bytes = toStrict $ toByteString new_uuid new_evt = NewEvent { newEventId = putField uuid_bytes diff --git a/Database/EventStore/Internal/Writer.hs b/Database/EventStore/Internal/Writer.hs --- a/Database/EventStore/Internal/Writer.hs +++ b/Database/EventStore/Internal/Writer.hs @@ -12,21 +12,20 @@ module Database.EventStore.Internal.Writer (writerThread) where -------------------------------------------------------------------------------- -import Control.Concurrent -import Control.Monad -import qualified Data.ByteString as B -import System.IO +import Control.Concurrent +import Control.Monad -------------------------------------------------------------------------------- import Data.Serialize.Put -------------------------------------------------------------------------------- +import Database.EventStore.Internal.Connection import Database.EventStore.Internal.Packages import Database.EventStore.Internal.Types -------------------------------------------------------------------------------- -writerThread :: Chan Package -> Handle -> IO () -writerThread chan hdl = forever $ do +writerThread :: Chan Package -> Connection -> IO () +writerThread chan c = forever $ do pkg <- readChan chan - B.hPut hdl (runPut $ putPackage pkg) - hFlush hdl + connSend c (runPut $ putPackage pkg) + connFlush c 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.7.0.1 +version: 0.7.1.0 -- A short (one-line) description of the package. synopsis: EventStore TCP Client @@ -58,6 +58,7 @@ -- Modules included in this library but not exported. other-modules: Database.EventStore.Catchup + Database.EventStore.Internal.Connection Database.EventStore.Internal.Packages Database.EventStore.Internal.Processor Database.EventStore.Internal.Reader