franz 0.5.1 → 0.5.2

raw patch · 6 files changed

+81/−55 lines, 6 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−0
• franz.cabal +2/−2
• src/Database/Franz/Client/Reconnect.hs +14/−7
• src/Database/Franz/Internal/Protocol.hs +3/−3
• src/Database/Franz/Internal/Reader.hs +23/−10
• tests/test.hs +34/−33

@@ -1,3 +1,8 @@+# v0.5.2++* Fixed the undesirable behaviour that causes an empty response when `BySeqNum (-1)` is specified+* Expanded `Reconnect` so that the query that caused a reconnection is visible+ # v0.5.1  * Fixed the bug adding all the following payloads in the result of `index`

@@ -1,12 +1,12 @@ cabal-version:  2.0-version:        0.5.1+version:        0.5.2 name:           franz description:    Please see the README on GitHub at <https://github.com/tsurucapital/franz#readme> homepage:       https://github.com/tsurucapital/franz#readme bug-reports:    https://github.com/tsurucapital/franz/issues author:         Fumiaki Kinoshita maintainer:     fumiexcel@gmail.com-copyright:      Copyright (c) 2021 Fumiaki Kinoshita+copyright:      Copyright (c) 2022 Fumiaki Kinoshita license:        BSD3 license-file:   LICENSE build-type:     Simple

@@ -1,5 +1,6 @@ {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE LambdaCase #-} module Database.Franz.Client.Reconnect@@ -35,20 +36,23 @@   -> (STM Response -> IO r)   -> IO r fetchWithPool pool q cont = withReconnection pool $ \conn -> fetch conn q cont+  `catch` \case+    r@(ReconnectInQuery _ _) -> throwM r -- Avoid deeply nested ReconnectInQuery+    r -> throwM $ ReconnectInQuery q r  -- | Run an action which takes a 'Connection', reconnecting whenever it throws an exception. withReconnection :: Pool -> (Connection -> IO a) -> IO a withReconnection Pool{..} cont = recovering   poolRetryPolicy-  [const $ Handler $ \Reconnect -> pure True]+  [const $ Handler $ \(_ :: Reconnect) -> pure True]   body   where      handler ex       | Just (ClientError err) <- fromException ex = Just err       | Just e <- fromException ex = Just (show (e :: IOException))-      | Just Reconnect <- fromException ex = Just-          $ "Connection to " <> fromFranzPath poolPath <> " timed out"+      | Just (e :: Reconnect) <- fromException ex = Just+          $ "Reconnecting to " <> fromFranzPath poolPath <> " due to " <> show e       | otherwise = Nothing      body _ = do@@ -59,7 +63,7 @@                 , fromFranzPath poolPath                 ]             conn <- tryJust handler (connect poolPath)-                >>= either (\e -> poolLogFunc e >> throwM Reconnect) pure+                >>= either (\e -> poolLogFunc e >> throwM ReconnectByError) pure             poolLogFunc $ "Connection #" <> show i <> " established"             pure ((i, Just conn), (i, conn))         v@(i, Just c) -> pure (v, (i, c))@@ -73,9 +77,12 @@                 -- another thread already established a new connection                 (j, Just _) | i == j -> (i + 1, Nothing) <$ disconnect conn                 x -> pure x-            throwM Reconnect+            throwM ReconnectByError -data Reconnect = Reconnect deriving (Show, Eq)+data Reconnect = ReconnectByTimeout+  | ReconnectByError+  | ReconnectInQuery !Query !Reconnect+  deriving (Show, Eq) instance Exception Reconnect  withPool :: RetryPolicyM IO@@ -93,4 +100,4 @@ atomicallyReconnecting :: Int -- ^ timeout in microseconds     -> STM a -> IO a atomicallyReconnecting timeout m = atomicallyWithin timeout m-  >>= maybe (throwM Reconnect) pure+  >>= maybe (throwM ReconnectByTimeout) pure

@@ -59,12 +59,12 @@ data RequestType = AllItems   | LastItem   | FirstItem-  deriving (Show, Generic)+  deriving (Show, Eq, Generic) instance Serialize RequestType  data ItemRef = BySeqNum !Int -- ^ sequential number   | ByIndex !IndexName !Int -- ^ index name and value-  deriving (Show, Generic)+  deriving (Show, Eq, Generic) instance Serialize ItemRef  data Query = Query@@ -72,7 +72,7 @@   , reqFrom :: !ItemRef -- ^ name of the index to search   , reqTo :: !ItemRef -- ^ name of the index to search   , reqType :: !RequestType-  } deriving (Show, Generic)+  } deriving (Show, Eq, Generic) instance Serialize Query  data RawRequest

