eventstore 0.13.1.6 → 0.13.1.7
raw patch · 6 files changed
+97/−50 lines, 6 filesdep −asyncdep ~classy-preludedep ~uuidPVP ok
version bump matches the API change (PVP)
Dependencies removed: async
Dependency ranges changed: classy-prelude, uuid
API changes (from Hackage documentation)
Files
- CHANGELOG.markdown +5/−0
- Database/EventStore.hs +10/−3
- Database/EventStore/Internal/Operation/Catchup.hs +15/−6
- eventstore.cabal +5/−2
- tests/Tests.hs +60/−37
- tests/integration.hs +2/−2
CHANGELOG.markdown view
@@ -1,3 +1,8 @@+0.13.1.7+--------+* Report stream name when facing a 'Stream not found' error from the server on catchup subscription.+* Prevent the driver from deadlocking when asking for catchup subscription on a stream that doesn't exist.+ 0.13.1.6 -------- * Fix a deadlock on user side when asking a catchup subscription on a stream that doesn't exist.
Database/EventStore.hs view
@@ -883,7 +883,14 @@ where go (S.SubConfirmed run) = atomically $ mk run go (S.EventAppeared e) = atomically $ do- SubState sm close <- rcv- let nxt = S.eventArrived e sm- send $ SubState nxt close+ st <- rcv+ case st of+ SubState sm close ->+ let nxt = S.eventArrived e sm in+ send $ SubState nxt close+ SubException _ ->+ -- At this moment [07 October 2016], this can only happen during+ -- the first phase of a catchup subscription where the user+ -- asked for a subscription on a stream that doesn't exist.+ return () go (S.Dropped r) = atomically $ quit r
Database/EventStore/Internal/Operation/Catchup.hs view
@@ -38,8 +38,9 @@ defaultBatchSize = 500 ---------------------------------------------------------------------------------streamNotFound :: OperationError-streamNotFound = InvalidOperation "Catchup. inexistant stream"+streamNotFound :: Text -> OperationError+streamNotFound stream =+ InvalidOperation $ "Catchup. inexistant stream [" <> stream <> "]" -------------------------------------------------------------------------------- -- | Catchup operation state.@@ -51,6 +52,11 @@ -- the $all stream. --------------------------------------------------------------------------------+streamName :: CatchupState -> Text+streamName (RegularCatchup stream _) = stream+streamName _ = "$all"++-------------------------------------------------------------------------------- -- | Stream catching up operation. catchup :: Settings -> CatchupState@@ -79,7 +85,7 @@ tmp_tpe = AllCatchup nxt_c nxt_p chk = CheckpointPosition $ sliceNext as return (sliceEOS as, sliceEvents as, chk, tmp_tpe)- Left rr -> fromReadResult rr $ \as ->+ Left rr -> fromReadResult (streamName tpe) rr $ \as -> let RegularCatchup s _ = tpe nxt = sliceNext as tmp_tpe = RegularCatchup s nxt@@ -90,10 +96,13 @@ when (not eos) $ go nxt_tpe ---------------------------------------------------------------------------------fromReadResult :: ReadResult 'RegularStream a -> (a -> SM b x) -> SM b x-fromReadResult res k =+fromReadResult :: Text+ -> ReadResult 'RegularStream a+ -> (a -> SM b x)+ -> SM b x+fromReadResult stream res k = case res of- ReadNoStream -> failure streamNotFound+ ReadNoStream -> failure $ streamNotFound stream ReadStreamDeleted s -> failure $ StreamDeleted s ReadNotModified -> failure $ ServerError Nothing ReadError e -> failure $ ServerError e
eventstore.cabal view
@@ -10,7 +10,7 @@ -- PVP summary: +-+------- breaking API changes -- | | +----- non-breaking API additions -- | | | +--- code changes with no API change-version: 0.13.1.6+version: 0.13.1.7 tested-with: GHC >= 7.8.3 && < 7.11 @@ -133,6 +133,8 @@ hs-source-dirs: tests main-is: integration.hs + default-extensions: NoImplicitPrelude+ other-modules: Tests ghc-options: -Wall@@ -147,4 +149,5 @@ time, dotnet-timespan, connection,- async+ classy-prelude,+ uuid
tests/Tests.hs view
@@ -16,14 +16,13 @@ module Tests where ---------------------------------------------------------------------------------import Control.Concurrent-import Control.Exception-import Data.Maybe (catMaybes)+import ClassyPrelude ---------------------------------------------------------------------------------import Control.Concurrent.Async import Data.Aeson import Data.DotNet.TimeSpan+import Data.UUID hiding (null)+import Data.UUID.V4 import Test.Tasty import Test.Tasty.HUnit @@ -55,24 +54,32 @@ , testCase "Shutdown connection" $ shutdownTest conn ] + --------------------------------------------------------------------------------+freshStreamId :: IO Text+freshStreamId = fmap toText nextRandom++-------------------------------------------------------------------------------- writeEventTest :: Connection -> IO () writeEventTest conn = do let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js - as <- sendEvent conn "write-event-test" anyVersion evt+ stream <- freshStreamId+ as <- sendEvent conn stream anyVersion evt _ <- waitAsync as return () -------------------------------------------------------------------------------- readEventTest :: Connection -> IO () readEventTest conn = do+ stream <- freshStreamId+ let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js- as <- sendEvent conn "read-event-test" anyVersion evt+ as <- sendEvent conn stream anyVersion evt _ <- waitAsync as- bs <- readEvent conn "read-event-test" 0 False+ bs <- readEvent conn stream 0 False rs <- waitAsync bs case rs of ReadSuccess re ->@@ -88,26 +95,28 @@ -------------------------------------------------------------------------------- deleteStreamTest :: Connection -> IO () deleteStreamTest conn = do+ stream <- freshStreamId let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js- _ <- sendEvent conn "delete-stream-test" anyVersion evt >>= waitAsync- _ <- deleteStream conn "delete-stream-test" anyVersion Nothing+ _ <- sendEvent conn stream anyVersion evt >>= waitAsync+ _ <- deleteStream conn stream anyVersion Nothing return () -------------------------------------------------------------------------------- transactionTest :: Connection -> IO () transactionTest conn = do+ stream <- freshStreamId let js = object [ "baz" .= True ] evt = createEvent "foo" Nothing $ withJson js- t <- startTransaction conn "transaction-test" anyVersion >>= waitAsync+ t <- startTransaction conn stream anyVersion >>= waitAsync _ <- transactionWrite t [evt] >>= waitAsync- rs <- readEvent conn "transaction-test" 0 False >>= waitAsync+ rs <- readEvent conn stream 0 False >>= waitAsync case rs of ReadNoStream -> return () e -> fail $ "transaction-test stream is supposed to not exist " ++ show e _ <- transactionCommit t >>= waitAsync- rs2 <- readEvent conn "transaction-test" 0 False >>= waitAsync+ rs2 <- readEvent conn stream 0 False >>= waitAsync case rs2 of ReadSuccess re -> case re of@@ -122,13 +131,14 @@ -------------------------------------------------------------------------------- readStreamEventForwardTest :: Connection -> IO () readStreamEventForwardTest conn = do+ stream <- freshStreamId let jss = [ object [ "baz" .= True] , object [ "foo" .= False] , object [ "bar" .= True] ] evts = fmap (createEvent "foo" Nothing . withJson) jss- _ <- sendEvents conn "read-forward-test" anyVersion evts >>= waitAsync- rs <- readStreamEventsForward conn "read-forward-test" 0 10 False >>= waitAsync+ _ <- sendEvents conn stream anyVersion evts >>= waitAsync+ rs <- readStreamEventsForward conn stream 0 10 False >>= waitAsync case rs of ReadSuccess sl -> do let jss_evts = catMaybes $ fmap resolvedEventDataAsJson@@ -168,14 +178,16 @@ -------------------------------------------------------------------------------- subscribeTest :: Connection -> IO () subscribeTest conn = do+ stream <- freshStreamId+ let jss = [ object [ "baz" .= True] , object [ "foo" .= False] , object [ "bar" .= True] ] evts = fmap (createEvent "foo" Nothing . withJson) jss- sub <- subscribe conn "subscribe-test" False+ sub <- subscribe conn stream False _ <- waitConfirmation sub- _ <- sendEvents conn "subscribe-test" anyVersion evts >>= waitAsync+ _ <- sendEvents conn stream anyVersion evts >>= waitAsync let loop 3 = return [] loop i = do e <- nextEvent sub@@ -193,6 +205,8 @@ -------------------------------------------------------------------------------- subscribeFromTest :: Connection -> IO () subscribeFromTest conn = do+ stream <- freshStreamId+ let jss = [ object [ "1" .= (1 :: Int)] , object [ "2" .= (2 :: Int)] , object [ "3" .= (3 :: Int)]@@ -204,10 +218,10 @@ alljss = jss ++ jss2 evts = fmap (createEvent "foo" Nothing . withJson) jss evts2 = fmap (createEvent "foo" Nothing . withJson) jss2- _ <- sendEvents conn "subscribe-from-test" anyVersion evts >>= waitAsync- sub <- subscribeFrom conn "subscribe-from-test" False Nothing (Just 1)+ _ <- sendEvents conn stream anyVersion evts >>= waitAsync+ sub <- subscribeFrom conn stream False Nothing (Just 1) _ <- waitConfirmation sub- _ <- sendEvents conn "subscribe-from-test" anyVersion evts2 >>= waitAsync+ _ <- sendEvents conn stream anyVersion evts2 >>= waitAsync let loop [] = do m <- nextEventMaybe sub@@ -243,7 +257,8 @@ -------------------------------------------------------------------------------- subscribeFromNoStreamTest :: Connection -> IO () subscribeFromNoStreamTest conn = do- sub <- subscribeFrom conn "non-existent-stream" False Nothing Nothing+ stream <- freshStreamId+ sub <- subscribeFrom conn stream False Nothing Nothing let subAction = do res <- try $ waitTillCatchup sub case res of@@ -261,28 +276,31 @@ -------------------------------------------------------------------------------- setStreamMetadataTest :: Connection -> IO () setStreamMetadataTest conn = do+ stream <- freshStreamId let metadata = buildStreamMetadata $ setCustomProperty "foo" (1 :: Int)- _ <- setStreamMetadata conn "set-metadata-test" anyVersion metadata >>= waitAsync+ _ <- setStreamMetadata conn stream anyVersion metadata >>= waitAsync return () -------------------------------------------------------------------------------- getStreamMetadataTest :: Connection -> IO () getStreamMetadataTest conn = do+ stream <- freshStreamId let metadata = buildStreamMetadata $ setCustomProperty "foo" (1 :: Int)- _ <- setStreamMetadata conn "get-metadata-test" anyVersion metadata >>= waitAsync- r <- getStreamMetadata conn "get-metadata-test" >>= waitAsync+ _ <- setStreamMetadata conn stream anyVersion metadata >>= waitAsync+ r <- getStreamMetadata conn stream >>= waitAsync case r of StreamMetadataResult _ _ m -> case getCustomProperty m "foo" of Just i -> assertEqual "Should have equal value" (1 :: Int) i _ -> fail "Can't find foo property"- _ -> fail "Stream get-metadata-test doesn't exist"+ _ -> fail $ "Stream " <> unpack stream <>" doesn't exist" -------------------------------------------------------------------------------- createPersistentTest :: Connection -> IO () createPersistentTest conn = do let def = defaultPersistentSubscriptionSettings- r <- createPersistentSubscription conn "group" "create-sub" def >>= waitAsync+ stream <- freshStreamId+ r <- createPersistentSubscription conn "group" stream def >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -291,8 +309,9 @@ updatePersistentTest :: Connection -> IO () updatePersistentTest conn = do let def = defaultPersistentSubscriptionSettings- _ <- createPersistentSubscription conn "group" "update-sub" def >>= waitAsync- r <- updatePersistentSubscription conn "group" "update-sub" def >>= waitAsync+ stream <- freshStreamId+ _ <- createPersistentSubscription conn "group" stream def >>= waitAsync+ r <- updatePersistentSubscription conn "group" stream def >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -301,8 +320,9 @@ deletePersistentTest :: Connection -> IO () deletePersistentTest conn = do let def = defaultPersistentSubscriptionSettings- _ <- createPersistentSubscription conn "group" "delete-sub" def >>= waitAsync- r <- deletePersistentSubscription conn "group" "delete-sub" >>= waitAsync+ stream <- freshStreamId+ _ <- createPersistentSubscription conn "group" stream def >>= waitAsync+ r <- deletePersistentSubscription conn "group" stream >>= waitAsync case r of Nothing -> return () Just e -> fail $ "Exception arised: " ++ show e@@ -317,9 +337,10 @@ , js2 ] evts = fmap (createEvent "foo" Nothing . withJson) jss- _ <- createPersistentSubscription conn "group" "connect-sub" def >>= waitAsync- _ <- sendEvents conn "connect-sub" anyVersion evts >>= waitAsync- sub <- connectToPersistentSubscription conn "group" "connect-sub" 1+ stream <- freshStreamId+ _ <- createPersistentSubscription conn "group" stream def >>= waitAsync+ _ <- sendEvents conn stream anyVersion evts >>= waitAsync+ sub <- connectToPersistentSubscription conn "group" stream 1 _ <- waitConfirmation sub r <- nextEvent sub case resolvedEventDataAsJson r of@@ -349,22 +370,24 @@ metadata = buildStreamMetadata $ setMaxAge timespan evt = createEvent "foo" Nothing $ withJson (object ["type" .= (3 :: Int)])- _ <- sendEvent conn "test-max-age" anyVersion evt >>= waitAsync- _ <- setStreamMetadata conn "test-max-age" anyVersion metadata >>= waitAsync- r <- getStreamMetadata conn "test-max-age" >>= waitAsync+ stream <- freshStreamId+ _ <- sendEvent conn stream anyVersion evt >>= waitAsync+ _ <- setStreamMetadata conn stream anyVersion metadata >>= waitAsync+ r <- getStreamMetadata conn stream >>= waitAsync case r of StreamMetadataResult _ _ m -> assertEqual "Should have equal timespan" (Just timespan) (streamMetadataMaxAge m)- _ -> fail "Stream test-max-age doesn't exist"+ _ -> fail $ "Stream " <> unpack stream <> " doesn't exist" -------------------------------------------------------------------------------- shutdownTest :: Connection -> IO () shutdownTest conn = do+ stream <- freshStreamId let js = object ["baz" .= True] evt = createEvent "foo" Nothing $ withJson js action = do- _ <- sendEvent conn "shutdown-test" anyVersion evt+ _ <- sendEvent conn stream anyVersion evt return False shutdown conn waitTillClosed conn
tests/integration.hs view
@@ -14,7 +14,7 @@ module Main where ---------------------------------------------------------------------------------import Data.Time+import ClassyPrelude import Database.EventStore import Database.EventStore.Logging import Test.Tasty@@ -40,7 +40,7 @@ _logger l = do t <- getCurrentTime putStr "["- putStr $ show t+ putStr $ pack $ show t putStr "] " showLog l where