franz 0.5 → 0.5.1

raw patch · 8 files changed

+107/−23 lines, 8 filesdep ~basedep ~fast-builderPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependency ranges changed: base, fast-builder

API changes (from Hackage documentation)

Files

• ChangeLog.md +5/−0
• franz.cabal +12/−1
• src/Database/Franz/Client.hs +3/−6
• src/Database/Franz/Contents.hs +5/−1
• src/Database/Franz/Internal/Protocol.hs +4/−1
• src/Database/Franz/Internal/Reader.hs +20/−8
• src/Database/Franz/Server.hs +5/−6
• tests/test.hs +53/−0

@@ -1,3 +1,8 @@+# v0.5.1++* Fixed the bug adding all the following payloads in the result of `index`+* Added `FirstItem` to `RequestType`+ # v0.5  * Renamed internal modules under `Internal` prefix

@@ -1,5 +1,5 @@ cabal-version:  2.0-version:        0.5+version:        0.5.1 name:           franz description:    Please see the README on GitHub at <https://github.com/tsurucapital/franz#readme> homepage:       https://github.com/tsurucapital/franz#readme@@ -92,3 +92,14 @@     , network     , optparse-applicative   default-language: Haskell2010++test-suite test+    build-depends: base+        , franz+        , fast-builder+        , temporary+    type: exitcode-stdio-1.0+    hs-source-dirs: tests+    default-language: Haskell2010+    main-is: test.hs+    ghc-options: -Wall -Wcompat

@@ -247,17 +247,14 @@   = handleQuery (FranzPrefix "") connReader (FranzDirectory connDir) query   (cont . throwSTM)   $ \stream transaction -> atomically transaction >>= \case-    (False, _) -> do+    Nothing -> do       vResp <- newEmptyTMVarIO       tid <- flip forkFinally (atomically . putTMVar vResp) $ do-        result <- atomically $ do-          (ready, result) <- transaction-          guard ready-          pure result+        result <- atomically $ transaction >>= maybe retry pure         readContents stream result       cont (pure $ Right $ takeTMVar vResp >>= either throwSTM pure)         `finally` throwTo tid requestFinished-    (True, result) -> do+    Just result -> do       vResp <- newEmptyTMVarIO       tid <- forkFinally (readContents stream result) (atomically . putTMVar vResp)       cont (takeTMVar vResp >>= either throwSTM (pure . Left))

@@ -6,6 +6,7 @@   , Database.Franz.Internal.Contents.indexNames   , Item(..)   , toList+  , toVector   , last   , length   , index@@ -29,6 +30,9 @@ toList :: Contents -> [Item] toList contents = [unsafeIndex contents i | i <- [0..length contents - 1]] +toVector :: Contents -> V.Vector Item+toVector contents = V.generate (length contents) (unsafeIndex contents)+ last :: Contents -> Maybe Item last contents   | i >= 0 = Just $ unsafeIndex contents i@@ -47,7 +51,7 @@     ofs0 = maybe payloadOffset fst $ indicess V.!? (i - 1)     (ofs1, indices) = indicess V.! i     seqNo = seqnoOffset + i + 1-    payload = B.take (ofs1 - payloadOffset) $ B.drop (ofs0 - payloadOffset) payloads+    payload = B.take (ofs1 - ofs0) $ B.drop (ofs0 - payloadOffset) payloads  lookupIndex :: Contents -> IndexName -> Maybe (Item -> Int64) lookupIndex Contents{indexNames} name

@@ -56,7 +56,10 @@ instance Serialize FranzException instance Exception FranzException -data RequestType = AllItems | LastItem deriving (Show, Generic)+data RequestType = AllItems+  | LastItem+  | FirstItem+  deriving (Show, Generic) instance Serialize RequestType  data ItemRef = BySeqNum !Int -- ^ sequential number

