eventstore 0.7.0.1 → 0.7.1.0

raw patch · 10 files changed

+151/−90 lines, 10 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +5/−0
• Database/EventStore.hs +1/−0
• Database/EventStore/Internal/Connection.hs +101/−0
• Database/EventStore/Internal/Operation/TransactionStartOperation.hs +3/−4
• Database/EventStore/Internal/Operation/WriteEventsOperation.hs +3/−4
• Database/EventStore/Internal/Processor.hs +14/−60
• Database/EventStore/Internal/Reader.hs +8/−9
• Database/EventStore/Internal/Types.hs +7/−4
• Database/EventStore/Internal/Writer.hs +7/−8
• eventstore.cabal +2/−1

@@ -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.

@@ -20,6 +20,7 @@     , withJsonAndMetadata       -- * Connection     , Connection+    , ConnectionException(..)     , Credentials     , Settings(..)     , Retry

@@ -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 <yo.eight@gmail.com>+-- 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+    }

@@ -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

@@ -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

@@ -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 ()

@@ -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

@@ -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

@@ -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

@@ -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