@@ -136,23 +136,26 @@ type QueryResult = ((Int, Int) -- SeqNo *before* the first result, byte offset     , (Int, Int)) -- SeqNo of the last result, byte offset +isEmptyResult :: QueryResult -> Bool+isEmptyResult ((i, _), (j, _)) = i >= j+ range :: Int -- ^ from   -> Int -- ^ to   -> RequestType   -> IM.IntMap Int -- ^ offsets-  -> Maybe QueryResult-range begin end rt allOffsets = case rt of-    AllItems -> (firstItem, maybe firstItem fst $ IM.maxViewWithKey body) <$ guard ready+  -> (Bool, QueryResult)+range begin end rt allOffsets = (,) ready $ case rt of+    AllItems -> (firstItem, maybe firstItem fst $ IM.maxViewWithKey body)     LastItem -> case IM.maxViewWithKey body of-      Nothing -> (zero, zero) <$ guard ready+      Nothing -> (zero, zero)       Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of-        Just (ofs, _) -> (ofs, ofs') <$ guard ready-        Nothing -> (zero, ofs') <$ guard ready+        Just (ofs, _) -> (ofs, ofs')+        Nothing -> (zero, ofs')     FirstItem -> case IM.minViewWithKey body of-      Nothing -> (zero, zero) <$ guard ready+      Nothing -> (zero, zero)       Just (ofs', _) -> case IM.maxViewWithKey left of-        Just (ofs, _) -> (ofs, ofs') <$ guard ready-        Nothing -> (zero, ofs') <$ guard ready+        Just (ofs, _) -> (ofs, ofs')+        Nothing -> (zero, ofs')   where     ------------------------------------     --          begin     end@@ -325,7 +328,17 @@           let (body, lastItem, _) = IM.splitLookup val m           let body' = maybe id (IM.insert val) lastItem body           return $! maybe minBound fst $ IM.maxView body'-    return $! range begin end rt allOffsets+    return $! case range begin end rt allOffsets of+      (ready, result)+        -- When BySeqNum (-1) is specified, the client expects at least one item.+        -- More concretely, (ByIndex "time" t, BySeqNum (-1)) should resolve ByIndex+        -- until there's an item with a timestamp later than t.+        -- Perhaps there should be a special constructor for ItemRef that points to+        -- the latest item at transaction, and redefine "Nth-from-end" semantics+        -- where the "end" is frozen at the reception of the query.+        | BySeqNum i <- end_, i < 0, isEmptyResult result -> Nothing+        | ready -> Just result+        | otherwise -> Nothing   where     acquire = join $ atomically $ do       allStreams <- readTVar vStreams

@@ -12,42 +12,43 @@  main :: IO () main = withSystemTempDirectory "franz-test" $ \path -> do-    withWriter (Identity "") path-        $ \h -> forM_ ([0..9] :: [Int])-        $ \i -> write h (Identity (fromIntegral i * 10)) (BB.intDec i)+  withWriter (Identity "") path+    $ \h -> forM_ ([0..9] :: [Int])+    $ \i -> write h (Identity (fromIntegral i * 10)) (BB.intDec i) -    numFails <- newIORef (0 :: Int)+  numFails <- newIORef (0 :: Int) -    withConnection (LocalFranzPath path) $ \conn -> do-        let test :: RequestType -> ItemRef -> ItemRef -> [Item] -> IO ()-            test ty i j expected = do-                putStr $ unwords ["test", show ty, showsPrec 11 i "", showsPrec 11 j "", show expected, ": "]-                fetchSimple conn 500000 (Query "" i j ty) >>= \case-                    Nothing -> error "timeout"-                    Just xs-                        | C.toList xs == expected -> putStrLn "ok"-                        | otherwise -> do-                            modifyIORef' numFails (+1)-                            putStrLn $ "got " <> show (C.toList xs)+  withConnection (LocalFranzPath path) $ \conn -> do+    let test :: RequestType -> ItemRef -> ItemRef -> Maybe [Item] -> IO ()+        test ty i j expected = do+          putStr $ unwords ["test", show ty, showsPrec 11 i "", showsPrec 11 j "", show expected, ": "]+          result <- fmap C.toList <$> fetchSimple conn 100000 (Query "" i j ty)+          if result == expected+            then putStrLn "ok"+            else do+              modifyIORef' numFails (+1)+              putStrLn $ "got " <> show result -        test AllItems (BySeqNum 1) (BySeqNum 3) [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"},Item {seqNo = 3, indices = [30], payload = "3"}]-        test AllItems (ByIndex "" 15) (ByIndex "" 15) []-        test AllItems (ByIndex "" 15) (ByIndex "" 25) [Item {seqNo = 2, indices = [20], payload = "2"}]-        test AllItems (ByIndex "" 10) (ByIndex "" 20) [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"}]-        test AllItems (ByIndex "" 15) (ByIndex "" 20) [Item {seqNo = 2, indices = [20], payload = "2"}]+    test AllItems (BySeqNum 1) (BySeqNum 3) $ Just [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"},Item {seqNo = 3, indices = [30], payload = "3"}]+    test AllItems (ByIndex "" 15) (ByIndex "" 15) $ Just []+    test AllItems (ByIndex "" 15) (ByIndex "" 25) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}]+    test AllItems (ByIndex "" 10) (ByIndex "" 20) $ Just [Item {seqNo = 1, indices = [10], payload = "1"},Item {seqNo = 2, indices = [20], payload = "2"}]+    test AllItems (ByIndex "" 15) (ByIndex "" 20) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}] -        test FirstItem (BySeqNum 1) (BySeqNum 5) [Item {seqNo = 1, indices = [10], payload = "1"}]-        test FirstItem (ByIndex "" 10) (ByIndex "" 45) [Item {seqNo = 1, indices = [10], payload = "1"}]-        test FirstItem (ByIndex "" 15) (ByIndex "" 45) [Item {seqNo = 2, indices = [20], payload = "2"}]-        test FirstItem (ByIndex "" 10) (ByIndex "" 10) [Item {seqNo = 1, indices = [10], payload = "1"}]-        test FirstItem (ByIndex "" 15) (ByIndex "" 15) []+    test FirstItem (BySeqNum 1) (BySeqNum 5) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}]+    test FirstItem (ByIndex "" 10) (ByIndex "" 45) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}]+    test FirstItem (ByIndex "" 15) (ByIndex "" 45) $ Just [Item {seqNo = 2, indices = [20], payload = "2"}]+    test FirstItem (ByIndex "" 10) (ByIndex "" 10) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}]+    test FirstItem (ByIndex "" 15) (ByIndex "" 15) $ Just [] -        test LastItem (BySeqNum 1) (BySeqNum 5) [Item {seqNo = 5, indices = [50], payload = "5"}]-        test LastItem (ByIndex "" 10) (ByIndex "" 45) [Item {seqNo = 4, indices = [40], payload = "4"}]-        test LastItem (ByIndex "" 15) (ByIndex "" 50) [Item {seqNo = 5, indices = [50], payload = "5"}]-        test LastItem (ByIndex "" 10) (ByIndex "" 10) [Item {seqNo = 1, indices = [10], payload = "1"}]-        test LastItem (ByIndex "" 15) (ByIndex "" 15) []+    test LastItem (BySeqNum 1) (BySeqNum 5) $ Just [Item {seqNo = 5, indices = [50], payload = "5"}]+    test LastItem (ByIndex "" 10) (ByIndex "" 45) $ Just [Item {seqNo = 4, indices = [40], payload = "4"}]+    test LastItem (ByIndex "" 15) (ByIndex "" 50) $ Just [Item {seqNo = 5, indices = [50], payload = "5"}]+    test LastItem (ByIndex "" 10) (ByIndex "" 10) $ Just [Item {seqNo = 1, indices = [10], payload = "1"}]+    test LastItem (ByIndex "" 15) (ByIndex "" 15) $ Just [] -    readIORef numFails >>= \case-        0 -> putStrLn "All good"-        _ -> fail "Failed"+    test LastItem (BySeqNum 10) (BySeqNum (-1)) Nothing++  readIORef numFails >>= \case+      0 -> putStrLn "All good"+      _ -> fail "Failed"