@@ -133,22 +133,34 @@   where     logFollower = hPutStrLn stderr . unwords . (:) "[follower]" -type QueryResult = ((Int, Int) -- starting SeqNo, byte offset-    , (Int, Int)) -- ending SeqNo, byte offset+type QueryResult = ((Int, Int) -- SeqNo *before* the first result, byte offset+    , (Int, Int)) -- SeqNo of the last result, byte offset  range :: Int -- ^ from   -> Int -- ^ to   -> RequestType   -> IM.IntMap Int -- ^ offsets-  -> (Bool, QueryResult)+  -> Maybe QueryResult range begin end rt allOffsets = case rt of-    AllItems -> (ready, (firstItem, maybe firstItem fst $ IM.maxViewWithKey body))+    AllItems -> (firstItem, maybe firstItem fst $ IM.maxViewWithKey body) <$ guard ready     LastItem -> case IM.maxViewWithKey body of-      Nothing -> (False, (zero, zero))+      Nothing -> (zero, zero) <$ guard ready       Just (ofs', r) -> case IM.maxViewWithKey (IM.union left r) of-        Just (ofs, _) -> (ready, (ofs, ofs'))-        Nothing -> (ready, (zero, ofs'))+        Just (ofs, _) -> (ofs, ofs') <$ guard ready+        Nothing -> (zero, ofs') <$ guard ready+    FirstItem -> case IM.minViewWithKey body of+      Nothing -> (zero, zero) <$ guard ready+      Just (ofs', _) -> case IM.maxViewWithKey left of+        Just (ofs, _) -> (ofs, ofs') <$ guard ready+        Nothing -> (zero, ofs') <$ guard ready   where+    ------------------------------------+    --          begin     end+    --            v        v+    -- |    wing     | lastItem | cont |+    -- | left |      body       |      |+    ------------------------------------+     zero = (-1, 0)     ready = isJust lastItem || not (null cont)     (wing, lastItem, cont) = IM.splitLookup end allOffsets@@ -280,7 +292,7 @@   -> FranzDirectory   -> Query   -> (FranzException -> IO r)-  -> (Stream -> STM (Bool, QueryResult) -> IO r) -> IO r+  -> (Stream -> STM (Maybe QueryResult) -> IO r) -> IO r handleQuery prefix FranzReader{..} dir (Query name begin_ end_ rt) onError cont   = bracket (try acquire) (either mempty removeActivity) $ \case   Left err -> onError err

@@ -55,18 +55,17 @@   handleQuery prefix franzReader path req throwIO $ \stream query ->     atomically (trySTM query) >>= \case       Left e -> sendHeader env $ ResponseError reqId e-      Right (ready, offsets)-        | ready -> sendContents env (Response reqId) stream offsets-        | otherwise -> do+      Right (Just offsets) -> sendContents env (Response reqId) stream offsets+      Right Nothing -> do           tid <- flip forkFinally pop $ bracket_             (atomically $ addActivity stream)             (removeActivity stream) $ do               sendHeader env $ ResponseWait reqId               join $ atomically $ trySTM query >>= \case                 Left e -> pure $ sendHeader env $ ResponseError reqId e-                Right (ready', offsets') -> do-                  check ready'-                  pure $ sendContents env (Response reqId) stream offsets'+                Right Nothing -> retry+                Right (Just offsets) ->  +                  pure $ sendContents env (Response reqId) stream offsets           -- Store the thread ID of the thread yielding a future           -- response such that we can kill it mid-way if user           -- sends a cancel request or we're killed with an

@@ -0,0 +1,53 @@+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedLists #-}+{-# LANGUAGE OverloadedStrings #-}+import Control.Monad+import Database.Franz.Writer+import Database.Franz.Client+import Database.Franz.Contents as C+import qualified Data.ByteString.FastBuilder as BB+import Data.Functor.Identity+import System.IO.Temp+import Data.IORef++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)++    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)++        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 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 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) []++    readIORef numFails >>= \case+        0 -> putStrLn "All good"+        _ -> fail "Failed"